#!/usr/bin/env python3 """Original LockedIn Labs structured-output exercise. Python 3.11+. Install fde-first-schema-requirements.txt, then: python fde-first-schema.py test python fde-first-schema.py compare python fde-first-schema.py run --case wrong-type python fde-first-schema.py check --file answer.json Synthetic source and responses; actual JSON parsing and JSON Schema validation. No model, provider account, network connection or external action. """ import argparse from contextlib import ExitStack, contextmanager from copy import deepcopy from importlib.metadata import version import json from pathlib import Path import unittest from unittest.mock import patch LIMIT = 4096 SOURCE = { "id": "NOTE-7", "revision": 3, "text": "Pump P7 is running. Temperature is 41 C. No shutdown is recorded.", "condition": "running", "support": "Pump P7 is running.", } SCHEMA = { "$schema": "https://json-schema.org/draft/2020-12/schema", "type": "object", "properties": { "source_id": {"type": "string", "minLength": 1, "maxLength": 40}, "source_revision": {"type": "integer", "minimum": 1}, "condition": {"type": "string", "enum": ["running", "stopped", "unknown"]}, "evidence": {"type": "string", "minLength": 1, "maxLength": 180}, }, "required": ["source_id", "source_revision", "condition", "evidence"], "additionalProperties": False, } BASE = {"source_id": "NOTE-7", "source_revision": 3, "condition": "running", "evidence": "Pump P7 is running."} @contextmanager def without_network(): with ExitStack() as stack: for name in ("socket.create_connection", "socket.getaddrinfo", "socket.socket.connect"): stack.enter_context(patch(name, side_effect=RuntimeError("network_disabled_in_lab"))) yield def reject_constant(_): raise ValueError("non_json_constant") def unique_object(pairs): result = {} for key, value in pairs: if key in result: raise ValueError("duplicate_key") result[key] = value return result def inspect_response(raw): """Return bounded diagnostics, never the submitted response or exception text.""" stages = [ {"id": "parse", "label": "Read JSON", "status": "not-run"}, {"id": "schema", "label": "Check the shape", "status": "not-run"}, {"id": "source", "label": "Compare the note", "status": "not-run"}, ] result = {"stages": stages, "first_stop": "parse", "issues": [], "reason": "invalid_json", "outcome": "hold"} stages[0]["status"] = "fail" if not isinstance(raw, str) or len(raw.encode("utf-8")) > LIMIT: result["reason"] = "input_too_large_or_not_text" return result try: value = json.loads(raw, parse_constant=reject_constant, object_pairs_hook=unique_object) except (json.JSONDecodeError, ValueError, RecursionError): return result stages[0]["status"] = "pass" from jsonschema import Draft202012Validator Draft202012Validator.check_schema(SCHEMA) errors = sorted(Draft202012Validator(SCHEMA).iter_errors(value), key=lambda e: ("/".join(map(str, e.absolute_path)), str(e.validator))) if errors: stages[1]["status"] = "fail" result.update(first_stop="schema", reason="schema_mismatch", issues=[{"path": "/" + "/".join(map(str, e.absolute_path)), "keyword": e.validator} for e in errors]) return result stages[1]["status"] = "pass" result["first_stop"] = "source" stages[2]["status"] = "fail" # Deliberately narrow fixture checks, not a general semantic truth evaluator. if value["source_id"] != SOURCE["id"]: result["reason"] = "source_not_found" elif value["source_revision"] != SOURCE["revision"]: result["reason"] = "source_revision_changed" elif value["condition"] != SOURCE["condition"]: result["reason"] = "condition_not_supported" elif value["evidence"] != SOURCE["support"]: result["reason"] = "quote_not_supported" else: stages[2]["status"] = "pass" result.update(first_stop="none", reason="supported_by_fixture", outcome="supported-draft") return result def cases(): def edited(**changes): return json.dumps({**BASE, **changes}, indent=2) missing = deepcopy(BASE) del missing["evidence"] return [ ("supported", "Matches the note", "All four fields refer to the supplied note.", edited()), ("syntax", "A trailing comma", "The text includes a comma after the last property.", edited()[:-2] + ",\n}"), ("wrong-type", "A number written as text", "The revision is quoted, so JSON reads it as a string.", edited(source_revision="3")), ("missing", "A field left out", "The response omits the evidence property.", json.dumps(missing, indent=2)), ("null", "A field set to null", "The evidence property is present, with a null value.", edited(evidence=None)), ("extra", "An unexpected instruction", "A response adds an execute property to the agreed fields.", edited(execute="shutdown_pump")), ("enum", "An unlisted condition", "The condition is a string, but not one of the permitted choices.", edited(condition="idle")), ("unsupported", "Correct shape, wrong claim", "The response says stopped while the note says running.", edited(condition="stopped")), ("stale", "An earlier revision", "The reference names revision 2; the supplied note is revision 3.", edited(source_revision=2)), ] def report(): return { "schema": "lockedin.first-schema/v1", "packages": {"jsonschema": version("jsonschema")}, "source": SOURCE, "contract": SCHEMA, "provenance": "Actual Python JSON parsing and jsonschema Draft 2020-12 validation over original synthetic responses. Source comparison is an explicit check against this one fixture, not model generation or general fact checking. No external action is performed.", "cases": [{"id": id_, "label": label, "description": description, "raw": raw, **inspect_response(raw)} for id_, label, description, raw in cases()], } class FirstSchemaTests(unittest.TestCase): def test_actual_boundaries_and_downstream_skips(self): expected = ["none", "parse", "schema", "schema", "schema", "schema", "schema", "source", "source"] rows = report()["cases"] self.assertEqual([r["first_stop"] for r in rows], expected) self.assertEqual([s["status"] for s in rows[1]["stages"]], ["fail", "not-run", "not-run"]) self.assertEqual([s["status"] for s in rows[2]["stages"]], ["pass", "fail", "not-run"]) def test_equal_shape_does_not_establish_equal_meaning(self): rows = {r["id"]: r for r in report()["cases"]} for id_ in ("supported", "unsupported", "stale"): self.assertEqual(rows[id_]["stages"][1]["status"], "pass") self.assertEqual(rows["supported"]["outcome"], "supported-draft") self.assertEqual(rows["unsupported"]["reason"], "condition_not_supported") self.assertEqual(rows["stale"]["reason"], "source_revision_changed") def test_missing_null_type_and_enum_have_distinct_reasons(self): rows = {r["id"]: r for r in report()["cases"]} for id_, keyword in [("missing", "required"), ("null", "type"), ("wrong-type", "type"), ("extra", "additionalProperties"), ("enum", "enum")]: self.assertEqual(rows[id_]["issues"][0]["keyword"], keyword) self.assertEqual(rows["missing"]["issues"][0]["path"], "/") self.assertEqual(rows["null"]["issues"][0]["path"], "/evidence") def test_booleans_are_not_integer_revisions_and_no_coercion_occurs(self): for value in [True, "3"]: row = inspect_response(json.dumps({**BASE, "source_revision": value})) self.assertEqual(row["first_stop"], "schema") self.assertIsInstance(BASE["source_revision"], int) def test_bad_json_duplicate_keys_and_nonstandard_constants_stop(self): for raw in ['{"x": 1,}', '{"condition":"running","condition":"stopped"}', '{"x": NaN}', '{"x": Infinity}']: self.assertEqual(inspect_response(raw)["first_stop"], "parse") # A platform may reject nesting while parsing; otherwise the object schema refuses it. self.assertIn(inspect_response("[" * 1500 + "]" * 1500)["first_stop"], ["parse", "schema"]) self.assertEqual(inspect_response(" " * (LIMIT + 1))["reason"], "input_too_large_or_not_text") def test_schema_shape_changes_do_not_make_a_claim_supported(self): altered = deepcopy(SCHEMA) altered["additionalProperties"] = True with patch.dict(SCHEMA, altered, clear=True): result = inspect_response(json.dumps({**BASE, "condition": "stopped", "execute": "shutdown_pump"})) self.assertEqual(result["first_stop"], "source") self.assertFalse(SCHEMA["additionalProperties"]) def test_diagnostics_do_not_echo_submitted_text(self): marker = "SYNTHETIC_PRIVATE_DETAIL_DO_NOT_EXPORT" for raw in [marker, json.dumps({**BASE, "evidence": marker}), json.dumps({**BASE, "condition": marker})]: self.assertNotIn(marker, json.dumps(inspect_response(raw))) def test_reporting_is_local_and_does_not_mutate_source_or_contract(self): before = json.dumps([SOURCE, SCHEMA, BASE], sort_keys=True) with without_network(): a, b = report(), report() self.assertEqual(a, b) self.assertEqual(json.dumps([SOURCE, SCHEMA, BASE], sort_keys=True), before) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("command", choices=["test", "compare", "run", "check"]) parser.add_argument("--case", choices=[row[0] for row in cases()], default="supported") parser.add_argument("--file", type=Path) args = parser.parse_args() try: with without_network(): if args.command == "test": suite = unittest.defaultTestLoader.loadTestsFromTestCase(FirstSchemaTests) return 0 if unittest.TextTestRunner(verbosity=2).run(suite).wasSuccessful() else 1 if args.command == "compare": result = report() elif args.command == "check": if args.file is None: parser.error("--file is required for check") with args.file.open("rb") as f: raw = f.read(LIMIT + 1) result = inspect_response(raw.decode("utf-8")) else: raw = next(row[3] for row in cases() if row[0] == args.case) result = inspect_response(raw) print(json.dumps(result, indent=2)) # The command ran; inspect outcome/first_stop for the response decision. return 0 except ImportError: print(json.dumps({"status": "setup_required", "reason": "install_fde_first_schema_requirements"})) return 1 except (OSError, UnicodeError, RuntimeError): print(json.dumps({"status": "incomplete", "reason": "check_local_file_and_environment"})) return 1 if __name__ == "__main__": raise SystemExit(main())