From ea26936a2a82c459c121117f0a5dabd2846488e6 Mon Sep 17 00:00:00 2001 From: Kaley Hayek Date: Wed, 21 Oct 2026 12:11:23 -0400 Subject: [PATCH] feat(runtime): bind task placeholders before dispatch - add one-shot audited post-submit task binding\n- serialize bind and dispatch across controller processes\n- preserve fail-closed task and audit state with focused tests\n\nCloses #45 --- .../prompts/orchestrator/system_prompt.md | 9 +- zenith/src/zenith_harness/controller.py | 239 +++++- zenith/src/zenith_harness/models.py | 55 +- zenith/src/zenith_harness/server.py | 44 +- zenith/src/zenith_harness/storage.py | 157 +++- zenith/tests/test_server.py | 57 ++ zenith/tests/test_task_binding.py | 757 ++++++++++++++++++ 7 files changed, 1298 insertions(+), 20 deletions(-) create mode 100644 zenith/tests/test_task_binding.py diff --git a/zenith/src/zenith_harness/bundled/prompts/orchestrator/system_prompt.md b/zenith/src/zenith_harness/bundled/prompts/orchestrator/system_prompt.md index 22f6589..9a16901 100644 --- a/zenith/src/zenith_harness/bundled/prompts/orchestrator/system_prompt.md +++ b/zenith/src/zenith_harness/bundled/prompts/orchestrator/system_prompt.md @@ -21,13 +21,13 @@ Drive the mission only through the orchestrator runtime tools: - Unknown current state: call `inspect_project(project_id)` before deciding. - No project or `draft`: call `start_project(brief, workspace_dir)`. - `mission_planning`: investigate, load playbook(s), write contract files, author the task list, then call `submit_plan`. -- `mission_running`: call `advance_project` to dispatch runnable work or evaluate ready gates. If it returns still `mission_running` with runnable work, call it again. +- `mission_running`: optionally call `bind_task_placeholders` exactly once to bind post-submit launch evidence into one task before any attempt; then call `advance_project` to dispatch runnable work or evaluate ready gates. If it returns still `mission_running` with runnable work, call it again. - Quiescent `mission_running`: when no runnable work remains and evidence supports closure, call `end_mission`. - `attention_needed`: read every open item and its evidence, decide exactly once per item with `decide_attention`, then call `advance_project`. - User-requested or justified mission cancellation: call `abort_project(project_id, reason)`. - `done`, `failed`, `aborted`: terminal states. Do not continue mission work unless the user starts a new scope or explicitly asks for forensic inspection. -`submit_plan` and `decide_attention` persist state; they do not dispatch work. `advance_project` dispatches workers, validators, merge work, and gate evaluation according to runtime state. `end_mission` requests runtime closure and terminal review; call it only after planning, validation, gates, evidence review, and open attention handling support closure. +`submit_plan`, `bind_task_placeholders`, and `decide_attention` persist state; they do not dispatch work. `bind_task_placeholders` is an orchestrator-only one-shot transition valid only immediately after submit and before any task attempt. `advance_project` dispatches workers, validators, merge work, and gate evaluation according to runtime state. `end_mission` requests runtime closure and terminal review; call it only after planning, validation, gates, evidence review, and open attention handling support closure. Use the returned `projectId`, `projectRoot`, and `harnessRoot`. Do not assume historical project paths or edit runtime cursors directly. @@ -264,7 +264,7 @@ Validators must depend on the work they validate. Do not make a validation task ## Execution, Attention, And Replanning -After `submit_plan`, drive execution through `advance_project`. Do not implement product work yourself, invent manual worker handoffs, or treat local commands as a substitute for runtime workers, validators, gates, or terminal review. +After `submit_plan`, optionally perform the one-shot `bind_task_placeholders` launch-evidence binding, then drive execution through `advance_project`. Do not implement product work yourself, invent manual worker handoffs, or treat local commands as a substitute for runtime workers, validators, gates, or terminal review. The orchestrator may directly fix small mechanical runtime hygiene problems when they block runtime progress and do not require a product decision. This exception is limited to issues such as stale runtime state, trivial conflict markers in orchestrator-owned mission artifacts, path hygiene, or metadata that is clearly mechanical. @@ -357,6 +357,7 @@ Orchestrator tools: - `inspect_project(project_id)`: read-only state, full task-list view, and open attention. Use it when waking without reliable current state. - `start_project(brief, workspace_dir)`: create a project rooted at the provided directory, write the original brief, and enter `mission_planning`. - `submit_plan(project_id, task_list)`: submit the contract-backed task list after contract files exist. It persists state; it does not dispatch work. +- `bind_task_placeholders(project_id, task_id, bindings)`: one-shot post-submit/pre-dispatch binding of explicit placeholder-to-replacement strings in exactly one non-gate task body. Every placeholder must occur exactly once; the complete task list is revalidated and the binding is durably audited without dispatching. Call at most once per mission before any attempt. - `advance_project(project_id, max_steps?)`: drive `mission_running` forward by dispatching runnable workers, validators, and gates. It may block. It does not request closure. - `decide_attention(project_id, decisions)`: resolve every open attention item with exactly one decision, then return to runtime flow. Call `advance_project` afterward. - `end_mission(project_id)`: request runtime closure and terminal review only after work is quiescent and evidence supports closure. @@ -392,7 +393,7 @@ Runtime coverage semantics are strict: every live contract assertion must have e - `supersede`: map old task id to replacement task id. The replacement must already exist or be included in `add`; downstream `depends_on` references are rewritten to the replacement. - `cancel`: retire task ids without replacement; downstream `depends_on` references are dropped. -There is no in-place task edit primitive. To change a task body, skill, targets, or dependencies, supersede the task. Cleared or running tasks cannot be superseded or cancelled. +There is no broad in-place task edit primitive. The sole bounded exception is the one-shot `bind_task_placeholders` transition immediately after `submit_plan`; otherwise, to change a task body, skill, targets, or dependencies, supersede the task. Cleared or running tasks cannot be superseded or cancelled. ## Anti-Patterns diff --git a/zenith/src/zenith_harness/controller.py b/zenith/src/zenith_harness/controller.py index f7aeeeb..784be37 100644 --- a/zenith/src/zenith_harness/controller.py +++ b/zenith/src/zenith_harness/controller.py @@ -1,4 +1,4 @@ -"""ProjectController — routes the 7 orchestrator MCP tools. +"""ProjectController — routes the 8 orchestrator MCP tools. See `specs/task_list/PRODUCT.md`. The controller owns: - envelope construction @@ -10,6 +10,7 @@ """ from __future__ import annotations +import hashlib from dataclasses import dataclass from .config import HarnessConfig @@ -27,11 +28,12 @@ MissionPlanning, MissionRunning, ProjectState, - TaskList, TaskListPatch, + TaskPlaceholderBindingAudit, + TaskList, TaskStateFile, ) -from .storage import ProjectStore +from .storage import ProjectStore, task_list_digest, utc_now_iso from .task_list_patch import apply_patch from .task_validation import ( ValidationError, @@ -105,7 +107,9 @@ def submit_plan(self, project_id: str, task_list: TaskList) -> Envelope: "invalid_task_list", "task list validation failed", details=errs ) - self.store.save_task_list(project_id, mid, task_list) + self.store.save_task_list( + project_id, mid, task_list, preserve_binding_audit=False + ) task_state = TaskStateFile() for task in task_list.tasks: task_state.set_status(task.id, "pending") @@ -117,11 +121,185 @@ def submit_plan(self, project_id: str, task_list: TaskList) -> Envelope: self.store.save_state(project_id, MissionRunning(mission_id=mid)) return self._build_envelope(project_id, dag_mode="frontier") + def bind_task_placeholders( + self, + project_id: str, + task_id: str, + bindings: dict[str, str], + ) -> Envelope: + with self.store.project_lock(project_id): + return self._bind_task_placeholders(project_id, task_id, bindings) + + def _bind_task_placeholders( + self, + project_id: str, + task_id: str, + bindings: dict[str, str], + ) -> Envelope: + """Bind one task's placeholders exactly once before any dispatch.""" + state = self._require_state(project_id) + if not isinstance(state, MissionRunning): + raise ToolError( + "wrong_state", + f"bind_task_placeholders requires MissionRunning; got {state.state}", + ) + mid = state.mission_id + + try: + existing_audit = self.store.load_task_binding_audit(project_id, mid) + except Exception as exc: # noqa: BLE001 + raise ToolError( + "invalid_task_binding_audit", + "task binding audit is invalid", + ) from exc + if existing_audit is not None: + raise ToolError( + "task_binding_already_applied", + "task placeholders already bound for this mission", + ) + + task_list = self.store.load_task_list(project_id, mid) + task_state = self.store.load_task_state(project_id, mid) + task_ids = {task.id for task in task_list.tasks} + if set(task_state.tasks) != task_ids: + raise ToolError( + "invalid_task_state", + "task state must contain exactly one entry per task", + ) + if any( + entry.status != "pending" or entry.last_attempt is not None + for entry in task_state.tasks.values() + ): + raise ToolError( + "invalid_task_state", + "all task states must be pending with no last_attempt", + ) + if self.store.list_attempts(project_id, mid): + raise ToolError( + "attempts_exist", + "task binding requires no existing attempts", + ) + + task = next((candidate for candidate in task_list.tasks if candidate.id == task_id), None) + if task is None: + raise ToolError("unknown_task", f"task {task_id!r} not found") + if task.type == "gate": + raise ToolError("gate_task", f"task {task_id!r} is a gate and cannot be bound") + + if not isinstance(bindings, dict) or not bindings: + raise ToolError("empty_bindings", "bindings must not be empty") + for placeholder, replacement in bindings.items(): + if not isinstance(placeholder, str) or not isinstance(replacement, str): + raise ToolError( + "invalid_binding", + "placeholder and replacement must be strings", + ) + + spans: list[tuple[int, int, str, str]] = [] + for placeholder in sorted(bindings): + replacement = bindings[placeholder] + if not placeholder.strip(): + raise ToolError("empty_placeholder", "placeholder must not be empty") + if not replacement.strip(): + raise ToolError("empty_replacement", "replacement must not be empty") + if placeholder == replacement: + raise ToolError( + "identical_binding", + "placeholder and replacement must differ", + ) + + start = task.body.find(placeholder) + if start < 0: + raise ToolError( + "placeholder_missing", + "placeholder must occur exactly once in task body", + ) + second = task.body.find(placeholder, start + len(placeholder)) + if second >= 0: + raise ToolError( + "ambiguous_placeholder", + "placeholder occurs more than once in task body", + ) + spans.append((start, start + len(placeholder), placeholder, replacement)) + + spans.sort(key=lambda span: (span[0], span[1], span[2])) + for previous, current in zip(spans, spans[1:]): + if current[0] < previous[1]: + raise ToolError( + "ambiguous_placeholder", + "placeholder occurrences overlap", + ) + + body_parts: list[str] = [] + cursor = 0 + for start, end, _, replacement in spans: + body_parts.append(task.body[cursor:start]) + body_parts.append(replacement) + cursor = end + body_parts.append(task.body[cursor:]) + bound_task = task.model_copy(update={"body": "".join(body_parts)}) + bound_task_list = TaskList( + tasks=[ + bound_task if candidate.id == task_id else candidate + for candidate in task_list.tasks + ] + ) + + contract_ids, parse_errs = parse_contract_dir( + self.store.contract_dir(project_id, mid) + ) + if parse_errs: + raise ToolError( + "invalid_contract_dir", + "contract directory invalid", + details=parse_errs, + ) + validation_errs = validate_task_list_submission(contract_ids, bound_task_list) + if validation_errs: + raise ToolError( + "invalid_task_list", + "task list validation failed", + details=validation_errs, + ) + + audit = TaskPlaceholderBindingAudit( + task_id=task_id, + before_task_list_sha256=task_list_digest(task_list), + after_task_list_sha256=task_list_digest(bound_task_list), + bindings={ + placeholder: hashlib.sha256(replacement.encode("utf-8")).hexdigest() + for placeholder, replacement in sorted(bindings.items()) + }, + bound_at=utc_now_iso(), + ) + try: + self.store.save_task_list_with_binding_audit( + project_id, + mid, + bound_task_list, + audit, + ) + except FileExistsError as exc: + raise ToolError( + "task_binding_already_applied", + "task placeholders already bound for this mission", + ) from exc + return self._build_envelope(project_id, dag_mode="frontier") + def advance_project( self, project_id: str, max_steps: int | None = None, ) -> Envelope: + with self.store.project_lock(project_id): + return self._advance_project(project_id, max_steps) + + def _advance_project( + self, + project_id: str, + max_steps: int | None = None, + ) -> Envelope: + self._verify_task_binding_audit(project_id) self.store.sync_workspace_skill_surfaces(project_id) coordinator = MissionCoordinator( self.store, project_id, self.dispatcher, self.terminal_reviewer @@ -138,12 +316,17 @@ def advance_project( return self._build_envelope(project_id, dag_mode="frontier") def end_mission(self, project_id: str) -> Envelope: + with self.store.project_lock(project_id): + return self._end_mission(project_id) + + def _end_mission(self, project_id: str) -> Envelope: state = self._require_state(project_id) if not isinstance(state, MissionRunning): raise ToolError( "wrong_state", f"end_mission requires MissionRunning; got {state.state}", ) + self._verify_task_binding_audit(project_id) coordinator = MissionCoordinator( self.store, project_id, self.dispatcher, self.terminal_reviewer ) @@ -405,6 +588,54 @@ def _require_state(self, project_id: str) -> ProjectState: raise ToolError("not_found", f"project {project_id!r} has no state") return state + def _verify_task_binding_audit(self, project_id: str) -> None: + """Block runtime execution when a one-shot claim is inconsistent.""" + record = self.store.load_project(project_id) + state = self.store.load_state(project_id) + mid = self._current_mission_id(record, state) + if mid is None: + return + try: + audit = self.store.load_task_binding_audit(project_id, mid) + except Exception as exc: # noqa: BLE001 + raise ToolError( + "invalid_task_binding_audit", + "task binding audit is invalid", + ) from exc + if audit is None: + return + try: + task_list = self.store.load_task_list(project_id, mid) + current_digest = task_list_digest(task_list) + task_state = self.store.load_task_state(project_id, mid) + attempts = self.store.list_attempts(project_id, mid) + except Exception as exc: # noqa: BLE001 + raise ToolError( + "invalid_task_binding_audit", + "task binding audit is invalid", + ) from exc + task_ids = {task.id for task in task_list.tasks} + if attempts: + return + if set(task_state.tasks) != task_ids: + raise ToolError( + "invalid_task_state", + "task state must contain exactly one entry per task", + ) + if any( + entry.status != "pending" or entry.last_attempt is not None + for entry in task_state.tasks.values() + ): + raise ToolError( + "invalid_task_state", + "all task states must be pending with no last_attempt", + ) + if current_digest != audit.after_task_list_sha256: + raise ToolError( + "task_binding_digest_mismatch", + "task binding audit does not match current task list", + ) + def _make_pending(): from .models import ContractStateEntry diff --git a/zenith/src/zenith_harness/models.py b/zenith/src/zenith_harness/models.py index 2ca8521..d780e2b 100644 --- a/zenith/src/zenith_harness/models.py +++ b/zenith/src/zenith_harness/models.py @@ -14,9 +14,10 @@ from __future__ import annotations import re +from datetime import datetime from typing import Annotated, Any, Literal, Union -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, field_validator # --------------------------------------------------------------------------- # Identifier conventions @@ -386,6 +387,57 @@ def set_last_attempt(self, task_id: str, spawn_ts: str) -> None: entry.last_attempt = spawn_ts +class TaskPlaceholderBindingAudit(BaseModel): + """Internal one-shot audit embedded in the atomic tasks.json object.""" + + model_config = ConfigDict(extra="forbid") + + version: Literal[1] = 1 + task_id: str + before_task_list_sha256: str + after_task_list_sha256: str + # Placeholder name -> SHA-256 of the replacement; values are never stored. + bindings: dict[str, str] + bound_at: str + + @field_validator("task_id") + @classmethod + def require_task_id(cls, value: str) -> str: + if not value.strip(): + raise ValueError("task_id must not be empty") + return value + + @field_validator("before_task_list_sha256", "after_task_list_sha256") + @classmethod + def require_sha256(cls, value: str) -> str: + if not re.fullmatch(r"[0-9a-f]{64}", value): + raise ValueError("digest must be 64 lowercase hexadecimal characters") + return value + + @field_validator("bindings") + @classmethod + def require_binding_metadata(cls, value: dict[str, str]) -> dict[str, str]: + if not value: + raise ValueError("bindings must not be empty") + if any(not placeholder.strip() for placeholder in value): + raise ValueError("placeholder names must not be empty") + for digest in value.values(): + if not re.fullmatch(r"[0-9a-f]{64}", digest): + raise ValueError("replacement digest must be 64 lowercase hexadecimal characters") + return value + + @field_validator("bound_at") + @classmethod + def require_timestamp(cls, value: str) -> str: + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as exc: + raise ValueError("bound_at must be an ISO-8601 timestamp") from exc + if parsed.tzinfo is None: + raise ValueError("bound_at must include a UTC offset") + return value + + class ContractStateEntry(BaseModel): model_config = ConfigDict(extra="forbid") @@ -464,6 +516,7 @@ class LoadedMarkdownAsset(BaseModel): "TerminalReviewHandoff", "TaskStateEntry", "TaskStateFile", + "TaskPlaceholderBindingAudit", "ContractStateEntry", "ContractStateFile", "ProjectRecord", diff --git a/zenith/src/zenith_harness/server.py b/zenith/src/zenith_harness/server.py index 19c37eb..f80ebee 100644 --- a/zenith/src/zenith_harness/server.py +++ b/zenith/src/zenith_harness/server.py @@ -39,7 +39,7 @@ def create_orchestrator_server( config: HarnessConfig, controller: ProjectController | None = None, ) -> FastMCP: - """7 orchestrator tools, registered on a stdio MCP server.""" + """8 orchestrator tools, registered on a stdio MCP server.""" if controller is None: from .dispatcher import MockDispatcher, MockTerminalReviewer @@ -55,9 +55,10 @@ def create_orchestrator_server( name="zenith", instructions=( "Mission orchestration harness. Mode: orchestrator. " - "7 tools: start_project, submit_plan, advance_project, " + "8 tools: start_project, submit_plan, bind_task_placeholders, advance_project, " "end_mission, decide_attention, inspect_project, abort_project. " - "Lifecycle: plan with submit_plan, run with advance_project, " + "Lifecycle: plan with submit_plan, optionally bind post-submit task placeholders " + "once before dispatch, then run with advance_project, " "request closure with end_mission, resolve attention with decide_attention, " "then call advance_project again." ), @@ -171,6 +172,43 @@ async def submit_plan( except ToolError as exc: return _to_payload(exc) + @mcp.tool( + name="bind_task_placeholders", + description=( + "One-shot orchestrator-only post-submit/pre-dispatch binding. Valid only " + "in mission_running before any task attempt. Bind exactly one task's " + "explicit placeholder-to-replacement mapping; each placeholder must " + "occur exactly once. Revalidates and durably persists the task list " + "without dispatching, syncing the workspace, or changing task/contract " + "state. Call at most once per mission, before advance_project." + ), + ) + async def bind_task_placeholders( + project_id: Annotated[str, Field(description="Project id.")], + task_id: Annotated[str, Field(description="The one task whose body is bound.")], + bindings: Annotated[ + dict[str, str], + Field( + description=( + "Explicit placeholder-to-replacement strings. Every placeholder " + "must occur exactly once in the selected task body." + ) + ), + ], + ) -> dict[str, Any]: + async with await _project_lock(project_id): + try: + return _to_payload( + await asyncio.to_thread( + controller.bind_task_placeholders, + project_id, + task_id, + bindings, + ) + ) + except ToolError as exc: + return _to_payload(exc) + @mcp.tool( name="advance_project", description=( diff --git a/zenith/src/zenith_harness/storage.py b/zenith/src/zenith_harness/storage.py index da1d9c0..e75039b 100644 --- a/zenith/src/zenith_harness/storage.py +++ b/zenith/src/zenith_harness/storage.py @@ -15,13 +15,17 @@ """ from __future__ import annotations +import hashlib import json import os import re import shutil +import time +from contextlib import contextmanager from dataclasses import dataclass from datetime import UTC, datetime from pathlib import Path +from typing import Iterator from .config import HarnessConfig from .models import ( @@ -32,6 +36,7 @@ ProjectRecord, ProjectState, TaskList, + TaskPlaceholderBindingAudit, TaskStateFile, TerminalReviewHandoff, ValidateHandoff, @@ -65,8 +70,14 @@ def atomic_write_text(path: str | Path, content: str) -> None: target = Path(path) target.parent.mkdir(parents=True, exist_ok=True) tmp_path = target.with_suffix(target.suffix + ".tmp") - tmp_path.write_text(content, encoding="utf-8") - os.replace(tmp_path, target) + try: + tmp_path.write_text(content, encoding="utf-8") + os.replace(tmp_path, target) + finally: + try: + tmp_path.unlink() + except FileNotFoundError: + pass def atomic_write_json(path: str | Path, payload: object) -> None: @@ -75,6 +86,57 @@ def atomic_write_json(path: str | Path, payload: object) -> None: ) +def task_list_digest(task_list: TaskList) -> str: + """Return a stable SHA-256 digest of a canonical task-list payload.""" + canonical = json.dumps( + task_list.model_dump(mode="json"), + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +@contextmanager +def _advisory_file_lock(path: Path) -> Iterator[None]: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("a+b") as stream: + if os.name == "nt": + import msvcrt + + locking = getattr(msvcrt, "locking") + lock_nonblocking = getattr(msvcrt, "LK_NBLCK") + stream.seek(0, os.SEEK_END) + if stream.tell() == 0: + stream.write(b"\0") + stream.flush() + stream.seek(0) + while True: + try: + locking(stream.fileno(), lock_nonblocking, 1) + break + except OSError: + time.sleep(0.05) + else: + import fcntl + + fcntl.flock(stream.fileno(), fcntl.LOCK_EX) + try: + yield + finally: + if os.name == "nt": + import msvcrt + + locking = getattr(msvcrt, "locking") + lock_unlock = getattr(msvcrt, "LK_UNLCK") + stream.seek(0) + locking(stream.fileno(), lock_unlock, 1) + else: + import fcntl + + fcntl.flock(stream.fileno(), fcntl.LOCK_UN) + + def attempt_to_markdown(handoff: WorkHandoff | ValidateHandoff) -> str: lines = ["---"] lines.append(f"node_id: {handoff.node_id}") @@ -157,6 +219,15 @@ def mission_runtime_dir(self, project_id: str, mission_id: str) -> Path: attempts/*.json).""" return self.zenith_runtime_dir(project_id) / "missions" / mission_id + def project_lock_path(self, project_id: str) -> Path: + return self.zenith_runtime_dir(project_id) / "controller.lock" + + @contextmanager + def project_lock(self, project_id: str) -> Iterator[None]: + """Acquire the per-project advisory lock for the full controller operation.""" + with _advisory_file_lock(self.project_lock_path(project_id)): + yield + # ------------------------------------------------------------------ # Project lifecycle # ------------------------------------------------------------------ @@ -311,18 +382,87 @@ def list_missions(self, project_id: str) -> list[str]: return [] return sorted(p.name for p in d.iterdir() if p.is_dir()) - def load_task_list(self, project_id: str, mission_id: str) -> TaskList: - path = self.mission_runtime_dir(project_id, mission_id) / "tasks.json" + def task_list_path(self, project_id: str, mission_id: str) -> Path: + return self.mission_runtime_dir(project_id, mission_id) / "tasks.json" + + def _load_task_file( + self, project_id: str, mission_id: str + ) -> tuple[TaskList, TaskPlaceholderBindingAudit | None]: + path = self.task_list_path(project_id, mission_id) if not path.exists(): raise FileNotFoundError(f"tasks.json missing for {project_id}/{mission_id}") - return TaskList.model_validate_json(path.read_text(encoding="utf-8")) + raw = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(raw, dict): + raise ValueError("tasks.json must contain an object") + public = dict(raw) + audit_raw = public.pop("_binding_audit", None) + audit = ( + None + if audit_raw is None + else TaskPlaceholderBindingAudit.model_validate(audit_raw) + ) + return TaskList.model_validate(public), audit - def save_task_list(self, project_id: str, mission_id: str, tl: TaskList) -> None: + def load_task_list(self, project_id: str, mission_id: str) -> TaskList: + task_list, _ = self._load_task_file(project_id, mission_id) + return task_list + + def _write_task_file( + self, + project_id: str, + mission_id: str, + task_list: TaskList, + audit: TaskPlaceholderBindingAudit | None, + ) -> None: + payload = task_list.model_dump(mode="json") + if audit is not None: + payload["_binding_audit"] = audit.model_dump(mode="json") atomic_write_json( - self.mission_runtime_dir(project_id, mission_id) / "tasks.json", - tl.model_dump(mode="json"), + self.task_list_path(project_id, mission_id), + payload, ) + def save_task_list( + self, + project_id: str, + mission_id: str, + tl: TaskList, + *, + preserve_binding_audit: bool = True, + ) -> None: + audit = None + path = self.task_list_path(project_id, mission_id) + if preserve_binding_audit and path.exists(): + _, audit = self._load_task_file(project_id, mission_id) + self._write_task_file(project_id, mission_id, tl, audit) + + def task_binding_audit_path(self, project_id: str, mission_id: str) -> Path: + """Compatibility accessor for the audit-bearing tasks.json file.""" + return self.task_list_path(project_id, mission_id) + + def load_task_binding_audit( + self, project_id: str, mission_id: str + ) -> TaskPlaceholderBindingAudit | None: + path = self.task_list_path(project_id, mission_id) + if not path.exists(): + return None + _, audit = self._load_task_file(project_id, mission_id) + return audit + + def save_task_list_with_binding_audit( + self, + project_id: str, + mission_id: str, + task_list: TaskList, + audit: TaskPlaceholderBindingAudit, + ) -> None: + """Atomically replace tasks.json with public tasks plus one audit object.""" + validated_audit = TaskPlaceholderBindingAudit.model_validate(audit) + existing = self.load_task_binding_audit(project_id, mission_id) + if existing is not None: + raise FileExistsError("task placeholders already bound for this mission") + self._write_task_file(project_id, mission_id, task_list, validated_audit) + def contract_dir(self, project_id: str, mission_id: str) -> Path: return self.mission_dir(project_id, mission_id) / "contract" @@ -806,6 +946,7 @@ def _copy_missing_tree(source: Path, target: Path) -> None: "slugify", "atomic_write_text", "atomic_write_json", + "task_list_digest", "attempt_to_markdown", "terminal_review_to_markdown", ] diff --git a/zenith/tests/test_server.py b/zenith/tests/test_server.py index 4b5533c..479ee07 100644 --- a/zenith/tests/test_server.py +++ b/zenith/tests/test_server.py @@ -62,6 +62,7 @@ async def test_orchestrator_tools_registered(config: HarnessConfig) -> None: assert names == { "start_project", "submit_plan", + "bind_task_placeholders", "advance_project", "end_mission", "decide_attention", @@ -82,6 +83,62 @@ async def test_terminal_reviewer_tool_isolated() -> None: assert await _tool_names(server) == {"submit_terminal_review"} +@pytest.mark.asyncio +async def test_bind_task_placeholders_is_orchestrator_only( + config: HarnessConfig, workspace: Path +) -> None: + controller = ProjectController( + config, + MockDispatcher( + lambda request: WorkHandoff(node_id=request.task.id, done=True, report="unused") + ), + MockTerminalReviewer(TerminalReviewHandoff(done=True, report="")), + ) + orchestrator = create_orchestrator_server(config, controller) + + await orchestrator.call_tool( + "start_project", + {"brief": "Bind a receipt.", "workspace_dir": str(workspace)}, + ) + project_id = ProjectStore(config).list_projects()[0].id + contract_dir = controller.store.ensure_contract_dir(project_id, "mission-001") + (contract_dir / "VAL-001.md").write_text("# VAL-001\n") + await orchestrator.call_tool( + "submit_plan", + { + "project_id": project_id, + "task_list": { + "tasks": [ + { + "id": "w1", + "type": "work", + "body": "consume {{RECEIPT}}", + "targets": ["VAL-001"], + "skill": "worker", + "depends_on": [], + } + ] + }, + }, + ) + + await orchestrator.call_tool( + "bind_task_placeholders", + { + "project_id": project_id, + "task_id": "w1", + "bindings": {"{{RECEIPT}}": "receipt.json"}, + }, + ) + + task = controller.store.load_task_list(project_id, "mission-001").tasks[0] + assert task.body == "consume receipt.json" + assert "bind_task_placeholders" not in await _tool_names(create_worker_server()) + assert "bind_task_placeholders" not in await _tool_names( + create_terminal_reviewer_server() + ) + + # --------------------------------------------------------------------------- # end_node writes to ZENITH_HANDOFF_PATH # --------------------------------------------------------------------------- diff --git a/zenith/tests/test_task_binding.py b/zenith/tests/test_task_binding.py new file mode 100644 index 0000000..5cc3d64 --- /dev/null +++ b/zenith/tests/test_task_binding.py @@ -0,0 +1,757 @@ +"""Controller tests for one-shot post-submit task placeholder binding.""" +from __future__ import annotations + +import hashlib +import json +import multiprocessing as mp +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +import pytest + +from zenith_harness.config import HarnessConfig +from zenith_harness.controller import ProjectController, ToolError +from zenith_harness.dispatcher import ( + MockDispatcher, + MockTerminalReviewer, +) +from zenith_harness.models import ( + Decision, + MissionRunning, + Task, + TaskList, + TaskListPatch, + TerminalReviewHandoff, + ValidateHandoff, + ValidationItem, + WorkHandoff, +) +from zenith_harness import storage as storage_module + + +@pytest.fixture +def config(harness_home: Path) -> HarnessConfig: + bundled = Path(__file__).resolve().parents[1] / "src" / "zenith_harness" / "bundled" + return HarnessConfig( + bundled_dir=bundled, + harness_home=harness_home, + projects_dir=harness_home / "projects", + orchestrator_provider_name="claude", + worker_provider_name="claude", + worker_acp_command=None, + validator_provider_name=None, + validator_acp_command=None, + terminal_reviewer_provider_name=None, + terminal_reviewer_acp_command=None, + max_parallel_nodes=1, + ) + + +def _task( + task_id: str, + task_type: str, + targets: list[str], + *, + body: str, + skill: str | None = "worker", + depends_on: list[str] | None = None, +) -> Task: + return Task( + id=task_id, + type=task_type, # type: ignore[arg-type] + body=body, + targets=targets, + skill=None if task_type == "gate" else skill, + depends_on=depends_on or [], + ) + + +def _controller(config: HarnessConfig) -> ProjectController: + return ProjectController( + config, + MockDispatcher( + lambda request: WorkHandoff(node_id=request.task.id, done=True, report="unused") + ), + MockTerminalReviewer(TerminalReviewHandoff(done=True, report="")), + ) + + +def _seed_project( + controller: ProjectController, + workspace: Path, + *, + task_body: str = "launch={{RECEIPT}} path={{PATH}}", +) -> str: + controller.start_project("Bind a launch receipt.", str(workspace)) + project_id = controller.store.list_projects()[0].id + contract_dir = controller.store.ensure_contract_dir(project_id, "mission-001") + (contract_dir / "VAL-001.md").write_text("# VAL-001\n\nReceipt is consumed.\n") + controller.submit_plan( + project_id, + TaskList( + tasks=[ + _task( + "w1", + "work", + ["VAL-001"], + body=task_body, + skill="worker", + ), + _task( + "v1", + "validate", + ["VAL-001"], + body="audit receipt consumption", + skill="auditor", + depends_on=["w1"], + ), + _task( + "g1", + "gate", + ["VAL-001"], + body="", + skill=None, + depends_on=["v1"], + ), + ] + ), + ) + return project_id + + +def _task_list_digest(task_list: TaskList) -> str: + canonical = json.dumps( + task_list.model_dump(mode="json"), + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +def _audit_path(controller: ProjectController, project_id: str) -> Path: + return controller.store.mission_runtime_dir(project_id, "mission-001") / "tasks.json" + + +def _has_audit(controller: ProjectController, project_id: str) -> bool: + path = _audit_path(controller, project_id) + if not path.exists(): + return False + return "_binding_audit" in json.loads(path.read_text(encoding="utf-8")) + + +def test_bind_task_placeholders_replaces_only_one_task_and_records_safe_audit( + config: HarnessConfig, workspace: Path +) -> None: + controller = _controller(config) + project_id = _seed_project(controller, workspace) + store = controller.store + before = store.load_task_list(project_id, "mission-001") + before_state = store.load_task_state(project_id, "mission-001") + before_contract_state = store.load_contract_state(project_id, "mission-001") + + controller.bind_task_placeholders( + project_id, + "w1", + {"{{RECEIPT}}": "/tmp/launch-receipt.json", "{{PATH}}": "receipt.json"}, + ) + + after = store.load_task_list(project_id, "mission-001") + target_before = next(task for task in before.tasks if task.id == "w1") + target_after = next(task for task in after.tasks if task.id == "w1") + assert target_after.body == "launch=/tmp/launch-receipt.json path=receipt.json" + assert target_after.model_dump(exclude={"body"}) == target_before.model_dump( + exclude={"body"} + ) + assert [task for task in after.tasks if task.id != "w1"] == [ + task for task in before.tasks if task.id != "w1" + ] + assert store.load_task_state(project_id, "mission-001") == before_state + assert store.load_contract_state(project_id, "mission-001") == before_contract_state + assert store.list_attempts(project_id, "mission-001") == [] + + audit_path = _audit_path(controller, project_id) + assert audit_path.exists() + audit_text = audit_path.read_text(encoding="utf-8") + audit_payload = json.loads(audit_text) + assert "_binding_audit" in audit_payload + audit = audit_payload["_binding_audit"] + assert audit["task_id"] == "w1" + assert audit["before_task_list_sha256"] == _task_list_digest(before) + assert audit["after_task_list_sha256"] == _task_list_digest(after) + assert audit["bindings"] == { + "{{PATH}}": hashlib.sha256(b"receipt.json").hexdigest(), + "{{RECEIPT}}": hashlib.sha256(b"/tmp/launch-receipt.json").hexdigest(), + } + audit_metadata_text = json.dumps(audit) + assert "/tmp/launch-receipt.json" not in audit_metadata_text + assert "receipt.json" not in audit_metadata_text + + +def test_ordinary_task_list_save_preserves_internal_audit( + config: HarnessConfig, workspace: Path +) -> None: + controller = _controller(config) + project_id = _seed_project(controller, workspace) + controller.bind_task_placeholders(project_id, "w1", {"{{RECEIPT}}": "bound"}) + audit_before = controller.store.load_task_binding_audit(project_id, "mission-001") + task_list = controller.store.load_task_list(project_id, "mission-001") + + controller.store.save_task_list(project_id, "mission-001", task_list) + + assert controller.store.load_task_binding_audit(project_id, "mission-001") == audit_before + + +def test_atomic_task_file_failure_keeps_complete_old_json( + config: HarnessConfig, workspace: Path, monkeypatch +) -> None: + controller = _controller(config) + project_id = _seed_project(controller, workspace) + path = _audit_path(controller, project_id) + before_bytes = path.read_bytes() + before = controller.store.load_task_list(project_id, "mission-001") + + monkeypatch.setattr( + storage_module.os, + "replace", + lambda source, target: (_ for _ in ()).throw(OSError("injected replace failure")), + ) + with pytest.raises(OSError, match="injected replace failure"): + controller.bind_task_placeholders(project_id, "w1", {"{{RECEIPT}}": "bound"}) + + assert path.read_bytes() == before_bytes + assert controller.store.load_task_list(project_id, "mission-001") == before + assert controller.store.load_task_binding_audit(project_id, "mission-001") is None + + +def test_valid_json_with_malformed_audit_is_rejected( + config: HarnessConfig, workspace: Path +) -> None: + controller = _controller(config) + project_id = _seed_project(controller, workspace) + path = _audit_path(controller, project_id) + raw = json.loads(path.read_text(encoding="utf-8")) + raw["_binding_audit"] = { + "version": 1, + "task_id": "w1", + "before_task_list_sha256": "not-a-digest", + "after_task_list_sha256": "0" * 64, + "bindings": {"{{RECEIPT}}": "0" * 64}, + "bound_at": "2026-01-01T00:00:00Z", + } + path.write_text(json.dumps(raw), encoding="utf-8") + + with pytest.raises(ToolError) as exc: + controller.advance_project(project_id) + + assert exc.value.code == "invalid_task_binding_audit" + + +def test_bind_task_placeholders_rejects_repeat_without_touching_task_list( + config: HarnessConfig, workspace: Path +) -> None: + controller = _controller(config) + project_id = _seed_project(controller, workspace) + controller.bind_task_placeholders(project_id, "w1", {"{{RECEIPT}}": "first"}) + store = controller.store + before_tasks = store.load_task_list(project_id, "mission-001") + before_audit = _audit_path(controller, project_id).read_bytes() + + with pytest.raises(ToolError) as exc: + controller.bind_task_placeholders(project_id, "w1", {"{{PATH}}": "second"}) + + assert exc.value.code == "task_binding_already_applied" + assert exc.value.message == "task placeholders already bound for this mission" + assert store.load_task_list(project_id, "mission-001") == before_tasks + assert _audit_path(controller, project_id).read_bytes() == before_audit + + +@pytest.mark.parametrize( + ("bindings", "code", "message"), + [ + ({}, "empty_bindings", "bindings must not be empty"), + ({"": "value"}, "empty_placeholder", "placeholder must not be empty"), + ({"{{TOKEN}}": ""}, "empty_replacement", "replacement must not be empty"), + ( + {"{{TOKEN}}": "{{TOKEN}}"}, + "identical_binding", + "placeholder and replacement must differ", + ), + ( + {"{{MISSING}}": "value"}, + "placeholder_missing", + "placeholder must occur exactly once in task body", + ), + ( + {"{{TOKEN}}": "value"}, + "ambiguous_placeholder", + "placeholder occurs more than once in task body", + ), + ], +) +def test_bind_task_placeholders_rejects_invalid_binding_without_partial_write( + config: HarnessConfig, + workspace: Path, + bindings: dict[str, str], + code: str, + message: str, +) -> None: + controller = _controller(config) + project_id = _seed_project(controller, workspace, task_body="{{TOKEN}} {{TOKEN}}") + store = controller.store + before = store.load_task_list(project_id, "mission-001") + + with pytest.raises(ToolError) as exc: + controller.bind_task_placeholders(project_id, "w1", bindings) + + assert exc.value.code == code + assert exc.value.message == message + assert store.load_task_list(project_id, "mission-001") == before + assert not _has_audit(controller, project_id) + + +def test_bind_task_placeholders_rejects_unknown_and_gate_tasks( + config: HarnessConfig, workspace: Path +) -> None: + controller = _controller(config) + project_id = _seed_project(controller, workspace) + + with pytest.raises(ToolError) as unknown: + controller.bind_task_placeholders(project_id, "unknown", {"{{RECEIPT}}": "x"}) + assert unknown.value.code == "unknown_task" + + with pytest.raises(ToolError) as gate: + controller.bind_task_placeholders(project_id, "g1", {"{{RECEIPT}}": "x"}) + assert gate.value.code == "gate_task" + + assert not _has_audit(controller, project_id) + + +def test_bind_task_placeholders_requires_running_state( + config: HarnessConfig, workspace: Path +) -> None: + controller = _controller(config) + controller.start_project("Still planning.", str(workspace)) + project_id = controller.store.list_projects()[0].id + + with pytest.raises(ToolError) as exc: + controller.bind_task_placeholders(project_id, "w1", {"{{TOKEN}}": "x"}) + + assert exc.value.code == "wrong_state" + assert exc.value.message == "bind_task_placeholders requires MissionRunning; got mission_planning" + + +def test_bind_task_placeholders_rejects_any_task_state_or_attempt( + config: HarnessConfig, workspace: Path +) -> None: + controller = _controller(config) + project_id = _seed_project(controller, workspace) + store = controller.store + + task_state = store.load_task_state(project_id, "mission-001") + task_state.set_status("v1", "cleared") + store.save_task_state(project_id, "mission-001", task_state) + with pytest.raises(ToolError) as non_pending: + controller.bind_task_placeholders(project_id, "w1", {"{{RECEIPT}}": "x"}) + assert non_pending.value.code == "invalid_task_state" + assert not _has_audit(controller, project_id) + + task_state.set_status("v1", "pending") + task_state.set_last_attempt("v1", "2026-01-01T00-00-00Z") + store.save_task_state(project_id, "mission-001", task_state) + with pytest.raises(ToolError) as last_attempt: + controller.bind_task_placeholders(project_id, "w1", {"{{RECEIPT}}": "x"}) + assert last_attempt.value.code == "invalid_task_state" + assert not _has_audit(controller, project_id) + + task_state.tasks["v1"].last_attempt = None + store.save_task_state(project_id, "mission-001", task_state) + store.save_attempt( + project_id, + "mission-001", + "2026-01-01T00-00-00Z", + "v1", + WorkHandoff(node_id="v1", done=True, report="already attempted"), + ) + with pytest.raises(ToolError) as existing_attempt: + controller.bind_task_placeholders(project_id, "w1", {"{{RECEIPT}}": "x"}) + assert existing_attempt.value.code == "attempts_exist" + assert not _has_audit(controller, project_id) + + +def test_bind_task_placeholders_requires_exact_task_state_entries( + config: HarnessConfig, workspace: Path +) -> None: + controller = _controller(config) + project_id = _seed_project(controller, workspace) + store = controller.store + task_state = store.load_task_state(project_id, "mission-001") + task_state.tasks.pop("v1") + store.save_task_state(project_id, "mission-001", task_state) + + with pytest.raises(ToolError) as missing: + controller.bind_task_placeholders(project_id, "w1", {"{{RECEIPT}}": "x"}) + assert missing.value.code == "invalid_task_state" + + task_state.set_status("v1", "pending") + task_state.set_status("extra", "pending") + store.save_task_state(project_id, "mission-001", task_state) + with pytest.raises(ToolError) as extra: + controller.bind_task_placeholders(project_id, "w1", {"{{RECEIPT}}": "x"}) + assert extra.value.code == "invalid_task_state" + assert not _has_audit(controller, project_id) + + +def test_advance_rejects_marker_with_old_task_list_before_sync_or_dispatch( + config: HarnessConfig, workspace: Path, monkeypatch +) -> None: + dispatches: list[str] = [] + + def responder(request): + dispatches.append(request.task.id) + return WorkHandoff(node_id=request.task.id, done=True, report="unexpected") + + controller = ProjectController( + config, + MockDispatcher(responder), + MockTerminalReviewer(TerminalReviewHandoff(done=True, report="")), + ) + project_id = _seed_project(controller, workspace) + original = controller.store.load_task_list(project_id, "mission-001") + controller.bind_task_placeholders(project_id, "w1", {"{{RECEIPT}}": "bound"}) + controller.store.save_task_list(project_id, "mission-001", original) + monkeypatch.setattr( + controller.store, + "sync_workspace_skill_surfaces", + lambda project: pytest.fail("workspace sync must not run"), + ) + + with pytest.raises(ToolError) as exc: + controller.advance_project(project_id) + + assert exc.value.code == "task_binding_digest_mismatch" + assert exc.value.message == "task binding audit does not match current task list" + assert dispatches == [] + + +def test_advance_rejects_audit_with_task_state_or_task_list_inconsistency( + config: HarnessConfig, workspace: Path, monkeypatch +) -> None: + dispatches: list[str] = [] + controller = ProjectController( + config, + MockDispatcher( + lambda request: dispatches.append(request.task.id) + or WorkHandoff(node_id=request.task.id, done=True, report="unexpected") + ), + MockTerminalReviewer(TerminalReviewHandoff(done=True, report="")), + ) + project_id = _seed_project(controller, workspace) + controller.bind_task_placeholders(project_id, "w1", {"{{RECEIPT}}": "bound"}) + path = _audit_path(controller, project_id) + raw = json.loads(path.read_text(encoding="utf-8")) + raw["tasks"][0]["id"] = "renamed-before-attempt" + path.write_text(json.dumps(raw), encoding="utf-8") + monkeypatch.setattr( + controller.store, + "sync_workspace_skill_surfaces", + lambda project: pytest.fail("workspace sync must not run"), + ) + + with pytest.raises(ToolError) as exc: + controller.advance_project(project_id) + + assert exc.value.code == "invalid_task_state" + assert dispatches == [] + + path.write_text(json.dumps(raw), encoding="utf-8") + task_state_path = controller.store.mission_runtime_dir(project_id, "mission-001") / "task-state.json" + task_state_path.unlink() + with pytest.raises(ToolError) as missing_state: + controller.advance_project(project_id) + assert missing_state.value.code == "invalid_task_state" + + +def test_advance_rejects_corrupt_binding_audit_before_sync_or_dispatch( + config: HarnessConfig, workspace: Path, monkeypatch +) -> None: + dispatches: list[str] = [] + controller = ProjectController( + config, + MockDispatcher( + lambda request: dispatches.append(request.task.id) + or WorkHandoff(node_id=request.task.id, done=True, report="unexpected") + ), + MockTerminalReviewer(TerminalReviewHandoff(done=True, report="")), + ) + project_id = _seed_project(controller, workspace) + _audit_path(controller, project_id).write_text("not-json", encoding="utf-8") + monkeypatch.setattr( + controller.store, + "sync_workspace_skill_surfaces", + lambda project: pytest.fail("workspace sync must not run"), + ) + + with pytest.raises(ToolError) as exc: + controller.advance_project(project_id) + + assert exc.value.code == "invalid_task_binding_audit" + assert exc.value.message == "task binding audit is invalid" + assert dispatches == [] + + +def _bind_result(controller: ProjectController, project_id: str) -> str: + try: + controller.bind_task_placeholders(project_id, "w1", {"{{RECEIPT}}": "winner"}) + except ToolError as exc: + return exc.code + return "success" + + +def test_concurrent_binding_has_one_exclusive_winner( + config: HarnessConfig, workspace: Path +) -> None: + seed = _controller(config) + project_id = _seed_project(seed, workspace) + controllers = [_controller(config), _controller(config)] + + with ThreadPoolExecutor(max_workers=2) as pool: + outcomes = list( + pool.map(lambda controller: _bind_result(controller, project_id), controllers) + ) + + assert sorted(outcomes) == ["success", "task_binding_already_applied"] + audit = json.loads(_audit_path(seed, project_id).read_text(encoding="utf-8"))[ + "_binding_audit" + ] + current = seed.store.load_task_list(project_id, "mission-001") + assert audit["after_task_list_sha256"] == _task_list_digest(current) + + +def test_binding_audit_survives_failed_task_patch_recovery( + config: HarnessConfig, workspace: Path +) -> None: + def responder(request): + if request.task.type == "work" and request.task.id == "w1": + return WorkHandoff(node_id="w1", done=False, report="blocked") + if request.task.type == "validate": + return ValidateHandoff( + node_id=request.task.id, + done=True, + report="audited", + items=[ValidationItem(item_id="VAL-001", passed=True)], + passed=True, + ) + return WorkHandoff(node_id=request.task.id, done=True, report="ok") + + controller = ProjectController( + config, + MockDispatcher(responder), + MockTerminalReviewer(TerminalReviewHandoff(done=True, report="")), + ) + project_id = _seed_project(controller, workspace) + controller.bind_task_placeholders(project_id, "w1", {"{{RECEIPT}}": "bound"}) + audit_before = controller.store.load_task_binding_audit(project_id, "mission-001") + controller.advance_project(project_id) + attention = controller.store.load_attention(project_id) + assert attention and attention[0].kind == "node_failed" + + patch = TaskListPatch( + add=[ + _task( + "w2", + "work", + ["VAL-001"], + body="recovered", + skill="worker", + ) + ], + supersede={"w1": "w2"}, + ) + controller.decide_attention( + project_id, + [Decision(item_id=attention[0].id, action="patch", patch=patch)], + ) + controller.advance_project(project_id, max_steps=1) + + assert controller.store.load_task_binding_audit(project_id, "mission-001") == audit_before + + +def test_bind_task_placeholders_revalidates_complete_task_list_before_write( + config: HarnessConfig, workspace: Path +) -> None: + controller = _controller(config) + project_id = _seed_project(controller, workspace) + store = controller.store + current = store.load_task_list(project_id, "mission-001") + invalid = TaskList( + tasks=[ + *current.tasks, + _task( + "w2", + "work", + ["VAL-001"], + body="duplicate owner", + skill="worker", + ), + ] + ) + store.save_task_list(project_id, "mission-001", invalid) + task_state = store.load_task_state(project_id, "mission-001") + task_state.set_status("w2", "pending") + store.save_task_state(project_id, "mission-001", task_state) + + with pytest.raises(ToolError) as exc: + controller.bind_task_placeholders(project_id, "w1", {"{{RECEIPT}}": "x"}) + + assert exc.value.code == "invalid_task_list" + assert any( + detail.code == "over_covered_assertion" for detail in (exc.value.details or []) + ) + assert store.load_task_list(project_id, "mission-001") == invalid + assert not _has_audit(controller, project_id) + + +def test_bind_task_placeholders_rejects_overlapping_occurrences( + config: HarnessConfig, workspace: Path +) -> None: + controller = _controller(config) + project_id = _seed_project(controller, workspace, task_body="{{TOKEN}}") + + with pytest.raises(ToolError) as exc: + controller.bind_task_placeholders( + project_id, + "w1", + {"{{TOKEN}}": "one", "{{TOKEN": "two"}, + ) + + assert exc.value.code == "ambiguous_placeholder" + assert exc.value.message == "placeholder occurrences overlap" + assert not _has_audit(controller, project_id) + + +def test_binding_does_not_change_project_state( + config: HarnessConfig, workspace: Path +) -> None: + controller = _controller(config) + project_id = _seed_project(controller, workspace) + controller.bind_task_placeholders(project_id, "w1", {"{{RECEIPT}}": "x"}) + + state = controller.store.load_state(project_id) + assert isinstance(state, MissionRunning) + assert state.mission_id == "mission-001" + + +class _ProcessBodyDispatcher: + def __init__(self, queue) -> None: + self.queue = queue + + def dispatch(self, request): + self.queue.put(("body", request.task.body)) + return WorkHandoff(node_id=request.task.id, done=True, report="ok") + + def dispatch_batch(self, requests): + return [self.dispatch(request) for request in requests] + + +def _mp_bind_worker(config, project_id: str, start, queue) -> None: + controller = _controller(config) + start.wait() + try: + controller.bind_task_placeholders(project_id, "w1", {"{{RECEIPT}}": "winner"}) + except ToolError as exc: + queue.put(("result", exc.code)) + else: + queue.put(("result", "success")) + + +def _mp_bind_holding_lock_worker(config, project_id: str, locked, start, queue) -> None: + controller = _controller(config) + try: + with controller.store.project_lock(project_id): + locked.set() + start.wait() + controller._bind_task_placeholders( + project_id, "w1", {"{{RECEIPT}}": "bound"} + ) + queue.put(("bind", "success")) + except ToolError as exc: + queue.put(("bind", exc.code)) + + +def _mp_advance_worker(config, project_id: str, start, queue) -> None: + controller = ProjectController( + config, + _ProcessBodyDispatcher(queue), + MockTerminalReviewer(TerminalReviewHandoff(done=True, report="")), + ) + start.wait() + try: + controller.advance_project(project_id, max_steps=1) + except ToolError as exc: + queue.put(("advance", exc.code)) + else: + queue.put(("advance", "success")) + + +def _mp_context(): + methods = mp.get_all_start_methods() + return mp.get_context("fork" if "fork" in methods else "spawn") + + +def test_multiprocess_binders_have_one_winner(config: HarnessConfig, workspace: Path) -> None: + seed = _controller(config) + project_id = _seed_project(seed, workspace) + ctx = _mp_context() + start = ctx.Event() + queue = ctx.Queue() + processes = [ + ctx.Process(target=_mp_bind_worker, args=(config, project_id, start, queue)) + for _ in range(2) + ] + for process in processes: + process.start() + start.set() + for process in processes: + process.join(timeout=15) + assert process.exitcode == 0 + + outcomes = [queue.get(timeout=5)[1] for _ in processes] + assert sorted(outcomes) == ["success", "task_binding_already_applied"] + audit = seed.store.load_task_binding_audit(project_id, "mission-001") + assert audit is not None + assert audit.after_task_list_sha256 == _task_list_digest( + seed.store.load_task_list(project_id, "mission-001") + ) + + +def test_multiprocess_bind_holds_lock_before_advance( + config: HarnessConfig, workspace: Path +) -> None: + seed = _controller(config) + project_id = _seed_project(seed, workspace) + ctx = _mp_context() + start = ctx.Event() + locked = ctx.Event() + queue = ctx.Queue() + binder = ctx.Process( + target=_mp_bind_holding_lock_worker, + args=(config, project_id, locked, start, queue), + ) + advance = ctx.Process( + target=_mp_advance_worker, + args=(config, project_id, start, queue), + ) + binder.start() + advance.start() + assert locked.wait(timeout=10) + start.set() + binder.join(timeout=15) + advance.join(timeout=15) + assert binder.exitcode == 0 + assert advance.exitcode == 0 + + messages = [queue.get(timeout=5) for _ in range(2)] + assert ("bind", "success") in messages + bodies = [value for kind, value in messages if kind == "body"] + assert bodies == ["launch=bound path={{PATH}}"]