From d60c73c284ea23ed2b2a4a883b54fc1c463ae3d3 Mon Sep 17 00:00:00 2001 From: Jiri Puc Date: Mon, 20 Jul 2026 13:22:22 +0200 Subject: [PATCH 1/3] feat: emit engine contracts as descriptors; stop prompts restating them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root-causes the contract-drift bug class (3 rejected-control cycles across live runs, each patched by hand-editing prompt prose). The engine now emits a machine-readable contract descriptor per iteration, referenced from paths.json, derived from real engine symbols: - LayerHandoff schema from LayerHandoff.model_json_schema() - reference grammar from the LOGICAL_REFERENCE_* scope/marker constants - control fields from ControlSignal.model_json_schema()/accepted_field_names() - eval-receipt applicability from check_runner_roles ∩ receipt-producing expected_outputs (real field is eval_receipt_refs) The outer and planner prompts drop the hand-restated schema/grammar/ eval_refs prose and point at the authoritative descriptor instead, so they can no longer drift from validation. Implemented by codex (gpt-5.6-sol xhigh); pending Grok + human review. Co-Authored-By: Codex (gpt-5.6-sol) Claude-Session: https://claude.ai/code/session_01EPYwF6cRYs2mfsfHF1hibc --- src/loopy_loop/contract_descriptors.py | 118 ++++++++++++++++++ src/loopy_loop/models.py | 15 +++ src/loopy_loop/references.py | 13 +- src/loopy_loop/sessions.py | 1 + .../workflows/outer/prompt.txt | 34 ++--- .../workflows/planner/prompt.txt | 33 +---- src/loopy_loop/worker.py | 30 ++++- src/tests/test_contract_descriptors.py | 117 +++++++++++++++++ src/tests/test_template_contracts.py | 69 +++++++++- src/tests/test_worker.py | 7 ++ 10 files changed, 373 insertions(+), 64 deletions(-) create mode 100644 src/loopy_loop/contract_descriptors.py create mode 100644 src/tests/test_contract_descriptors.py diff --git a/src/loopy_loop/contract_descriptors.py b/src/loopy_loop/contract_descriptors.py new file mode 100644 index 0000000..a5cd3ad --- /dev/null +++ b/src/loopy_loop/contract_descriptors.py @@ -0,0 +1,118 @@ +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any + +from pydantic import ValidationError + +from loopy_loop.models import ControlSignal +from loopy_loop.models import LayerHandoff +from loopy_loop.models import WorkflowRoster +from loopy_loop.models import WorkflowSetContract +from loopy_loop.references import LOGICAL_REFERENCE_IMPLICIT_SCOPES +from loopy_loop.references import LOGICAL_REFERENCE_NAMED_SCOPES +from loopy_loop.references import LOGICAL_REFERENCE_PATH_MARKER +from loopy_loop.sessions import EVAL_RECEIPTS_DIRNAME + + +def build_contracts_descriptor( + *, + workflow_contract: WorkflowSetContract | None, + workflow_roster: WorkflowRoster | Mapping[str, object] | None, +) -> dict[str, Any]: + """Describe agent-authored artifacts from their engine-owned definitions.""" + + roster = _validated_roster(workflow_roster=workflow_roster) + check_runner_roles = ( + workflow_contract.check_runner_roles if workflow_contract is not None else [] + ) + receipt_producing_roles = _receipt_producing_check_runner_roles( + check_runner_roles=check_runner_roles, workflow_roster=roster + ) + eval_receipt_refs_applicable = bool(receipt_producing_roles) + protocol_version = ( + workflow_contract.session_protocol_version + if workflow_contract is not None + else None + ) + + return { + "schema_version": 1, + "layer_handoff": { + "source": "loopy_loop.models.LayerHandoff", + "json_schema": LayerHandoff.model_json_schema(), + }, + "logical_references": { + "source": "loopy_loop.references", + "implicit": { + "shape": f"{LOGICAL_REFERENCE_PATH_MARKER}", + "scopes": sorted(LOGICAL_REFERENCE_IMPLICIT_SCOPES), + }, + "named": { + "shape": f":{LOGICAL_REFERENCE_PATH_MARKER}", + "scopes": sorted(LOGICAL_REFERENCE_NAMED_SCOPES), + }, + }, + "terminal_control": { + "source": "loopy_loop.models.ControlSignal", + "json_schema": ControlSignal.model_json_schema(), + "active_protocol_version": protocol_version, + "accepted_fields": ControlSignal.accepted_field_names( + schema_version=protocol_version, + eval_receipt_refs_applicable=eval_receipt_refs_applicable, + ), + "completion_role": ( + workflow_contract.completion_role + if workflow_contract is not None + else None + ), + "terminal_blocker_reporting_roles": ( + workflow_contract.terminal_blocker_reporting_roles + if workflow_contract is not None + else [] + ), + "eval_receipt_refs": { + "applicable": eval_receipt_refs_applicable, + "check_runner_roles": check_runner_roles, + "receipt_producing_check_runner_roles": receipt_producing_roles, + }, + }, + } + + +def _validated_roster( + *, workflow_roster: WorkflowRoster | Mapping[str, object] | None +) -> WorkflowRoster | None: + """Use a valid frozen roster without making descriptor emission a new gate.""" + + if workflow_roster is None: + return None + if isinstance(workflow_roster, WorkflowRoster): + return workflow_roster + try: + return WorkflowRoster.model_validate(workflow_roster) + except (ValidationError, ValueError): + return None + + +def _receipt_producing_check_runner_roles( + *, check_runner_roles: list[str], workflow_roster: WorkflowRoster | None +) -> list[str]: + """Return declared runners whose frozen roster advertises receipt output.""" + + if workflow_roster is None: + return [] + declared = set(check_runner_roles) + return sorted( + role.workflow_id + for role in workflow_roster.roles + if role.workflow_id in declared + and any(_is_eval_receipt_output(path=path) for path in role.expected_outputs) + ) + + +def _is_eval_receipt_output(*, path: str) -> bool: + normalized = path.strip().rstrip("/") + return normalized == EVAL_RECEIPTS_DIRNAME or normalized.startswith( + f"{EVAL_RECEIPTS_DIRNAME}/" + ) diff --git a/src/loopy_loop/models.py b/src/loopy_loop/models.py index ad53989..9ccec0a 100644 --- a/src/loopy_loop/models.py +++ b/src/loopy_loop/models.py @@ -685,6 +685,21 @@ class ControlSignal(BaseModel): evidence_refs: list[str] = Field(default_factory=list) created_at: datetime | None = Field(default=None) + @classmethod + def accepted_field_names( + cls, *, schema_version: int | None, eval_receipt_refs_applicable: bool + ) -> list[str]: + """Project model fields onto one protocol and frozen eval contract.""" + + fields = set(cls.model_fields) + if schema_version == 2: + fields.difference_update({"eval_receipt_refs", "handoff_ref"}) + elif schema_version == 3: + fields.discard("eval_receipt_ref") + if not eval_receipt_refs_applicable: + fields.discard("eval_receipt_refs") + return sorted(fields) + @field_validator("schema_version") @classmethod def validate_schema_version(cls, value: int) -> int: diff --git a/src/loopy_loop/references.py b/src/loopy_loop/references.py index f6e19a8..e156f74 100644 --- a/src/loopy_loop/references.py +++ b/src/loopy_loop/references.py @@ -18,8 +18,9 @@ TRACE_MANIFEST_FILENAME = "trace_manifest.json" _SAFE_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]*\Z") -_IMPLICIT_SCOPES = frozenset({"repo", "session", "root", "parent"}) -_NAMED_SCOPES = frozenset({"session", "trace"}) +LOGICAL_REFERENCE_IMPLICIT_SCOPES = frozenset({"repo", "session", "root", "parent"}) +LOGICAL_REFERENCE_NAMED_SCOPES = frozenset({"session", "trace"}) +LOGICAL_REFERENCE_PATH_MARKER = ":/" class LogicalReferenceError(ValueError): @@ -701,18 +702,20 @@ def _parse_reference(*, reference: str) -> tuple[str, str | None, tuple[str, ... raise LogicalReferenceError( f"logical reference contains a forbidden character: {reference!r}" ) - marker = ":/" + marker = LOGICAL_REFERENCE_PATH_MARKER if marker not in reference: raise LogicalReferenceError( f"logical reference does not match the required grammar: {reference!r}" ) prefix, relative = reference.split(marker, 1) prefix_parts = prefix.split(":") - if len(prefix_parts) == 1 and prefix_parts[0] in _IMPLICIT_SCOPES: + if len(prefix_parts) == 1 and prefix_parts[0] in LOGICAL_REFERENCE_IMPLICIT_SCOPES: scope = prefix_parts[0] identifier = None elif ( - len(prefix_parts) == 2 and prefix_parts[0] in _NAMED_SCOPES and prefix_parts[1] + len(prefix_parts) == 2 + and prefix_parts[0] in LOGICAL_REFERENCE_NAMED_SCOPES + and prefix_parts[1] ): scope, identifier = prefix_parts _validate_id(value=identifier, label=f"{scope} reference ID") diff --git a/src/loopy_loop/sessions.py b/src/loopy_loop/sessions.py index 2abed20..c920839 100644 --- a/src/loopy_loop/sessions.py +++ b/src/loopy_loop/sessions.py @@ -63,6 +63,7 @@ GOAL_CHECK_FILENAME = "goal_check.json" EVAL_REQUEST_FILENAME = "eval_request.md" PATHS_FILENAME = "paths.json" +CONTRACTS_FILENAME = "contracts.json" WORKER_SESSIONS_FILENAME = "worker_sessions.json" ASSIGNMENT_FILENAME = "assignment.json" WORKFLOW_SNAPSHOT_DIRNAME = "workflow_snapshot" diff --git a/src/loopy_loop/templates/inner_outer_eval/.loopy_loop/workflow_sets/inner_outer_eval/workflows/outer/prompt.txt b/src/loopy_loop/templates/inner_outer_eval/.loopy_loop/workflow_sets/inner_outer_eval/workflows/outer/prompt.txt index ce391f4..f9f6f0f 100644 --- a/src/loopy_loop/templates/inner_outer_eval/.loopy_loop/workflow_sets/inner_outer_eval/workflows/outer/prompt.txt +++ b/src/loopy_loop/templates/inner_outer_eval/.loopy_loop/workflow_sets/inner_outer_eval/workflows/outer/prompt.txt @@ -19,11 +19,8 @@ What you own table, covered by a test. - Start from: the tile model in the board module and the drop table in the design notes. -- project_state/handoff.json must match the engine's LayerHandoff schema - exactly — extra fields are rejected. Fields: schema_version (1), session_id, - goal_sha256, revision (bump on every change), producer {workflow_id, - attempt_id}, summary, accepted_outcomes, open_work, risks, decision_refs, - evidence_refs, delivery_refs, eval_refs, updated_at. Rewrite it atomically +- project_state/handoff.json is the rolling layer handoff. Its authoritative + schema is in paths.json → contracts; conform exactly and rewrite it atomically after any material change. Sizing the work @@ -50,28 +47,11 @@ not a veto. Completion When this layer's goal is met, bring the plan, tasks, ledger, and handoff current, -then atomically publish successful control to control.json: - -{ - "schema_version": 3, - "control_id": "stable-unique-id", - "state": "stopped", - "stop_reason": "goal_met", - "reason": "Why this layer's goal is complete despite any open or conflicting evidence.", - "producer": {"session_id": "from assignment header", "workflow_id": "outer", "attempt_id": "from assignment header"}, - "evidence_refs": [], - "handoff_ref": "session:/project_state/handoff.json", - "created_at": "RFC3339 timestamp" -} -Evidence refs must use the engine's logical-reference grammar `:/` -with scopes repo, session, parent, root — e.g. `session:/project_state/evidence/audit.md`, -`repo:/design/decisions.md`. Bare URLs, git SHAs, and absolute filesystem paths -are rejected; put those inside a referenced file instead. -Do not include an `eval_refs` field: this stock contract runs eval as advisory -(no sealed eval receipts are produced), and `eval_refs` only accepts a -current-session `eval_receipts/*.json` file — citing anything else is rejected. -Cite eval verdicts through `evidence_refs` pointing at -`session:/project_state/eval_results.md` instead. +then atomically publish successful control to control.json. The authoritative +active control fields and logical-reference grammar are in +paths.json → contracts; conform exactly. In human terms: use this attempt's +identity, explain why the goal is met, and cite only durable evidence that +contract permits. Stopping short of the goal is only for a genuinely terminal blocker, after autonomous routes are exhausted. diff --git a/src/loopy_loop/templates/pm_planner_dispatcher/.loopy_loop/workflow_sets/pm_planner_dispatcher/workflows/planner/prompt.txt b/src/loopy_loop/templates/pm_planner_dispatcher/.loopy_loop/workflow_sets/pm_planner_dispatcher/workflows/planner/prompt.txt index 924fd01..8072e7e 100644 --- a/src/loopy_loop/templates/pm_planner_dispatcher/.loopy_loop/workflow_sets/pm_planner_dispatcher/workflows/planner/prompt.txt +++ b/src/loopy_loop/templates/pm_planner_dispatcher/.loopy_loop/workflow_sets/pm_planner_dispatcher/workflows/planner/prompt.txt @@ -14,11 +14,8 @@ What you own - One record per milestone outcome under project_state/tasks/: its objective, what done looks like, dependencies, the child request/session it maps to, and open questions. Think here so the plan can stay short. -- project_state/handoff.json must match the engine's LayerHandoff schema - exactly — extra fields are rejected. Fields: schema_version (1), session_id, - goal_sha256, revision (bump on every change), producer {workflow_id, - attempt_id}, summary, accepted_outcomes, open_work, risks, decision_refs, - evidence_refs, delivery_refs, eval_refs, updated_at. Rewrite it atomically +- project_state/handoff.json is the rolling program handoff. Its authoritative + schema is in paths.json → contracts; conform exactly and rewrite it atomically after any material change. How to plan @@ -37,28 +34,10 @@ How to plan Completion When the program goal itself is met — not merely a child's scoped goal — bring the plan, tasks, ledger, and handoff current, then atomically publish successful -control to control.json: - -{ - "schema_version": 3, - "control_id": "stable-unique-id", - "state": "stopped", - "stop_reason": "goal_met", - "reason": "Why the program goal is complete despite any open or conflicting evidence.", - "producer": {"session_id": "from assignment header", "workflow_id": "planner", "attempt_id": "from assignment header"}, - "evidence_refs": [], - "handoff_ref": "session:/project_state/handoff.json", - "created_at": "RFC3339 timestamp" -} -Evidence refs must use the engine's logical-reference grammar `:/` -with scopes repo, session, parent, root — e.g. `session:/project_state/evidence/audit.md`, -`repo:/design/decisions.md`. Bare URLs, git SHAs, and absolute filesystem paths -are rejected; put those inside a referenced file instead. -Do not include an `eval_refs` field: this stock contract runs eval as advisory -(no sealed eval receipts are produced), and `eval_refs` only accepts a -current-session `eval_receipts/*.json` file — citing anything else is rejected. -Cite eval verdicts through `evidence_refs` pointing at -`session:/project_state/eval_results.md` instead. +control to control.json. The authoritative active control fields and +logical-reference grammar are in paths.json → contracts; conform exactly. In +human terms: use this attempt's identity, explain why the program goal is met, +and cite only durable evidence that contract permits. Evaluation, when you use it, is advisory: a failed or missing check is input to your judgment, not a veto. Stopping short of the goal is only for a genuinely diff --git a/src/loopy_loop/worker.py b/src/loopy_loop/worker.py index de77c3a..56c0a61 100644 --- a/src/loopy_loop/worker.py +++ b/src/loopy_loop/worker.py @@ -25,6 +25,7 @@ from loopy_loop.config import load_workflow_config from loopy_loop.config import load_workflow_set_preamble from loopy_loop.config import workflow_set_workflows_dir_path +from loopy_loop.contract_descriptors import build_contracts_descriptor from loopy_loop.git_evidence import capture_git_evidence from loopy_loop.git_evidence import GitEvidenceError from loopy_loop.harness_runner import run_harness_iteration @@ -43,9 +44,11 @@ from loopy_loop.models import utc_now from loopy_loop.models import WORKER_PROTOCOL_VERSION from loopy_loop.models import WorkerIdentity +from loopy_loop.models import WorkflowSetContract from loopy_loop.sessions import append_jsonl_record from loopy_loop.sessions import assignment_path from loopy_loop.sessions import child_requests_dir_path +from loopy_loop.sessions import CONTRACTS_FILENAME from loopy_loop.sessions import control_path from loopy_loop.sessions import ensure_iteration_dir from loopy_loop.sessions import eval_checks_dir_path @@ -283,6 +286,7 @@ def _run_task( scratch_dir: Path | None = None assignment: AttemptAssignment | None = None assignment_file: Path | None = None + workflow_contract: WorkflowSetContract | None = None caller_context: dict[str, object] | None = None fatal_error: str | None = None started = time.monotonic() @@ -326,7 +330,7 @@ def _run_task( workflow_snapshot=task.workflow_snapshot, repository_id=task.repository_id, ) - (config_payload, prompt_text, _, frozen_config_snapshot) = ( + (config_payload, prompt_text, workflow_contract, frozen_config_snapshot) = ( verify_workflow_snapshot( descriptor=task.workflow_snapshot, repo_root=root, @@ -501,6 +505,7 @@ def _run_task( repo_root=root, assignment=assignment, assignment_file=assignment_file, + workflow_contract=workflow_contract, ) write_iteration_inputs( iteration_dir=iteration_dir, rendered_prompt=rendered_prompt @@ -804,6 +809,7 @@ def _render_prompt( repo_root: Path | None = None, assignment: AttemptAssignment | None = None, assignment_file: Path | None = None, + workflow_contract: WorkflowSetContract | None = None, ) -> str: """Render the diet iteration header plus the workflow body. @@ -842,6 +848,7 @@ def _render_prompt( emits_goal_check=emits_goal_check, assignment=assignment, assignment_file=assignment_file, + workflow_contract=workflow_contract, ) preamble = load_workflow_set_preamble(repo_root=root, workflow_set=workflow_set) @@ -895,14 +902,15 @@ def _write_iteration_paths( emits_goal_check: bool, assignment: AttemptAssignment | None, assignment_file: Path | None, + workflow_contract: WorkflowSetContract | None, ) -> None: """Write the full machine path map the diet header references by name. Holds every absolute path the old header inlined plus the complete v3 assignment path map (rosters, scheduler view, workflow contract as files), - and previous_worker_sessions: the prior iteration's team-harness - worker_sessions.json for selective session reuse (context-and-eval-economy - A4), or null when none exists. + the engine-derived artifact contracts, and previous_worker_sessions: the + prior iteration's team-harness worker_sessions.json for selective session + reuse (context-and-eval-economy A4), or null when none exists. """ goal_check_output = ( @@ -930,6 +938,19 @@ def _write_iteration_paths( if root_session_id is not None else None ) + contracts_path = (path.parent / CONTRACTS_FILENAME).resolve() + roster_payload = ( + assignment.context.get("workflow_roster") if assignment is not None else None + ) + write_json_atomic( + path=contracts_path, + payload=build_contracts_descriptor( + workflow_contract=workflow_contract, + workflow_roster=( + roster_payload if isinstance(roster_payload, dict) else None + ), + ), + ) payload: dict[str, object] = { "schema_version": 1, "session_id": session_id, @@ -949,6 +970,7 @@ def _write_iteration_paths( if previous_worker_sessions is not None else None ), + "contracts": str(contracts_path), "session_paths": { "goal": str( session_goal_path(repo_root=repo_root, session_id=session_id).resolve() diff --git a/src/tests/test_contract_descriptors.py b/src/tests/test_contract_descriptors.py new file mode 100644 index 0000000..68f972f --- /dev/null +++ b/src/tests/test_contract_descriptors.py @@ -0,0 +1,117 @@ +from __future__ import annotations + +from loopy_loop.contract_descriptors import build_contracts_descriptor +from loopy_loop.models import ControlSignal +from loopy_loop.models import LayerHandoff +from loopy_loop.models import utc_now +from loopy_loop.models import WorkflowRoster +from loopy_loop.models import WorkflowRosterRole +from loopy_loop.models import WorkflowSetContract +from loopy_loop.references import LOGICAL_REFERENCE_IMPLICIT_SCOPES +from loopy_loop.references import LOGICAL_REFERENCE_NAMED_SCOPES + + +def _workflow_contract() -> WorkflowSetContract: + return WorkflowSetContract.model_validate( + { + "schema_version": 1, + "session_protocol_version": 3, + "roles": { + "outer": {"responsibility": "Orchestrate."}, + "eval_runner": {"responsibility": "Run checks."}, + }, + "orchestration": { + "completion_role": "outer", + "plan_owner": "outer", + "handoff_owner": "outer", + "task_acceptance_owner": "outer", + }, + "evaluation": { + "advisory": True, + "check_author_roles": [], + "check_runner_roles": ["eval_runner"], + }, + "terminal_blocker_reporting_roles": ["outer", "eval_runner"], + } + ) + + +def _workflow_roster(*, expected_outputs: list[str]) -> WorkflowRoster: + return WorkflowRoster( + session_id="session-contract-test", + workflow_contract_sha256="sha256:" + "a" * 64, + created_at=utc_now(), + completion_role="outer", + roles=[ + WorkflowRosterRole( + workflow_id="outer", + responsibility="Orchestrate.", + cadence={}, + expected_outputs=["project_state/handoff.json"], + ), + WorkflowRosterRole( + workflow_id="eval_runner", + responsibility="Run checks.", + cadence={}, + expected_outputs=expected_outputs, + authorities=["eval_check_runner"], + ), + ], + ) + + +def test_descriptor_uses_engine_schema_and_reference_scope_constants() -> None: + descriptor = build_contracts_descriptor( + workflow_contract=_workflow_contract(), + workflow_roster=_workflow_roster( + expected_outputs=["project_state/eval_state.md"] + ), + ) + + assert descriptor["layer_handoff"]["json_schema"] == ( + LayerHandoff.model_json_schema() + ) + assert descriptor["terminal_control"]["json_schema"] == ( + ControlSignal.model_json_schema() + ) + references = descriptor["logical_references"] + assert references["implicit"] == { + "shape": ":/", + "scopes": sorted(LOGICAL_REFERENCE_IMPLICIT_SCOPES), + } + assert references["named"] == { + "shape": "::/", + "scopes": sorted(LOGICAL_REFERENCE_NAMED_SCOPES), + } + + +def test_eval_receipt_refs_applicability_follows_frozen_runner_outputs() -> None: + contract = _workflow_contract() + retired = build_contracts_descriptor( + workflow_contract=contract, + workflow_roster=_workflow_roster( + expected_outputs=["project_state/eval_state.md"] + ), + ) + receipt_producing = build_contracts_descriptor( + workflow_contract=contract, + workflow_roster=_workflow_roster(expected_outputs=["eval_receipts/"]), + ) + + retired_control = retired["terminal_control"] + assert retired_control["eval_receipt_refs"] == { + "applicable": False, + "check_runner_roles": ["eval_runner"], + "receipt_producing_check_runner_roles": [], + } + assert "eval_receipt_refs" not in retired_control["accepted_fields"] + + receipt_control = receipt_producing["terminal_control"] + assert receipt_control["eval_receipt_refs"] == { + "applicable": True, + "check_runner_roles": ["eval_runner"], + "receipt_producing_check_runner_roles": ["eval_runner"], + } + assert "eval_receipt_refs" in receipt_control["accepted_fields"] + assert "eval_receipt_ref" not in receipt_control["accepted_fields"] + assert set(receipt_control["accepted_fields"]) <= set(ControlSignal.model_fields) diff --git a/src/tests/test_template_contracts.py b/src/tests/test_template_contracts.py index c84abf0..ed8e714 100644 --- a/src/tests/test_template_contracts.py +++ b/src/tests/test_template_contracts.py @@ -11,6 +11,7 @@ from loopy_loop.config import load_workflow_set_preamble from loopy_loop.config import run_preflight from loopy_loop.scheduler import choose_next_workflow +from loopy_loop.worker import _render_prompt TEMPLATES_ROOT = Path(__file__).resolve().parents[1] / "loopy_loop" / "templates" TEMPLATE_SETS = ("inner_outer_eval", "pm_planner_dispatcher") @@ -165,6 +166,53 @@ def test_role_prompts_fit_one_screen() -> None: assert line_count <= 80, f"{path} has {line_count} lines" +def test_stock_orchestrator_render_stays_within_header_budget( + tmp_path: Path, snapshot_factory: Any +) -> None: + """External contract files do not inflate the CI-budgeted prompt header.""" + + for template, workflow in ( + ("inner_outer_eval", "outer"), + ("pm_planner_dispatcher", "planner"), + ): + repo_root = _template_root(template) + preflight = run_preflight(repo_root=repo_root) + prompt_body = _workflow_prompt( + template=template, workflow_set=template, workflow=workflow + ) + snapshot = snapshot_factory( + goal=preflight.root_config.goal, + workflow_set=template, + completion_criteria=preflight.root_config.completion_criteria, + stop_criteria=preflight.root_config.stop_criteria, + ) + iteration_dir = tmp_path / template / workflow + rendered = _render_prompt( + config_snapshot=snapshot, + session_id=f"session-{template}", + workflow_set=template, + iteration=1, + workflow_id=workflow, + iteration_dir=iteration_dir, + harness_output_root=iteration_dir / "harness", + workflow_prompt=prompt_body, + repo_root=repo_root, + workflow_contract=preflight.workflow_contract, + ) + + before_body = rendered.split("\n\nWorkflow body:", 1)[0] + preamble = load_workflow_set_preamble( + repo_root=repo_root, workflow_set=template + ) + assert preamble is not None + scaffold_bytes = ( + len(before_body.encode("utf-8")) + - len(snapshot.goal.encode("utf-8")) + - len(preamble.encode("utf-8")) + ) + assert scaffold_bytes <= 2048, (template, scaffold_bytes) + + def test_prompts_drop_retired_ceremony_and_model_mandates() -> None: """Prompts and the shared preamble carry no retired ceremony or vendor names.""" @@ -229,7 +277,8 @@ def test_only_layer_orchestrators_publish_successful_terminal_control() -> None: writers: set[tuple[str, str]] = set() for template in TEMPLATE_SETS: for path in _prompt_paths(template): - if '"stop_reason": "goal_met"' in path.read_text(encoding="utf-8"): + words = " ".join(path.read_text(encoding="utf-8").split()) + if "publish successful control to control.json" in words: writers.add((template, path.parent.name)) assert writers == { @@ -238,6 +287,24 @@ def test_only_layer_orchestrators_publish_successful_terminal_control() -> None: } +def test_orchestrator_prompts_point_to_emitted_contracts_without_restatement() -> None: + """Outer/planner defer strict artifact shape to engine-emitted contracts.""" + + for template, workflow in ( + ("inner_outer_eval", "outer"), + ("pm_planner_dispatcher", "planner"), + ): + prompt = _workflow_prompt( + template=template, workflow_set=template, workflow=workflow + ) + + assert prompt.count("paths.json → contracts") == 2 + assert "Fields: schema_version" not in prompt + assert ":/" not in prompt + assert "Do not include an `eval_refs` field" not in prompt + assert '"stop_reason": "goal_met"' not in prompt + + def test_dispatcher_teaches_v3_child_request() -> None: """Dispatcher transports the milestone as one authored v3 goal text.""" diff --git a/src/tests/test_worker.py b/src/tests/test_worker.py index fb52db2..45c378c 100644 --- a/src/tests/test_worker.py +++ b/src/tests/test_worker.py @@ -9,6 +9,7 @@ import pytest from loopy_loop.models import IterationResult +from loopy_loop.models import LayerHandoff from loopy_loop.models import TaskResponse from loopy_loop.sessions import create_session_dir from loopy_loop.sessions import eval_readiness_dir_path @@ -214,6 +215,12 @@ def fake_run_harness_iteration(**kwargs: Any) -> IterationResult: assert paths["schema_version"] == 1 assert paths["session_paths"]["project_state"].endswith("project_state") assert paths["previous_worker_sessions"] is None + contracts_path = Path(paths["contracts"]) + assert contracts_path == paths_file.parent / "contracts.json" + contracts = json.loads(contracts_path.read_text(encoding="utf-8")) + assert contracts["layer_handoff"]["json_schema"] == ( + LayerHandoff.model_json_schema() + ) def test_worker_includes_goal_check_path_for_emitting_workflow( From b8aaee33ee32e9db78611c0411ef5bb3fe1d74ff Mon Sep 17 00:00:00 2001 From: Jiri Puc Date: Mon, 20 Jul 2026 14:13:10 +0200 Subject: [PATCH 2/3] fix(contracts): authoritative accepted/required fields, validated example, ref rules Addresses adversarial review (Grok) of the contract descriptor: 1. Stop presenting the raw multi-version ControlSignal.model_json_schema() as "conform exactly" (it under-described required-for-stopped fields and advertised fields not accepted in the active protocol, e.g. a singular eval_receipt_ref). The descriptor now exposes authoritative accepted_fields + required_fields for the active protocol, derived from ControlSignal validation; the raw schema is subordinate. 2. Restore load-bearing guidance inside the descriptor (drift-proof): a model-validated minimal goal_met example built only from accepted/required fields; evidence-ref rules incl. the path-segment restrictions the reference parser enforces (no empty/./.. segments, no backslash/NUL, not absolute after :/), and the file requirement matching the coordinator's real asserted-artifact validator. 3. Discovery: contracts path added to the diet header's key-paths line. 4. Tests: stock inner_outer_eval roster -> applicable False + eval fields omitted; older-roster compat; round-trip (build from descriptor -> ControlSignal.model_validate accepts, reject without control_id); worker asserts protocol/accepted/required/applicability; header budget rendered with a roster-carrying assignment. Co-Authored-By: Codex (gpt-5.6-sol) Claude-Session: https://claude.ai/code/session_01EPYwF6cRYs2mfsfHF1hibc --- src/loopy_loop/contract_descriptors.py | 135 +++++++++++++++++++++++-- src/loopy_loop/models.py | 67 ++++++++++-- src/loopy_loop/references.py | 13 ++- src/loopy_loop/worker.py | 4 +- src/tests/test_contract_descriptors.py | 120 ++++++++++++++++++---- src/tests/test_worker.py | 80 +++++++++++++++ 6 files changed, 375 insertions(+), 44 deletions(-) diff --git a/src/loopy_loop/contract_descriptors.py b/src/loopy_loop/contract_descriptors.py index a5cd3ad..8d977e1 100644 --- a/src/loopy_loop/contract_descriptors.py +++ b/src/loopy_loop/contract_descriptors.py @@ -9,7 +9,10 @@ from loopy_loop.models import LayerHandoff from loopy_loop.models import WorkflowRoster from loopy_loop.models import WorkflowSetContract +from loopy_loop.references import LOGICAL_REFERENCE_ABSOLUTE_PATH_PREFIX +from loopy_loop.references import LOGICAL_REFERENCE_FORBIDDEN_CHARACTERS from loopy_loop.references import LOGICAL_REFERENCE_IMPLICIT_SCOPES +from loopy_loop.references import LOGICAL_REFERENCE_INVALID_PATH_SEGMENTS from loopy_loop.references import LOGICAL_REFERENCE_NAMED_SCOPES from loopy_loop.references import LOGICAL_REFERENCE_PATH_MARKER from loopy_loop.sessions import EVAL_RECEIPTS_DIRNAME @@ -35,6 +38,23 @@ def build_contracts_descriptor( if workflow_contract is not None else None ) + accepted_fields = ControlSignal.accepted_field_names( + schema_version=protocol_version, + eval_receipt_refs_applicable=eval_receipt_refs_applicable, + ) + required_fields = ControlSignal.required_field_names( + schema_version=protocol_version, state="stopped", stop_reason="goal_met" + ) + completion_role = ( + workflow_contract.completion_role if workflow_contract is not None else None + ) + goal_met_example = _goal_met_control_example( + protocol_version=protocol_version, + completion_role=completion_role, + session_id=roster.session_id if roster is not None else None, + accepted_fields=accepted_fields, + required_fields=required_fields, + ) return { "schema_version": 1, @@ -52,25 +72,44 @@ def build_contracts_descriptor( "shape": f":{LOGICAL_REFERENCE_PATH_MARKER}", "scopes": sorted(LOGICAL_REFERENCE_NAMED_SCOPES), }, + "path_rules": { + "path_must_be_relative_after_marker": True, + "forbidden_absolute_path_prefix": ( + LOGICAL_REFERENCE_ABSOLUTE_PATH_PREFIX + ), + "forbidden_embedded_path_marker": LOGICAL_REFERENCE_PATH_MARKER, + "forbidden_characters": sorted(LOGICAL_REFERENCE_FORBIDDEN_CHARACTERS), + "forbidden_segments": sorted(LOGICAL_REFERENCE_INVALID_PATH_SEGMENTS), + }, }, "terminal_control": { "source": "loopy_loop.models.ControlSignal", - "json_schema": ControlSignal.model_json_schema(), "active_protocol_version": protocol_version, - "accepted_fields": ControlSignal.accepted_field_names( - schema_version=protocol_version, - eval_receipt_refs_applicable=eval_receipt_refs_applicable, - ), - "completion_role": ( - workflow_contract.completion_role - if workflow_contract is not None - else None + "accepted_fields": accepted_fields, + "accepted_fields_are_authoritative": True, + "required_fields": required_fields, + "required_fields_for": {"state": "stopped", "stop_reason": "goal_met"}, + "accepted_field_schemas": _accepted_control_field_schemas( + accepted_fields=accepted_fields ), + "goal_met_example": goal_met_example, + "completion_role": completion_role, "terminal_blocker_reporting_roles": ( workflow_contract.terminal_blocker_reporting_roles if workflow_contract is not None else [] ), + "evidence_refs": { + "value_kind": "logical_reference", + "must_resolve_to": "file", + "path_must_be_nonempty": True, + "not_valid_reference_values": [ + "URL", + "git SHA", + "absolute filesystem path", + ], + "grammar_and_path_rules": "#/logical_references", + }, "eval_receipt_refs": { "applicable": eval_receipt_refs_applicable, "check_runner_roles": check_runner_roles, @@ -80,6 +119,84 @@ def build_contracts_descriptor( } +def _accepted_control_field_schemas(*, accepted_fields: list[str]) -> dict[str, object]: + """Return model-owned field schemas without exposing other protocol versions.""" + + schema = ControlSignal.model_json_schema() + properties = schema.get("properties", {}) + definitions = schema.get("$defs", {}) + return { + field_name: _inline_local_schema_refs( + value=properties[field_name], definitions=definitions + ) + for field_name in accepted_fields + if field_name in properties + } + + +def _inline_local_schema_refs( + *, value: object, definitions: Mapping[str, object] +) -> object: + """Inline local Pydantic definitions so each emitted field schema stands alone.""" + + if isinstance(value, list): + return [ + _inline_local_schema_refs(value=item, definitions=definitions) + for item in value + ] + if not isinstance(value, dict): + return value + reference = value.get("$ref") + if isinstance(reference, str) and reference.startswith("#/$defs/"): + definition = definitions.get(reference.removeprefix("#/$defs/")) + if isinstance(definition, dict): + value = { + **definition, + **{key: item for key, item in value.items() if key != "$ref"}, + } + return { + key: _inline_local_schema_refs(value=item, definitions=definitions) + for key, item in value.items() + } + + +def _goal_met_control_example( + *, + protocol_version: int | None, + completion_role: str | None, + session_id: str | None, + accepted_fields: list[str], + required_fields: list[str], +) -> dict[str, object] | None: + """Build and model-validate the minimal required goal-met authoring example.""" + + if protocol_version is None: + return None + missing_from_contract = set(required_fields) - set(accepted_fields) + if missing_from_contract: + raise ValueError( + "required terminal-control fields are not accepted: " + f"{sorted(missing_from_contract)}" + ) + placeholders: dict[str, object] = { + "schema_version": protocol_version, + "control_id": "control-goal-met-current-attempt", + "state": "stopped", + "stop_reason": "goal_met", + "reason": "Explain why this layer's scoped goal is complete.", + "producer": { + "session_id": session_id or "current-session-id", + "workflow_id": completion_role or "declared-completion-role", + "attempt_id": "current-attempt-id", + }, + "eval_receipt_ref": "session:/eval_receipts/current-eval-receipt.json", + "created_at": "1970-01-01T00:00:00Z", + } + example = {field_name: placeholders[field_name] for field_name in required_fields} + ControlSignal.model_validate(example) + return example + + def _validated_roster( *, workflow_roster: WorkflowRoster | Mapping[str, object] | None ) -> WorkflowRoster | None: diff --git a/src/loopy_loop/models.py b/src/loopy_loop/models.py index 9ccec0a..03e7522 100644 --- a/src/loopy_loop/models.py +++ b/src/loopy_loop/models.py @@ -4,6 +4,7 @@ from datetime import UTC import re from typing import Any +from typing import ClassVar from typing import Literal from typing import Self @@ -672,6 +673,17 @@ class SignalProducer(BaseModel): class ControlSignal(BaseModel): + _STOPPED_REQUIRED_FIELD_NAMES: ClassVar[frozenset[str]] = frozenset({"stop_reason"}) + _IDENTITY_BOUND_STOPPED_REQUIRED_FIELD_NAMES: ClassVar[frozenset[str]] = frozenset( + {"control_id", "producer", "created_at"} + ) + _V2_GOAL_MET_REQUIRED_FIELD_NAMES: ClassVar[frozenset[str]] = frozenset( + {"eval_receipt_ref"} + ) + _IDENTITY_BOUND_BLOCKER_REQUIRED_FIELD_NAMES: ClassVar[frozenset[str]] = frozenset( + {"attempted_routes"} + ) + state: Literal["running", "stopped"] = Field(...) reason: str = Field(...) stop_reason: Literal["goal_met", "unresolvable_error"] | None = Field(default=None) @@ -700,6 +712,30 @@ def accepted_field_names( fields.discard("eval_receipt_refs") return sorted(fields) + @classmethod + def required_field_names( + cls, + *, + schema_version: int | None, + state: Literal["running", "stopped"], + stop_reason: Literal["goal_met", "unresolvable_error"] | None, + ) -> list[str]: + """Project model and conditional validator requirements onto one signal.""" + + fields = { + name for name, field in cls.model_fields.items() if field.is_required() + } + if state != "stopped": + return sorted(fields) + fields.update(cls._STOPPED_REQUIRED_FIELD_NAMES) + if schema_version in {2, 3}: + fields.update(cls._IDENTITY_BOUND_STOPPED_REQUIRED_FIELD_NAMES) + if stop_reason == "unresolvable_error": + fields.update(cls._IDENTITY_BOUND_BLOCKER_REQUIRED_FIELD_NAMES) + if schema_version == 2 and stop_reason == "goal_met": + fields.update(cls._V2_GOAL_MET_REQUIRED_FIELD_NAMES) + return sorted(fields) + @field_validator("schema_version") @classmethod def validate_schema_version(cls, value: int) -> int: @@ -717,31 +753,41 @@ def validate_stop_reason(self) -> Self: if self.state == "running" and self.stop_reason is not None: raise ValueError("running control state must not set stop_reason") - if self.state == "stopped" and self.stop_reason is None: + if self.state == "stopped" and any( + getattr(self, field_name) is None + for field_name in self._STOPPED_REQUIRED_FIELD_NAMES + ): raise ValueError("stopped control state must set stop_reason") if self.schema_version in {2, 3} and self.state == "stopped": - if ( - self.control_id is None - or self.producer is None - or self.created_at is None + if any( + getattr(self, field_name) is None + for field_name in self._IDENTITY_BOUND_STOPPED_REQUIRED_FIELD_NAMES ): raise ValueError( "identity-bound stopped control requires control_id, producer, " "and created_at" ) - if not SAFE_DURABLE_ID_PATTERN.fullmatch(self.control_id): + control_id = self.control_id + assert control_id is not None + if not SAFE_DURABLE_ID_PATTERN.fullmatch(control_id): raise ValueError( "identity-bound control_id must be a filesystem-safe identifier" ) if not self.reason.strip(): raise ValueError("terminal control reason must be nonblank") if self.schema_version == 2 and self.state == "stopped": - if self.stop_reason == "goal_met" and self.eval_receipt_ref is None: + if self.stop_reason == "goal_met" and any( + getattr(self, field_name) is None + for field_name in self._V2_GOAL_MET_REQUIRED_FIELD_NAMES + ): raise ValueError("v2 goal_met control requires eval_receipt_ref") if self.eval_receipt_refs or self.handoff_ref is not None: raise ValueError("v2 control must not set v3 reference fields") if self.stop_reason == "unresolvable_error": - if not self.attempted_routes: + if any( + not getattr(self, field_name) + for field_name in self._IDENTITY_BOUND_BLOCKER_REQUIRED_FIELD_NAMES + ): raise ValueError("v2 unresolvable_error requires attempted_routes") if any(not route.strip() for route in self.attempted_routes): raise ValueError( @@ -759,7 +805,10 @@ def validate_stop_reason(self) -> Self: if self.handoff_ref is not None and not self.handoff_ref.strip(): raise ValueError("v3 handoff_ref must be nonblank when set") if self.stop_reason == "unresolvable_error": - if not self.attempted_routes: + if any( + not getattr(self, field_name) + for field_name in self._IDENTITY_BOUND_BLOCKER_REQUIRED_FIELD_NAMES + ): raise ValueError("v3 unresolvable_error requires attempted_routes") if any(not route.strip() for route in self.attempted_routes): raise ValueError( diff --git a/src/loopy_loop/references.py b/src/loopy_loop/references.py index e156f74..b0128a8 100644 --- a/src/loopy_loop/references.py +++ b/src/loopy_loop/references.py @@ -21,6 +21,9 @@ LOGICAL_REFERENCE_IMPLICIT_SCOPES = frozenset({"repo", "session", "root", "parent"}) LOGICAL_REFERENCE_NAMED_SCOPES = frozenset({"session", "trace"}) LOGICAL_REFERENCE_PATH_MARKER = ":/" +LOGICAL_REFERENCE_FORBIDDEN_CHARACTERS = frozenset({"\x00", "\\"}) +LOGICAL_REFERENCE_INVALID_PATH_SEGMENTS = frozenset({"", ".", ".."}) +LOGICAL_REFERENCE_ABSOLUTE_PATH_PREFIX = "/" class LogicalReferenceError(ValueError): @@ -698,7 +701,9 @@ def _parse_reference(*, reference: str) -> tuple[str, str | None, tuple[str, ... if not reference: raise LogicalReferenceError("logical reference must be a non-empty string") - if "\x00" in reference or "\\" in reference: + if any( + character in reference for character in LOGICAL_REFERENCE_FORBIDDEN_CHARACTERS + ): raise LogicalReferenceError( f"logical reference contains a forbidden character: {reference!r}" ) @@ -723,12 +728,14 @@ def _parse_reference(*, reference: str) -> tuple[str, str | None, tuple[str, ... raise LogicalReferenceError( f"logical reference has an unknown or malformed scope: {reference!r}" ) - if ":/" in relative or relative.startswith("/"): + if marker in relative or relative.startswith( + LOGICAL_REFERENCE_ABSOLUTE_PATH_PREFIX + ): raise LogicalReferenceError(f"malformed logical-reference path: {reference!r}") if relative == "": return scope, identifier, () parts = tuple(relative.split("/")) - if any(part in {"", ".", ".."} for part in parts): + if any(part in LOGICAL_REFERENCE_INVALID_PATH_SEGMENTS for part in parts): raise LogicalReferenceError( f"logical-reference path contains an invalid segment: {reference!r}" ) diff --git a/src/loopy_loop/worker.py b/src/loopy_loop/worker.py index 56c0a61..a4d9d71 100644 --- a/src/loopy_loop/worker.py +++ b/src/loopy_loop/worker.py @@ -865,7 +865,9 @@ def _render_prompt( f"- scratch dir (this iteration): {scratch.resolve()} " "(raw/verbose output only; evidence goes in the durable tree)", f"- paths.json: {paths_json_path} " - "full path map, rosters, scheduler view — read if needed", + "full path map, rosters, scheduler view — read if needed; " + f"contracts: {(iteration_dir / CONTRACTS_FILENAME).resolve()} " + "engine-derived artifact contracts", ] criteria: list[str] = [] if config_snapshot.completion_criteria: diff --git a/src/tests/test_contract_descriptors.py b/src/tests/test_contract_descriptors.py index 68f972f..ed321ad 100644 --- a/src/tests/test_contract_descriptors.py +++ b/src/tests/test_contract_descriptors.py @@ -1,14 +1,26 @@ from __future__ import annotations +from pathlib import Path + +import pytest + +from loopy_loop.config import run_preflight from loopy_loop.contract_descriptors import build_contracts_descriptor +from loopy_loop.coordinator_app import _build_workflow_roster from loopy_loop.models import ControlSignal from loopy_loop.models import LayerHandoff from loopy_loop.models import utc_now from loopy_loop.models import WorkflowRoster from loopy_loop.models import WorkflowRosterRole from loopy_loop.models import WorkflowSetContract +from loopy_loop.references import LOGICAL_REFERENCE_ABSOLUTE_PATH_PREFIX +from loopy_loop.references import LOGICAL_REFERENCE_FORBIDDEN_CHARACTERS from loopy_loop.references import LOGICAL_REFERENCE_IMPLICIT_SCOPES +from loopy_loop.references import LOGICAL_REFERENCE_INVALID_PATH_SEGMENTS from loopy_loop.references import LOGICAL_REFERENCE_NAMED_SCOPES +from loopy_loop.references import LOGICAL_REFERENCE_PATH_MARKER + +TEMPLATES_ROOT = Path(__file__).resolve().parents[1] / "loopy_loop" / "templates" def _workflow_contract() -> WorkflowSetContract: @@ -60,20 +72,28 @@ def _workflow_roster(*, expected_outputs: list[str]) -> WorkflowRoster: ) +def _stock_inner_outer_contract_and_roster() -> tuple[ + WorkflowSetContract, WorkflowRoster +]: + preflight = run_preflight(repo_root=TEMPLATES_ROOT / "inner_outer_eval") + roster = _build_workflow_roster( + session_id="session-stock-inner-outer", + preflight=preflight, + created_at=utc_now(), + ) + return preflight.workflow_contract, roster + + def test_descriptor_uses_engine_schema_and_reference_scope_constants() -> None: + contract, roster = _stock_inner_outer_contract_and_roster() descriptor = build_contracts_descriptor( - workflow_contract=_workflow_contract(), - workflow_roster=_workflow_roster( - expected_outputs=["project_state/eval_state.md"] - ), + workflow_contract=contract, workflow_roster=roster ) assert descriptor["layer_handoff"]["json_schema"] == ( LayerHandoff.model_json_schema() ) - assert descriptor["terminal_control"]["json_schema"] == ( - ControlSignal.model_json_schema() - ) + assert "json_schema" not in descriptor["terminal_control"] references = descriptor["logical_references"] assert references["implicit"] == { "shape": ":/", @@ -83,29 +103,54 @@ def test_descriptor_uses_engine_schema_and_reference_scope_constants() -> None: "shape": "::/", "scopes": sorted(LOGICAL_REFERENCE_NAMED_SCOPES), } + assert references["path_rules"] == { + "path_must_be_relative_after_marker": True, + "forbidden_absolute_path_prefix": LOGICAL_REFERENCE_ABSOLUTE_PATH_PREFIX, + "forbidden_embedded_path_marker": LOGICAL_REFERENCE_PATH_MARKER, + "forbidden_characters": sorted(LOGICAL_REFERENCE_FORBIDDEN_CHARACTERS), + "forbidden_segments": sorted(LOGICAL_REFERENCE_INVALID_PATH_SEGMENTS), + } + assert descriptor["terminal_control"]["evidence_refs"] == { + "value_kind": "logical_reference", + "must_resolve_to": "file", + "path_must_be_nonempty": True, + "not_valid_reference_values": ["URL", "git SHA", "absolute filesystem path"], + "grammar_and_path_rules": "#/logical_references", + } -def test_eval_receipt_refs_applicability_follows_frozen_runner_outputs() -> None: - contract = _workflow_contract() - retired = build_contracts_descriptor( - workflow_contract=contract, - workflow_roster=_workflow_roster( - expected_outputs=["project_state/eval_state.md"] - ), +def test_stock_advisory_roster_omits_retired_eval_receipt_refs() -> None: + contract, roster = _stock_inner_outer_contract_and_roster() + assert contract.evaluation.advisory is True + assert contract.check_runner_roles == ["outer"] + outer = next(role for role in roster.roles if role.workflow_id == "outer") + assert "project_state/eval_state.md" in outer.expected_outputs + assert "eval_receipts/" not in outer.expected_outputs + + descriptor = build_contracts_descriptor( + workflow_contract=contract, workflow_roster=roster + ) + terminal_control = descriptor["terminal_control"] + assert terminal_control["eval_receipt_refs"] == { + "applicable": False, + "check_runner_roles": ["outer"], + "receipt_producing_check_runner_roles": [], + } + assert "eval_receipt_refs" not in terminal_control["accepted_fields"] + assert "eval_receipt_ref" not in terminal_control["accepted_fields"] + assert terminal_control["accepted_fields_are_authoritative"] is True + assert set(terminal_control["accepted_field_schemas"]) == set( + terminal_control["accepted_fields"] ) + + +def test_eval_receipt_refs_remain_applicable_to_frozen_receipt_roster() -> None: + contract = _workflow_contract() receipt_producing = build_contracts_descriptor( workflow_contract=contract, workflow_roster=_workflow_roster(expected_outputs=["eval_receipts/"]), ) - retired_control = retired["terminal_control"] - assert retired_control["eval_receipt_refs"] == { - "applicable": False, - "check_runner_roles": ["eval_runner"], - "receipt_producing_check_runner_roles": [], - } - assert "eval_receipt_refs" not in retired_control["accepted_fields"] - receipt_control = receipt_producing["terminal_control"] assert receipt_control["eval_receipt_refs"] == { "applicable": True, @@ -115,3 +160,34 @@ def test_eval_receipt_refs_applicability_follows_frozen_runner_outputs() -> None assert "eval_receipt_refs" in receipt_control["accepted_fields"] assert "eval_receipt_ref" not in receipt_control["accepted_fields"] assert set(receipt_control["accepted_fields"]) <= set(ControlSignal.model_fields) + + +def test_goal_met_example_round_trips_through_real_control_model() -> None: + contract, roster = _stock_inner_outer_contract_and_roster() + terminal_control = build_contracts_descriptor( + workflow_contract=contract, workflow_roster=roster + )["terminal_control"] + + accepted_fields = set(terminal_control["accepted_fields"]) + required_fields = terminal_control["required_fields"] + assert required_fields == ControlSignal.required_field_names( + schema_version=contract.session_protocol_version, + state="stopped", + stop_reason="goal_met", + ) + assert {"control_id", "producer", "created_at"} <= set(required_fields) + assert set(required_fields) <= accepted_fields + + example = terminal_control["goal_met_example"] + assert example is not None + payload = {field_name: example[field_name] for field_name in required_fields} + assert set(payload) == set(required_fields) + assert set(payload) <= accepted_fields + validated = ControlSignal.model_validate(payload) + assert validated.state == "stopped" + assert validated.stop_reason == "goal_met" + + incomplete = dict(payload) + incomplete.pop("control_id") + with pytest.raises(ValueError): + ControlSignal.model_validate(incomplete) diff --git a/src/tests/test_worker.py b/src/tests/test_worker.py index 45c378c..8b81c0c 100644 --- a/src/tests/test_worker.py +++ b/src/tests/test_worker.py @@ -8,9 +8,14 @@ import httpx import pytest +from loopy_loop.models import AttemptAssignment from loopy_loop.models import IterationResult from loopy_loop.models import LayerHandoff from loopy_loop.models import TaskResponse +from loopy_loop.models import utc_now +from loopy_loop.models import WorkflowRoster +from loopy_loop.models import WorkflowRosterRole +from loopy_loop.models import WorkflowSetContract from loopy_loop.sessions import create_session_dir from loopy_loop.sessions import eval_readiness_dir_path from loopy_loop.sessions import pending_finished_request_path @@ -347,6 +352,58 @@ def _render_synthetic_workflow_set( else [], stop_criteria=stop_criteria if stop_criteria is not None else [], ) + workflow_contract = WorkflowSetContract.model_validate( + { + "schema_version": 1, + "session_protocol_version": 3, + "roles": {"outer": {"responsibility": "Orchestrate."}}, + "orchestration": { + "completion_role": "outer", + "plan_owner": "outer", + "handoff_owner": "outer", + "task_acceptance_owner": "outer", + }, + "evaluation": { + "advisory": True, + "check_author_roles": [], + "check_runner_roles": ["outer"], + }, + "terminal_blocker_reporting_roles": ["outer"], + "child_interface": "none", + } + ) + workflow_roster = WorkflowRoster( + session_id=session_id, + workflow_contract_sha256="sha256:" + "a" * 64, + created_at=utc_now(), + completion_role="outer", + roles=[ + WorkflowRosterRole( + workflow_id="outer", + responsibility="Orchestrate.", + cadence={"run_every": 1}, + expected_outputs=[ + "project_state/handoff.json", + "project_state/eval_state.md", + ], + authorities=["completion", "eval_check_runner"], + ) + ], + ) + assignment = AttemptAssignment( + identity={ + "root_session_id": session_id, + "session_id": session_id, + "workflow_id": "outer", + "attempt_id": "attempt-synthetic-26", + }, + actor={}, + objective={}, + absolute_paths={}, + ownership={}, + provenance={}, + context={"workflow_roster": workflow_roster.model_dump(mode="json")}, + ) return _render_prompt( config_snapshot=snapshot, session_id=session_id, @@ -357,6 +414,8 @@ def _render_synthetic_workflow_set( harness_output_root=session_dir / "harness_outputs" / "0026_outer", workflow_prompt="Do the synthetic role work.", repo_root=repo_root, + assignment=assignment, + workflow_contract=workflow_contract, ) @@ -385,12 +444,33 @@ def test_render_header_matches_diet_shape( "- control.json", "- scratch dir (this iteration):", "- paths.json:", + "contracts:", ): assert label in prompt # No shared preamble → no ground-rules section. assert "Shared ground rules:" not in prompt assert prompt.rstrip().endswith("Workflow body:\nDo the synthetic role work.") + contracts = json.loads( + ( + repo_root + / ".loopy_loop" + / "sessions" + / "goal_20260419_143022_ab12cd34" + / "iterations" + / "0026_outer" + / "contracts.json" + ).read_text(encoding="utf-8") + ) + terminal_control = contracts["terminal_control"] + assert terminal_control["active_protocol_version"] == 3 + assert terminal_control["eval_receipt_refs"]["applicable"] is False + assert "eval_receipt_refs" not in terminal_control["accepted_fields"] + assert "eval_receipt_ref" not in terminal_control["accepted_fields"] + assert {"control_id", "producer", "created_at"} <= set( + terminal_control["required_fields"] + ) + def test_render_includes_preamble_when_present( repo_root: Any, snapshot_factory: Any From 685a255de2134749a035238701c3838c08ab9723 Mon Sep 17 00:00:00 2001 From: Jiri Puc Date: Mon, 20 Jul 2026 14:21:47 +0200 Subject: [PATCH 3/3] docs: explain the emitted contract descriptor on success-and-control MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Documents that terminal control and the layer handoff must conform to the engine-emitted contracts file (paths.json -> contracts) rather than a prompt-restated schema: authoritative accepted/required fields + a validated goal_met example, eval-receipt applicability, the evidence-ref grammar, and the LayerHandoff schema — all derived from engine ground truth so they cannot drift. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01EPYwF6cRYs2mfsfHF1hibc --- .../src/app/docs/success-and-control/page.mdx | 29 +++++++++++++++++++ 1 file changed, 29 insertions(+) diff --git a/website/src/app/docs/success-and-control/page.mdx b/website/src/app/docs/success-and-control/page.mdx index 4944161..5fa2818 100644 --- a/website/src/app/docs/success-and-control/page.mdx +++ b/website/src/app/docs/success-and-control/page.mdx @@ -103,6 +103,35 @@ after an iteration finishes. When a workflow writes those two values, it is maki an explicit, auditable, reversible stop decision. That is the point: a human or an agent chose to stop, and you can see exactly why. +### Conform to the emitted contract, not a restated one + +The full `control.json` above is the minimal shape. Terminal control (and the layer +handoff a successor reads) is validated strictly by the engine, and those rules — +which fields are accepted and required for the active protocol, the logical-reference +grammar for evidence refs, and the exact handoff schema — depend on the session's +frozen workflow contract. Rather than have each workflow prompt restate those rules +in prose (where they drift from the validator and produce rejected control), the +engine **emits the real contracts as a file every iteration** and points the prompts +at it. + +Each iteration's assignment lists a `contracts` path (in `paths.json` and the prompt +header). It is a descriptor generated directly from the engine's own models and +constants, so it cannot drift: + +- the accepted and required control fields for the active protocol version, plus a + validated minimal `goal_met` example (the raw multi-version JSON schema is included + only as subordinate reference — the accepted/required lists are authoritative); +- whether eval-receipt references apply at all (they do not under the stock advisory-eval + contract, so citing them is rejected); +- the evidence-reference grammar — logical `:/` refs to files, with the + scopes and path-segment rules the reference parser actually enforces; +- the `LayerHandoff` JSON schema, taken straight from the model the engine validates + against. + +A workflow author writing a control or handoff step should tell the agent to read +that contract file and conform to it exactly, instead of hard-coding a field list in +the prompt. + ## goal_check.json is evidence, not control The other file people expect to stop the loop is `goal_check.json`. It does not.