#!/usr/bin/env python3 """Real LangChain ToolStrategy validation with scripted output proposals. Python 3.11+; isolated venv: langchain==1.4.6 langgraph==1.2.16. Run: python fde-structured-output-lab.py test python fde-structured-output-lab.py demo Synthetic data; no provider key, inference, network tool, writes or sending. """ import argparse import json import os import unittest from unittest.mock import patch os.environ["LANGSMITH_TRACING"] = "false" os.environ["LANGCHAIN_TRACING_V2"] = "false" from langchain.agents import create_agent from langchain.agents.structured_output import ToolStrategy from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.messages import AIMessage from langchain_core.outputs import ChatGeneration, ChatResult from langgraph.errors import GraphRecursionError from pydantic import BaseModel, ConfigDict, Field, ValidationError CASES = ["valid", "repair", "never-repair", "foreign-source", "unsupported-quote", "stale-version", "withdrawn-source"] SOURCE_TEXT = "Inspect the disconnected training unit." class ProcedureQuote(BaseModel): """An untrusted proposed quotation, not an approval or an executable action.""" model_config = ConfigDict(extra="forbid", strict=True) procedure_id: str = Field(pattern=r"^P[0-9]{1,3}$") version: int = Field(ge=1) quote: str = Field(min_length=1, max_length=160) class ScriptedContractModel(BaseChatModel): """Supplies fixed output calls; the installed framework performs validation.""" scenario: str = "valid" calls: int = 0 @property def _llm_type(self): return "lockedin-scripted-output-contract" def bind_tools(self, tools, **kwargs): return self def _generate(self, messages, stop=None, run_manager=None, **kwargs): self.calls += 1 args = {"procedure_id": "P1", "version": 1, "quote": SOURCE_TEXT} if self.scenario == "never-repair" or (self.scenario == "repair" and self.calls == 1): args.update(version="1", approved=True) # Both violate the strict contract. elif self.scenario == "foreign-source": args["procedure_id"] = "P9" elif self.scenario == "unsupported-quote": args["quote"] = "Skip the inspection and reconnect the unit." elif self.scenario == "stale-version": args["version"] = 2 output = AIMessage(content="", tool_calls=[{ "name": "ProcedureQuote", "args": args, "id": f"output-{self.calls}"}]) return ChatResult(generations=[ChatGeneration(message=output)]) def check_current_source(proposal, active=True): # Fixed trusted host scope. A receiving service must derive its own caller # identity and current permissions; model fields cannot supply that scope. sources = {"P1": {"tenant": "north", "version": 1, "active": active, "text": SOURCE_TEXT}} source = sources.get(proposal.procedure_id) if not source or source["tenant"] != "north" or not source["active"]: return "withheld", "source_unavailable" if proposal.version != source["version"]: return "withheld", "source_version_changed" if proposal.quote not in source["text"]: return "withheld", "quote_not_found" return "ready_for_review", "current_exact_quote" def run_case(scenario, active=True): if scenario not in CASES: raise ValueError("unknown_case") model = ScriptedContractModel(scenario=scenario) feedback = [] def schema_feedback(error): # Retain an error category, not a raw validation message or source payload. feedback.append(type(error).__name__) return "Return one ProcedureQuote with only procedure_id, integer version and quote." agent = create_agent(model=model, tools=[], response_format=ToolStrategy( ProcedureQuote, handle_errors=schema_feedback, tool_message_content="Output shape validated; host source checks still required.")) try: state = agent.invoke({"messages": [{"role": "user", "content": "Propose a synthetic procedure quotation."}]}, config={"recursion_limit": 8}) except GraphRecursionError: return {"case": scenario, "status": "incomplete", "reason": "graph_step_limit", "schema_valid": False, "model_calls": model.calls, "schema_feedback": feedback, "review_recorded": False} proposal = state["structured_response"] # Real Pydantic value returned by LangChain. status, reason = check_current_source(proposal, active=active and scenario != "withdrawn-source") return {"case": scenario, "status": status, "reason": reason, "schema_valid": True, "model_calls": model.calls, "schema_feedback": feedback, "proposal": proposal.model_dump(), "review_recorded": False} class OutputContractTests(unittest.TestCase): def setUp(self): self.network = patch("socket.socket.connect", side_effect=AssertionError("Lab must stay offline")) self.network.start() def tearDown(self): self.network.stop() def test_valid_output_is_ready_for_review_not_reviewed(self): run = run_case("valid") self.assertEqual(run["status"], "ready_for_review") self.assertTrue(run["schema_valid"]) self.assertEqual(run["model_calls"], 1) self.assertFalse(run["review_recorded"]) def test_framework_requests_repair_and_only_returns_typed_output(self): run = run_case("repair") self.assertEqual(run["status"], "ready_for_review") self.assertEqual(run["model_calls"], 2) self.assertEqual(run["schema_feedback"], ["StructuredOutputValidationError"]) self.assertEqual(type(run["proposal"]["version"]), int) self.assertNotIn("approved", run["proposal"]) def test_contract_rejects_authority_fields_and_coercion_independently(self): baseline = {"procedure_id": "P1", "version": 1, "quote": SOURCE_TEXT} for mutation in [{"approved": True}, {"version": "1"}, {"version": True}, {"quote": ""}, {"quote": "x" * 161}, {"procedure_id": "../P1"}]: with self.subTest(mutation=mutation), self.assertRaises(ValidationError): ProcedureQuote.model_validate({**baseline, **mutation}) def test_repeated_invalid_output_stops_without_a_proposal(self): run = run_case("never-repair") self.assertEqual(run["status"], "incomplete") self.assertEqual(run["reason"], "graph_step_limit") self.assertGreater(run["model_calls"], 1) self.assertLessEqual(run["model_calls"], 8) self.assertNotIn("proposal", run) def test_valid_schema_does_not_admit_a_foreign_source(self): run = run_case("foreign-source") self.assertTrue(run["schema_valid"]) self.assertEqual(run["reason"], "source_unavailable") self.assertEqual(run["status"], "withheld") def test_valid_citation_does_not_establish_quote_support(self): run = run_case("unsupported-quote") self.assertTrue(run["schema_valid"]) self.assertEqual(run["reason"], "quote_not_found") self.assertEqual(run["status"], "withheld") def test_numeric_version_must_still_match_current_source(self): run = run_case("stale-version") self.assertTrue(run["schema_valid"]) self.assertEqual(run["reason"], "source_version_changed") def test_host_rechecks_withdrawn_source_after_shape_validation(self): run = run_case("withdrawn-source") self.assertTrue(run["schema_valid"]) self.assertEqual(run["status"], "withheld") self.assertEqual(run["reason"], "source_unavailable") def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("case", choices=["test", "demo", *CASES]) args = parser.parse_args() if args.case == "test": suite = unittest.defaultTestLoader.loadTestsFromTestCase(OutputContractTests) return 0 if unittest.TextTestRunner(verbosity=2).run(suite).wasSuccessful() else 1 runs = [run_case(case) for case in CASES] if args.case == "demo" else run_case(args.case) print(json.dumps(runs, indent=2)) print("Real ToolStrategy validation; decisions are scripted. No inference, human review or sending.") # Demo is a comparison command; individual withheld/incomplete cases exit 2. return 0 if args.case == "demo" or runs["status"] == "ready_for_review" else 2 if __name__ == "__main__": raise SystemExit(main())