diff --git a/src/loopy_loop/contract_descriptors.py b/src/loopy_loop/contract_descriptors.py new file mode 100644 index 0000000..8d977e1 --- /dev/null +++ b/src/loopy_loop/contract_descriptors.py @@ -0,0 +1,235 @@ +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_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 + + +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 + ) + 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, + "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), + }, + "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", + "active_protocol_version": protocol_version, + "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, + "receipt_producing_check_runner_roles": receipt_producing_roles, + }, + }, + } + + +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: + """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..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) @@ -685,6 +697,45 @@ 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) + + @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: @@ -702,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( @@ -744,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 f6e19a8..b0128a8 100644 --- a/src/loopy_loop/references.py +++ b/src/loopy_loop/references.py @@ -18,8 +18,12 @@ 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 = ":/" +LOGICAL_REFERENCE_FORBIDDEN_CHARACTERS = frozenset({"\x00", "\\"}) +LOGICAL_REFERENCE_INVALID_PATH_SEGMENTS = frozenset({"", ".", ".."}) +LOGICAL_REFERENCE_ABSOLUTE_PATH_PREFIX = "/" class LogicalReferenceError(ValueError): @@ -697,22 +701,26 @@ 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}" ) - 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") @@ -720,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/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..a4d9d71 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) @@ -858,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: @@ -895,14 +904,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 +940,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 +972,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..ed321ad --- /dev/null +++ b/src/tests/test_contract_descriptors.py @@ -0,0 +1,193 @@ +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: + 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 _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=contract, workflow_roster=roster + ) + + assert descriptor["layer_handoff"]["json_schema"] == ( + LayerHandoff.model_json_schema() + ) + assert "json_schema" not in descriptor["terminal_control"] + references = descriptor["logical_references"] + assert references["implicit"] == { + "shape": ":/", + "scopes": sorted(LOGICAL_REFERENCE_IMPLICIT_SCOPES), + } + assert references["named"] == { + "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_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/"]), + ) + + 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) + + +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_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..8b81c0c 100644 --- a/src/tests/test_worker.py +++ b/src/tests/test_worker.py @@ -8,8 +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 @@ -214,6 +220,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( @@ -340,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, @@ -350,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, ) @@ -378,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 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.