"""Worker dispatch plumbing shared by both lead dispatchers. `dispatch_core` (claude / external lead) and `codex_dispatch` (codex lead) differ in capability — only the former has cmux surfaces, blocking waits, and retry from a persisted record. What they do *not* differ in is the job value object and the run's state files: the same run-manifest keys, the same team-state document, the same wrapper argv. Those used to be two copies that had already drifted apart: one dispatcher wrapped a truncated team-state in `DispatchError` while the other let `json.JSONDecodeError` escape, one raised on an unknown `workerId` while the other silently did nothing, and one gated the worktree argument on task type while the other handed it to every phase. This module is the single reference point so a fix cannot land on one lead runtime only. Enforced by `tests/contract/test_dispatcher_shared_helpers.py`. """ from __future__ import annotations import contextlib import fcntl import json import os import tempfile import uuid from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable, Mapping, Sequence from . import cmux from .agent_invocation import ( AgentInvocationError, AgentModelAssignment, InvocationMetadataIdentity, agent_model_assignment_from_payload, invocation_metadata_identity, v2_role_assignment_authority_errors, verify_agent_invocation, ) from .domain.role import RoleCatalogError, role_for_duty from .execution_identity import Attempt, Invocation from .execution_manifest import ( finish_attempt_mutation, read_execution_manifest, record_invocation_attempt, ) from .execution_mutation_audit import ExecutionMutationAudit, MutationSnapshot from .final_report_paths import final_report_data_path from .report_inputs import report_narrative_path, uses_report_contract_v3 from .worker_prompt_body import REPORT_WRITER_WORKER_ID from .worker_prompt_contract import ( PromptRecord, validate_initial_prompt_records, validate_reverify_prompt, ) from .worker_runner import LIVE, QUIET from .worker_request import verifier_extra_dirs from .worker_artifact_paths import audit_sidecar_rel from .wrapper_status import ( log_path_for_prompt, prompt_derived_paths, status_path_for_prompt, ) from .write_policy import ( build_invocation_write_contract, planned_paths_from_run_manifest, write_policy_from_payload, ) from .convergence_store import write_json_atomic from .json_boundary import ( JsonBoundaryError, external_worker_json_source, load_external_json, load_owned_object, write_owned_object_atomic, ) BACKEND_CLI_WRAPPER = "cli-wrapper" BACKEND_CMUX_PANE = "cmux-pane" BACKEND_MIXED = "mixed" def detect_terminal_backend() -> str: """Which backend this run's workers get. Called once, by prepare. cmux is the only backend that gives a worker a surface of its own. Without it a worker runs as a cli-wrapper subprocess, which owns no pane and is read through its status sidecar instead. The answer is written to the run manifest and read back from there — consumers must not re-detect, or two phases of one run can disagree. """ if cmux.cmux_available(): return BACKEND_CMUX_PANE return BACKEND_CLI_WRAPPER # `livenessMode` picks which artifact answers "is this worker still alive": the # in-process worker's audit sidecar heartbeat, or the CLI wrapper's status # sidecar. Both dispatchers write it and `worker_liveness` reads it, so the # vocabulary lives here rather than in whichever module happened to need it. LIVENESS_AUDIT_HEARTBEAT = "audit-heartbeat" LIVENESS_WRAPPER_STATUS = "wrapper-status" # `okstra team teardown` stamps this on a dispatch it wrote off, and a later # `okstra team await` matches it to tell those apart from a genuine dispatch # error before letting the wrapper's own exit settle them. Two sides of one fact, # so the string lives here rather than in whichever module wrote it first. TEARDOWN_BEFORE_TERMINAL_REASON = "teardown before terminal status" # The worktree argument grants the wrapper `--add-dir` write access outside # project-root. Only the phases whose workers mutate a stage worktree get it; # analysis phases write their artifacts under project-root (see the contract in # scripts/okstra-codex-exec.sh). WORKTREE_TASK_TYPES = frozenset({"implementation", "final-verification"}) WORKER_STATUSES = frozenset( {"in-progress", "completed", "timeout", "error", "not-run"} ) # Which of those mean the dispatch is still expected to produce something. Two # readers key off this split — `team reclaim` closes a finished dispatch's pane # and must never touch a live one, and the compact-reminder hook calls a run # in-flight when any dispatch is still here. Both restated the terminal four # locally before, so a sixth status would have read as finished in one place and # as live in the other. NON_TERMINAL_WORKER_STATUSES = frozenset({"in-progress"}) TERMINAL_WORKER_STATUSES = WORKER_STATUSES - NON_TERMINAL_WORKER_STATUSES REASON_REQUIRED_STATUSES = frozenset({"timeout", "error", "not-run"}) def generate_claude_session_id() -> str: """A UUIDv4 for `claude --session-id`. The lead mints one for its own session (`run`) and dispatch mints one per worker (`dispatch_session_id`); they are the same kind of value, so one generator serves both. It sits here rather than in `session` because `session` imports this module, so a worker-side call the other way would close an import cycle. """ return str(uuid.uuid4()) class DispatchError(Exception): """Raised when a worker dispatch request cannot be executed.""" class IncompletePromptWriteArtifacts(DispatchError): """Raised when a worker prompt omits the standard artifact headers.""" @dataclass(frozen=True) class WorkerJob: worker_id: str provider: str backend: str project_root: Path model_execution_value: str wrapper_path: Path prompt_path: Path result_path: Path worker_result_path: Path completion_paths: tuple[Path, ...] worktree_path: str role: str # None 은 "이 디스패치가 예산을 정하지 않는다" 는 뜻이고, 그때 예산은 # 역할이 정한다(`domain/worker_role.role_spec`). 여기에 숫자 기본값을 두면 # 역할별 예산이 영영 도달 불가능해진다 — 엔트리포인트는 이 자리가 비었을 # 때만 역할을 보기 때문이다. idle_timeout_seconds: int | None dispatch_kind: str invocation_id: str = "" audience: str = "" assignment_ref: str = "" prompt_metadata_path: Path = Path() catalog_digest: str = "" assignment_digest: str = "" duty_digest: str = "" instruction_digest: str = "" prompt_digest: str = "" host_model_value: str | None = None enforcement_mode: str = "" # The session id dispatch mints and carries through to the provider CLI, so # token collection can find the worker's jsonl by this value rather than by # guessing at `agentName`. A worker started in a cmux pane is a separate CLI # process and leaves no `agentName` behind, so without this its usage cannot # be attributed at all. session_id: str = "" participant_ref: str = "" role_execution_ref: str = "" execution_label: str = "" duty_id: str = "" invocation_ref: str = "" attempt: int = 0 @property def command(self) -> list[str]: # The flag trails the positional contract the entrypoint reads by index. argv = [ str(self.wrapper_path), str(self.project_root), self.model_execution_value, str(self.prompt_path), self.worktree_path, self.wrapper_role, self._idle_timeout_argument, "--presentation", self._presentation(), ] if self.has_execution_identity: argv.extend([ "--invocation-metadata", str(self.prompt_metadata_path), ]) if self.session_id: argv += ["--session-id", self.session_id] return argv @property def _idle_timeout_argument(self) -> str: """엔트리포인트가 읽는 유휴 예산 자리. 비워 두면 엔트리포인트가 역할의 예산을 쓴다. 이 자리를 항상 채우던 동안에는 `worker_request.idle_timeout` 의 역할 분기가 한 번도 실행되지 않았고, executor/verifier 의 1500s 도 함께 죽어 있었다. """ if self.idle_timeout_seconds is None: return "" return str(self.idle_timeout_seconds) @property def wrapper_role(self) -> str: """The canonical role the entrypoint's role positional is read as. `self.role` is the roster label team-state carries (`Antigravity worker`) so the report's execution-status row can quote it. The entrypoint reads that same position through `normalize_role`, which knows only the eleven canonical ids, and compares it against `role_for_duty(dutyId)` from the invocation metadata. Handing it the label failed that check before the status sidecar was written, so a pane worker exited 64 leaving no `.log` and no `.status.json` while team-state still read `in-progress`. It also silently denied the verifier the toolchain dirs `verifier_extra_dirs` grants by role. A legacy v1 job carries no duty id; the entrypoint skips the metadata role check for it, so its existing value is preserved. """ if not self.duty_id: return self.role try: return role_for_duty(self.duty_id) except RoleCatalogError: return self.role def _presentation(self) -> str: # A pane is a screen a person watches; a cli-wrapper dispatch's stdout is # a subagent's context window. Progress belongs in the first and not the # second. return QUIET if self.backend == BACKEND_CLI_WRAPPER else LIVE def to_payload(self) -> dict[str, Any]: payload = { "provider": self.provider, "backend": self.backend, "modelExecutionValue": self.model_execution_value, "wrapperPath": str(self.wrapper_path), "promptPath": str(self.prompt_path), "resultPath": str(self.result_path), "workerResultPath": str(self.worker_result_path), "completionPaths": [str(path) for path in self.completion_paths], "worktreePath": self.worktree_path, "role": self.role, "dispatchKind": self.dispatch_kind, "command": self.command, } if self.has_execution_identity: payload.update({ "schemaVersion": "2.0", "executionIdentityVersion": 2, "participantRef": self.participant_ref, "roleExecutionRef": self.role_execution_ref, "executionLabel": self.execution_label, "dutyId": self.duty_id, "invocationRef": self.invocation_ref, "attempt": self.attempt, }) else: payload.update({"schemaVersion": 1, "workerId": self.worker_id}) if self.invocation_id: payload.update({ "invocationId": self.invocation_id, "audience": self.audience, "assignmentRef": self.assignment_ref, "promptMetadataPath": str(self.prompt_metadata_path), "digests": self.digests, "hostModelValue": self.host_model_value, "enforcementMode": self.enforcement_mode, }) return payload @property def has_execution_identity(self) -> bool: values = ( self.participant_ref, self.role_execution_ref, self.execution_label, self.duty_id, self.invocation_ref, ) present = tuple(bool(value) for value in values) if any(present) and not all(present): raise DispatchError("worker job has partial v2 execution identity") if all(present) and self.attempt < 1: raise DispatchError("worker job v2 attempt must be positive") return all(present) @property def digests(self) -> dict[str, str]: return { "catalogDigest": self.catalog_digest, "assignmentDigest": self.assignment_digest, "dutyDigest": self.duty_digest, "instructionDigest": self.instruction_digest, "promptDigest": self.prompt_digest, } # --- run-manifest reads ------------------------------------------------------- def string_value(value: Any) -> str: return value.strip() if isinstance(value, str) else "" def require_string(payload: Mapping[str, Any], key: str) -> str: value = payload.get(key) if not isinstance(value, str) or not value.strip(): raise DispatchError(f"missing required string field: {key}") return value.strip() def string_list(value: Any) -> list[str]: if not isinstance(value, list): return [] return [str(item).strip() for item in value if str(item).strip()] def resolve_project_path(project_root: Path, value: str) -> Path: path = Path(value) return path if path.is_absolute() else project_root / path def resolve_required_path( project_root: Path, manifest: Mapping[str, Any], key: str ) -> Path: return resolve_project_path(project_root, require_string(manifest, key)) def worker_prompt_path(manifest: Mapping[str, Any], worker_id: str) -> str: paths = manifest.get("workerPromptPathByWorkerId") if not isinstance(paths, Mapping): raise DispatchError("run manifest has no workerPromptPathByWorkerId object") return require_string(paths, worker_id) def worktree_path( manifest: Mapping[str, Any], active_context: Mapping[str, Any] ) -> str: if manifest.get("taskType") not in WORKTREE_TASK_TYPES: return "" worktree = active_context.get("executorWorktree") if not isinstance(worktree, Mapping): return "" return str(worktree.get("path") or "") # --- team-state document ------------------------------------------------------ def load_json_object(path: Path, label: str) -> dict[str, Any]: if not path.is_file(): raise DispatchError(f"{label} not found: {path}") try: payload = load_owned_object(path, artifact=label) except JsonBoundaryError as exc: raise DispatchError(f"{label} is invalid JSON: {path}: {exc}") from exc if not isinstance(payload, dict): raise DispatchError(f"{label} must be a JSON object: {path}") return payload def write_json(path: Path, payload: Mapping[str, Any]) -> None: # Write via temp file + os.replace so a crash mid-write cannot leave a # truncated team-state.json that every later load_json_object rejects; # os.replace is atomic on POSIX, matching run_context._atomic_write_json. write_owned_object_atomic(path, payload, artifact="team state") @contextlib.contextmanager def _team_state_lock(path: Path): lock_path = path.with_name(path.name + ".lock") lock_path.parent.mkdir(parents=True, exist_ok=True) handle = lock_path.open("a+") try: fcntl.flock(handle.fileno(), fcntl.LOCK_EX) yield finally: fcntl.flock(handle.fileno(), fcntl.LOCK_UN) handle.close() def mutate_team_state( team_state_path: Path, mutation: Callable[[dict[str, Any]], bool], ) -> bool: """Apply one arbitrary team-state mutation under the shared file lock.""" with _team_state_lock(team_state_path): payload = load_json_object(team_state_path, "team-state") changed = mutation(payload) if changed: write_json(team_state_path, payload) return changed def append_worker_dispatch( team_state_path: Path, record: Mapping[str, Any], ) -> None: """Record one worker dispatch without losing another state writer's update. A row carrying a dispatch id replaces the row that already holds it rather than stacking beside it: an observed lead re-sent `okstra team dispatch` for a prompt and attempt it had already sent, team-state kept both copies, and `validate-run` failed that run with `agent dispatch ID is duplicated`. The replaced row takes its `paneId` with it, and recorded ids are the only reclaim candidates `team._reclaimable_panes` has, so that pane is left on screen for the session. That is the price of the trade: before this, the same re-send cost the whole run its report. A job with no invocation carries no dispatch id, so it has no key to collapse on and folding those rows would merge two unrelated dispatches into one. They still stack, and `update_worker_dispatch_status` settles the newest of them. """ with _team_state_lock(team_state_path): payload = load_json_object(team_state_path, "team-state") dispatches = payload.setdefault("workerDispatches", []) if not isinstance(dispatches, list): raise DispatchError( f"team-state workerDispatches must be an array: {team_state_path}" ) new_record = dict(record) new_id = str(new_record.get("dispatchId") or "") replaced = False if new_id: for index, existing in enumerate(dispatches): if ( isinstance(existing, Mapping) and existing.get("dispatchId") == new_id ): dispatches[index] = new_record replaced = True break if not replaced: dispatches.append(new_record) write_json(team_state_path, payload) def update_worker_dispatch_status( team_state_path: Path, *, prompt_path: Path, attempt: int, status: str, reason: str, ) -> bool: """Settle the newest matching dispatch row under the shared state lock.""" with _team_state_lock(team_state_path): payload = load_json_object(team_state_path, "team-state") dispatches = payload.get("workerDispatches", []) if not isinstance(dispatches, list): raise DispatchError( f"team-state workerDispatches must be an array: {team_state_path}" ) matches = [ record for record in dispatches if dispatch_record_matches(record, prompt_path, attempt) ] if not matches: return False # A re-send can reuse prompt + attempt. The newest row is the live one. matches[-1]["status"] = status matches[-1]["reason"] = reason write_json(team_state_path, payload) return True _AGENT_DIGEST_KEYS = ( "catalogDigest", "assignmentDigest", "dutyDigest", "instructionDigest", "promptDigest", ) _AGENT_ENFORCEMENT_MODES = frozenset({ "core-pre-dispatch", "host-native-spec-link-gate", }) def record_verified_agent_dispatch( *, project_root: Path, run_manifest_path: Path, metadata_path: Path, enforcement_mode: str, allow_native_core: bool = False, ) -> dict[str, Any]: """Append one immutable dispatch-to-invocation association. Host-native calls use this record as a specification link; it deliberately does not claim that Okstra observed the bytes delivered by the host. A code-owned lead process may set ``allow_native_core`` because lead model assignments use the host's native model vocabulary even when Okstra owns the outer provider CLI process. """ project_root = project_root.resolve() run_manifest_path = resolve_project_path( project_root, str(run_manifest_path) ).resolve(strict=True) metadata_path = resolve_project_path( project_root, str(metadata_path) ).resolve(strict=True) if enforcement_mode not in _AGENT_ENFORCEMENT_MODES: raise DispatchError(f"invalid agent enforcement mode: {enforcement_mode}") manifest = load_json_object(run_manifest_path, "run manifest") metadata = load_json_object(metadata_path, "agent invocation metadata") assignment_ref = require_string(metadata, "assignmentRef") assignments = manifest.get("invocationAssignments") if not isinstance(assignments, Mapping): raise DispatchError("run manifest has no invocationAssignments object") try: assignment = agent_model_assignment_from_payload(assignments.get(assignment_ref)) except AgentInvocationError as exc: raise DispatchError(str(exc)) from exc try: execution_identity = invocation_metadata_identity(metadata) except AgentInvocationError as exc: raise DispatchError(str(exc)) from exc worker_id = ( require_string(metadata, "workerId") if execution_identity is None else None ) audience = require_string(metadata, "audience") invocation_id = require_string(metadata, "invocationId") if ( execution_identity is not None and execution_identity.invocation_ref != invocation_id ): raise DispatchError("v2 invocation reference does not match invocation ID") errors = verify_agent_invocation( metadata_path, project_root=project_root, expected_run_manifest_path=run_manifest_path, expected_assignment=assignment, expected_invocation_id=invocation_id, expected_worker_id=worker_id, expected_assignment_ref=assignment_ref, expected_audience=audience, expected_participant_ref=( execution_identity.participant_ref if execution_identity else None ), expected_role_execution_ref=( execution_identity.role_execution_ref if execution_identity else None ), expected_invocation_ref=( execution_identity.invocation_ref if execution_identity else None ), ) if errors: raise DispatchError("agent invocation verification failed: " + "; ".join(errors)) if ( enforcement_mode == "host-native-spec-link-gate" and assignment.runner != "native-session" and not _is_current_session_lead( run_manifest_path, execution_identity, assignment ) ): raise DispatchError( "host-native enforcement requires a native-session assignment" ) if ( enforcement_mode == "core-pre-dispatch" and assignment.runner == "native-session" and not allow_native_core ): raise DispatchError( "native-session assignment must use host-native-spec-link-gate" ) digests = metadata.get("digests") if not isinstance(digests, Mapping): raise DispatchError("agent invocation digests are missing") digest_values = { key: require_string(digests, key) for key in _AGENT_DIGEST_KEYS } prompt = metadata.get("prompt") if not isinstance(prompt, Mapping): raise DispatchError("agent invocation prompt is missing") prompt_path = require_string(prompt, "path") # Identity-bearing records carry the stored attempt. A path without # identity still records the first attempt; the conflict guard below # refuses a second write under that id. dispatch_id = build_dispatch_id( invocation_id, execution_identity.attempt if execution_identity else 1, ) record = { "dispatchId": dispatch_id, "audience": audience, "invocationId": invocation_id, "assignmentRef": assignment_ref, "promptPath": prompt_path, "promptMetadataPath": _relative_project_path(project_root, metadata_path), **digest_values, "modelExecutionValue": assignment.model_execution_value, "hostModelValue": assignment.host_model_value, "enforcementMode": enforcement_mode, "promptDeliveryVerified": False, "status": "dispatched", } if execution_identity is None: record["workerId"] = worker_id else: record.update({ "schemaVersion": "2.0", "executionIdentityVersion": 2, "invocationRef": execution_identity.invocation_ref, "participantRef": execution_identity.participant_ref, "roleExecutionRef": execution_identity.role_execution_ref, "executionLabel": execution_identity.execution_label, "dutyId": execution_identity.duty_id, "attempt": execution_identity.attempt, "dispatchKind": require_string(metadata, "dispatchKind"), }) team_state_path = resolve_required_path(project_root, manifest, "teamStatePath") write_policy = None write_enforcement = None mutation_snapshot_path = metadata_path.with_suffix( metadata_path.suffix + ".mutation-audit.json" ) if execution_identity is not None: write_policy, write_enforcement = _agent_write_contract( project_root, run_manifest_path, manifest, execution_identity.role_execution_ref, resolve_project_path(project_root, prompt_path), assignment_ref, ) record.update({ "writePolicyDigest": write_policy.digest, "writeEnforcement": write_enforcement.to_payload(), "mutationAuditSnapshotPath": ( _relative_project_path(project_root, mutation_snapshot_path) if write_enforcement.mutation_audit == "batch" else "" ), }) existing_state = load_json_object(team_state_path, "team-state") existing = [ item for item in existing_state.get("agentDispatches") or [] if isinstance(item, Mapping) and item.get("dispatchId") == dispatch_id ] if existing: if len(existing) != 1 or dict(existing[0]) != record: raise DispatchError(f"agent dispatch ID conflicts: {dispatch_id}") return record if execution_identity is not None: dispatch_kind = require_string(metadata, "dispatchKind") prompt_digest = require_string(digests, "promptDigest") if write_policy is None or write_enforcement is None: raise DispatchError("agent write contract was not prepared") record_invocation_attempt( run_manifest_path, Invocation( invocation_ref=execution_identity.invocation_ref, participant_ref=execution_identity.participant_ref, role_execution_ref=execution_identity.role_execution_ref, source_invocation_ref=None, recovery_ref=None, duty_id=execution_identity.duty_id, dispatch_kind=dispatch_kind, round=_execution_dispatch_round(dispatch_kind), input_digest=prompt_digest, write_policy=write_policy.to_payload(), write_policy_digest=write_policy.digest, write_enforcement=write_enforcement.to_payload(), ), Attempt( invocation_ref=execution_identity.invocation_ref, attempt=execution_identity.attempt, started_at=utc_now(), finished_at=None, status="started", result_path=None, error_path=None, change_summary={}, evidence_seal_ref=None, ), task_key=require_string(manifest, "taskKey"), ) if write_enforcement.mutation_audit == "batch": snapshot = ExecutionMutationAudit().snapshot( (write_policy,), orchestrator_paths=( # run 산출물 트리 전체. 같은 run 의 다른 라운드가 감사 창 # 안에 남기는 프롬프트·로그·사이드카를 파일 단위로 미리 셀 # 수 없고, 그 아래는 정의상 소스가 아니다. Path(run_manifest_path).resolve().parents[1], run_manifest_path, team_state_path, Path(f"{team_state_path}.lock"), mutation_snapshot_path, ), ) write_json_atomic(mutation_snapshot_path, snapshot.to_payload()) with _team_state_lock(team_state_path): team_state = load_json_object(team_state_path, "team-state") dispatches = team_state.setdefault("agentDispatches", []) if not isinstance(dispatches, list): raise DispatchError("team-state agentDispatches must be an array") existing = [ item for item in dispatches if isinstance(item, Mapping) and item.get("dispatchId") == dispatch_id ] if existing: if len(existing) != 1 or dict(existing[0]) != record: raise DispatchError(f"agent dispatch ID conflicts: {dispatch_id}") return record dispatches.append(record) write_json(team_state_path, team_state) return record def _is_current_session_lead( manifest_path: Path, execution_identity: InvocationMetadataIdentity | None, assignment: AgentModelAssignment, ) -> bool: """Report whether this dispatch is the lead attesting its own session. A current-session lead has no model binding, so the run manifest projects its assignment as ``cli-wrapper`` while the execution manifest records the participant as ``current-session``. It runs through no dispatch boundary at all: the spec link is the lead associating the already-running session with its verified invocation specification, which is what the host-native gate records. Keyed on the execution manifest's own participant row rather than on the projected runner string, so a genuine cli-wrapper worker never reaches the native gate. """ if execution_identity is None or assignment.runner == "native-session": return False manifest = read_execution_manifest(manifest_path) if manifest.legacy: return False participant = next( ( row for row in manifest.participant_assignments if row.participant_ref == execution_identity.participant_ref ), None, ) execution = next( ( row for row in manifest.role_executions if row.role_execution_ref == execution_identity.role_execution_ref ), None, ) return ( participant is not None and execution is not None and execution.role == "leader" and participant.runner == "current-session" and participant.entry_mode == "current-session" ) def _agent_write_contract( project_root: Path, manifest_path: Path, authority: Mapping[str, Any], role_execution_ref: str, prompt_path: Path, assignment_ref: str, ): execution_manifest = read_execution_manifest(manifest_path) execution = next( ( row for row in execution_manifest.role_executions if row.role_execution_ref == role_execution_ref ), None, ) capability = ( execution.binding.worker_write_capability if execution is not None and execution.binding is not None else None ) if execution is None: raise DispatchError("agent write policy has no role execution") if execution.role == "leader": artifacts = _leader_artifact_paths(project_root, authority) worktree = None maximum_precision = "none" elif capability is not None: try: artifacts, worktree = _prompt_write_paths(project_root, prompt_path) except IncompletePromptWriteArtifacts: artifacts, worktree = _manifest_worker_write_paths( project_root, authority, assignment_ref, prompt_path ) maximum_precision = capability.max_boundary_precision else: raise DispatchError("agent write policy has no runner capability") planned = ( planned_paths_from_run_manifest(project_root, authority) if execution.role == "implementer" else ((), True) ) try: return build_invocation_write_contract( role=execution.role, project_root=project_root, worktree=worktree, artifact_paths=artifacts, maximum_precision=maximum_precision, planned_paths=planned[0], planned_paths_declared=planned[1], auxiliary_roots=verifier_extra_dirs(execution.role), validated_auxiliary_roots=verifier_extra_dirs(execution.role), ) except (OSError, ValueError) as exc: raise DispatchError(f"agent write policy is invalid: {exc}") from exc def _leader_artifact_paths( project_root: Path, authority: Mapping[str, Any] ) -> tuple[Path, ...]: value = authority.get("taskRootPath") or authority.get("runDirectoryPath") if not isinstance(value, str) or not value.strip(): raise DispatchError("leader write policy has no artifact root") artifact_home = project_root / ".okstra" if not artifact_home.is_dir(): raise DispatchError("leader write policy artifact home does not exist") return (artifact_home,) def _manifest_worker_write_paths( project_root: Path, authority: Mapping[str, Any], assignment_ref: str, prompt_path: Path, ) -> tuple[tuple[Path, ...], Path | None]: prefix = "initial/" if not assignment_ref.startswith(prefix): raise DispatchError("agent prompt write artifact headers are incomplete") worker_id = assignment_ref.removeprefix(prefix) team_state_path = resolve_required_path( project_root, authority, "teamStatePath" ) team_state = load_json_object(team_state_path, "team-state") matches = [ row for row in team_state.get("workers") or [] if isinstance(row, Mapping) and row.get("workerId") == worker_id ] if len(matches) != 1: raise DispatchError("agent prompt has no unique worker artifact authority") result = resolve_project_path( project_root, require_string(matches[0], "resultPath") ) paths = { *dispatch_completion_paths(worker_id, result, authority, project_root), result, Path(audit_sidecar_rel(str(result))), status_path_for_prompt(prompt_path), log_path_for_prompt(prompt_path), } active_value = authority.get("activeRunContextPath") active = ( load_json_object( resolve_project_path(project_root, active_value), "active run context", ) if isinstance(active_value, str) and active_value else {} ) error_logs = active.get("errorLogs") sidecars = ( error_logs.get("sidecarsByWorkerId") if isinstance(error_logs, Mapping) else None ) error_value = sidecars.get(worker_id) if isinstance(sidecars, Mapping) else None if isinstance(error_value, str) and error_value: paths.add(resolve_project_path(project_root, error_value)) worktree_value = worktree_path(authority, active) worktree = Path(worktree_value) if worktree_value else None return tuple(sorted(paths, key=str)), worktree def _prompt_write_paths( project_root: Path, prompt_path: Path ) -> tuple[tuple[Path, ...], Path | None]: try: lines = prompt_path.read_text(encoding="utf-8").splitlines() except (OSError, UnicodeError) as exc: raise DispatchError("agent prompt cannot be read for write policy") from exc values: dict[str, str] = {} labels = { "Result Path", "Worker Result Path", "Audit sidecar path", "Errors sidecar path", "Worktree", } for line in lines: for label in labels: prefix = f"**{label}:** " if line.startswith(prefix): values[label] = line.removeprefix(prefix).strip() required = {"Result Path", "Audit sidecar path", "Errors sidecar path"} if not required <= values.keys(): raise IncompletePromptWriteArtifacts( "agent prompt write artifact headers are incomplete" ) artifact_values = [values[label] for label in sorted(required)] if "Worker Result Path" in values: artifact_values.append(values["Worker Result Path"]) artifacts = { _project_or_absolute(project_root, value) for value in artifact_values } artifacts.update(prompt_derived_paths(prompt_path)) worktree_value = values.get("Worktree") worktree = ( _project_or_absolute(project_root, worktree_value) if worktree_value else None ) return tuple(sorted(artifacts, key=str)), worktree def _project_or_absolute(project_root: Path, value: str) -> Path: path = Path(value) return path if path.is_absolute() else project_root / path def _execution_dispatch_round(dispatch_kind: str) -> int: prefix = "reverify-r" if not dispatch_kind.startswith(prefix): return 1 suffix = dispatch_kind[len(prefix):] if not suffix.isdigit() or int(suffix) < 1: raise DispatchError(f"invalid reverify dispatch kind: {dispatch_kind}") return int(suffix) def link_agent_dispatch_result( *, project_root: Path, run_manifest_path: Path, dispatch_id: str, result_path: Path, ) -> dict[str, Any]: """Link an existing result artifact to exactly one verified dispatch.""" project_root = project_root.resolve() manifest_path = resolve_project_path( project_root, str(run_manifest_path) ).resolve(strict=True) manifest = load_json_object(manifest_path, "run manifest") run_root = resolve_required_path( project_root, manifest, "runDirectoryPath" ).resolve(strict=True) resolved_result = resolve_project_path( project_root, str(result_path) ).resolve(strict=True) if not resolved_result.is_relative_to(run_root): raise DispatchError("agent result link must stay inside the current run") result_relative = _relative_project_path(project_root, resolved_result) team_state_path = resolve_required_path(project_root, manifest, "teamStatePath") preview = load_json_object(team_state_path, "team-state") native_dispatches = [ item for item in preview.get("agentDispatches") or [] if isinstance(item, Mapping) and item.get("dispatchId") == dispatch_id ] if len(native_dispatches) == 1: _finish_native_agent_attempt( project_root=project_root, manifest_path=manifest_path, manifest=manifest, dispatch=native_dispatches[0], result_path=resolved_result, result_relative=result_relative, ) with _team_state_lock(team_state_path): team_state = load_json_object(team_state_path, "team-state") dispatches = [ item for collection in ( team_state.get("agentDispatches") or [], team_state.get("workerDispatches") or [], ) for item in collection if isinstance(item, Mapping) and item.get("dispatchId") == dispatch_id ] if len(dispatches) != 1: raise DispatchError( f"agent result link requires exactly one dispatch record: {dispatch_id}" ) dispatch = dispatches[0] link: dict[str, Any] = { "dispatchId": dispatch_id, "resultPath": result_relative, } identity = worker_execution_identity(dispatch) if identity is not None: link.update({ "participantRef": identity["participant_ref"], "roleExecutionRef": identity["role_execution_ref"], "executionLabel": identity["execution_label"], "dutyId": identity["duty_id"], "invocationRef": identity["invocation_ref"], "attempt": identity["attempt"], }) links = team_state.setdefault("agentResultLinks", []) if not isinstance(links, list): raise DispatchError("team-state agentResultLinks must be an array") same_result = [ item for item in links if isinstance(item, Mapping) and item.get("resultPath") == result_relative ] # A superseded link is the record of a result the lead rejected, kept so # the chain shows the corrective round happened rather than hiding it. # It no longer owns the path, so the re-dispatch may claim it. live_conflicts = [ item for item in same_result if not item.get("supersededBy") and dict(item) != link ] if live_conflicts: raise DispatchError( f"agent result is already linked to another dispatch: " f"{result_relative}. If that result was rejected and re-dispatched, " f"record the rejection with `okstra agent-prompt reject-result " f"--dispatch-id {live_conflicts[0].get('dispatchId')} " f"--superseded-by {dispatch_id}` first" ) if any( isinstance(item, Mapping) and item.get("dispatchId") == dispatch_id and item.get("resultPath") != result_relative for item in links ): raise DispatchError( f"agent dispatch is already linked to another result: {dispatch_id}" ) if link not in links: links.append(link) write_json(team_state_path, team_state) return link def _finish_native_agent_attempt( *, project_root: Path, manifest_path: Path, manifest: Mapping[str, Any], dispatch: Mapping[str, Any], result_path: Path, result_relative: str, ) -> None: identity = worker_execution_identity(dispatch) if identity is None: return execution = read_execution_manifest(manifest_path) invocation = next( ( row for row in execution.invocations if row.invocation_ref == identity["invocation_ref"] ), None, ) if invocation is None: raise DispatchError("agent result has no canonical invocation policy") if dispatch.get("writePolicyDigest") != invocation.write_policy_digest: raise DispatchError("agent dispatch write policy digest changed") if dispatch.get("writeEnforcement") != invocation.write_enforcement: raise DispatchError("agent dispatch write enforcement changed") policy = write_policy_from_payload(invocation.write_policy) status, summary = _native_mutation_result( project_root=project_root, dispatch=dispatch, result_path=result_path, policy=policy, ) _close_native_attempt( manifest_path, manifest, identity, status, result_relative, summary, ) if status != "ok": violations = summary.get("violations") detail = "; ".join(violations) if isinstance(violations, list) else status raise DispatchError(f"agent mutation audit failed: {detail or status}") def _native_mutation_result( *, project_root: Path, dispatch: Mapping[str, Any], result_path: Path, policy: WritePolicy, ) -> tuple[str, Mapping[str, Any]]: snapshot_value = string_value(dispatch.get("mutationAuditSnapshotPath")) if snapshot_value: snapshot_payload = load_json_object( resolve_project_path(project_root, snapshot_value), "mutation audit snapshot", ) try: snapshot = MutationSnapshot.from_payload(snapshot_payload) except (KeyError, TypeError, ValueError) as exc: raise DispatchError("mutation audit snapshot is invalid") from exc result = ExecutionMutationAudit().compare( snapshot, (policy,), out_of_plan_edits=_native_out_of_plan_paths( result_path, project_root=project_root ), attempt_succeeded=True, result_present=True, ) summary = result.change_summary() status = result.status else: status = "ok" summary = { "mutationStatus": "ok", "changedPaths": [], "sourceChanged": False, "gitChanged": False, "attribution": "call", "retryAllowed": False, "violations": [], "beforeDigest": "", "afterDigest": "", } return status, summary def _close_native_attempt( manifest_path: Path, manifest: Mapping[str, Any], identity: Mapping[str, Any], status: str, result_path: str, summary: Mapping[str, Any], ) -> None: execution = read_execution_manifest(manifest_path) existing = next( ( row for row in execution.attempts if row.invocation_ref == identity["invocation_ref"] and row.attempt == identity["attempt"] ), None, ) if existing is not None and existing.finished_at is not None: if ( existing.status == status and existing.result_path == result_path and dict(existing.change_summary) == dict(summary) ): return raise DispatchError("agent attempt terminal mutation result drift") finish_attempt_mutation( manifest_path, invocation_ref=identity["invocation_ref"], attempt=identity["attempt"], finished_at=utc_now(), status=status, result_path=result_path, error_path=None, change_summary=summary, task_key=require_string(manifest, "taskKey"), ) def _native_out_of_plan_paths( result_path: Path, *, project_root: Path ) -> tuple[str, ...]: if result_path.suffix != ".json": return () try: # 외부 입력: 네이티브 모델 결과에서 선택적 수정 경로만 투영한다. payload = load_external_json( external_worker_json_source( result_path, trusted_root=project_root, lane_root=result_path.parent, ), artifact="native worker result", ) except JsonBoundaryError: return () implementation = payload.get("implementation") if isinstance(payload, Mapping) else None rows = implementation.get("outOfPlanEdits") if isinstance(implementation, Mapping) else None if not isinstance(rows, list): return () return tuple( str(row["file"]) for row in rows if isinstance(row, Mapping) and isinstance(row.get("file"), str) and row["file"] ) def reject_agent_dispatch_result( *, project_root: Path, run_manifest_path: Path, dispatch_id: str, superseded_by: str, reason: str, ) -> dict[str, str]: """Record that a linked result was rejected and re-dispatched. The contract tells the lead to re-dispatch a worker whose answer violated the response format, with a correction paragraph appended. That path did not finish: the prompt is immutable and already dispatched, so `--replace-undispatched` refuses (correctly), and a fresh invocation id produces a worker that writes the same result path — where `link-result` refused because the path already belonged to the first dispatch. The worker ran, wrote a good result, and the ledger could not accept it. Rejection is a fact worth recording rather than routing around, so it is written rather than allowed implicitly: the first link stays in the ledger marked `supersededBy` with the lead's reason, and only then may the re-dispatch claim the path. Nothing is deleted, so the chain still shows both attempts and why the second exists. """ if not superseded_by.strip(): raise DispatchError("superseding dispatch ID is required") if not reason.strip(): raise DispatchError( "a rejection reason is required — the ledger has to say why a " "returned result was not accepted" ) project_root = project_root.resolve() manifest_path = resolve_project_path( project_root, str(run_manifest_path) ).resolve(strict=True) manifest = load_json_object(manifest_path, "run manifest") team_state_path = resolve_required_path(project_root, manifest, "teamStatePath") with _team_state_lock(team_state_path): team_state = load_json_object(team_state_path, "team-state") links = team_state.get("agentResultLinks") if not isinstance(links, list): raise DispatchError("team-state agentResultLinks must be an array") matches = [ item for item in links if isinstance(item, Mapping) and item.get("dispatchId") == dispatch_id ] if len(matches) != 1: raise DispatchError( f"expected exactly one agent result link for {dispatch_id}, " f"found {len(matches)}" ) row = matches[0] if row.get("supersededBy"): raise DispatchError( f"agent result link is already superseded by " f"{row['supersededBy']}: {dispatch_id}" ) row["supersededBy"] = superseded_by row["rejectionReason"] = reason write_json(team_state_path, team_state) return dict(row) def _relative_project_path(project_root: Path, path: Path) -> str: try: return path.resolve(strict=False).relative_to(project_root).as_posix() except ValueError as exc: raise DispatchError(f"path is outside project root: {path}") from exc def worker_state(team_state: Mapping[str, Any], worker_id: str) -> Mapping[str, Any]: workers = team_state.get("workers") if not isinstance(workers, list): raise DispatchError("team-state workers must be an array") for worker in workers: if isinstance(worker, Mapping) and worker.get("workerId") == worker_id: return worker raise DispatchError(f"team-state has no workerId={worker_id}") def transition_worker_status( team_state_path: Path, worker_id: str, status: str, reason: str, *, model_execution_value: str = "", at: str | datetime | None = None, ) -> None: if status not in WORKER_STATUSES: allowed = ", ".join(sorted(WORKER_STATUSES)) raise DispatchError( f"unsupported worker status `{status}`; expected one of: {allowed}" ) reason = reason.strip() if status in REASON_REQUIRED_STATUSES and not reason: raise DispatchError(f"worker status `{status}` requires a non-empty reason") timestamp = _utc_timestamp(at) with _team_state_lock(team_state_path): payload = load_json_object(team_state_path, "team-state") workers = payload.get("workers") if not isinstance(workers, list): raise DispatchError(f"team-state workers must be an array: {team_state_path}") for worker in workers: if isinstance(worker, dict) and worker.get("workerId") == worker_id: worker["status"] = status worker["reason"] = reason if status == "in-progress": worker["startedAt"] = timestamp worker.pop("endedAt", None) elif status == "not-run": worker.pop("startedAt", None) worker.pop("endedAt", None) else: worker["endedAt"] = timestamp if model_execution_value: # `model` is the catalog display name the task-manifest # declares; only the execution identifier belongs here. Writing # both from one value destroys the display name on every worker # whose two differ, and `validate-run` compares team-state's # `model` against the manifest. worker["modelExecutionValue"] = model_execution_value write_json(team_state_path, payload) return raise DispatchError(f"team-state has no workerId={worker_id}: {team_state_path}") def _utc_timestamp(value: str | datetime | None) -> str: if value is None: instant = datetime.now(timezone.utc) elif isinstance(value, datetime): instant = value elif isinstance(value, str): try: instant = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError as exc: raise DispatchError(f"invalid UTC timestamp `{value}`") from exc else: raise DispatchError(f"invalid UTC timestamp `{value}`") if instant.tzinfo is None or instant.utcoffset() != timezone.utc.utcoffset(instant): raise DispatchError(f"invalid UTC timestamp `{value}`") return instant.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def record_dispatch_facts(team_state_path: Path, dispatch_mode: str) -> None: """Persist the team-state facts okstra owns at dispatch time. `dispatchMode` names the backend the workers actually went out on. `teamCreate` is the implicit-team audit marker: Claude Code v2.1.178 removed the TeamCreate tool, so `{attempted: false, status: "implicit"}` is a constant of a non-concurrent run rather than a lead judgment. validate-run requires it once any worker has been dispatched, and Phase 7 token attribution reads it to locate worker sessions — leaving it for the lead to hand-write failed every dispatch path whose host adapter does not spell the rule out. A marker the launch prompt pre-recorded (the concurrent-run `skipped` decision) belongs to that run and is never overwritten. """ with _team_state_lock(team_state_path): payload = load_json_object(team_state_path, "team-state") payload["dispatchMode"] = dispatch_mode if not _has_recorded_team_create(payload): payload["teamCreate"] = {"attempted": False, "status": "implicit"} write_json(team_state_path, payload) def _has_recorded_team_create(team_state: Mapping[str, Any]) -> bool: existing = team_state.get("teamCreate") return (isinstance(existing, dict) and bool(str(existing.get("status") or "").strip())) # --- job facts ---------------------------------------------------------------- def build_dispatch_id(invocation_id: str, attempt: int) -> str: """The only place Python assembles a dispatch id. Two generators drifted: this module pinned the attempt to 1 while `dispatch_core` used the real one, so a re-dispatch took the first attempt's id, team-state kept the same row twice, and `validate-run` failed that run with `agent dispatch ID is duplicated`. Prose mirrors of this shape live outside Python, where no guard reaches them: `prompts/lead/convergence.md`, `prompts/lead/report-writer.md`, `prompts/lead/plan-body-verification.md` and the five `adapters/hosts/*/relay.md` each spell an `--dispatch-id` argument out for the lead. They agree with this function only because `record_verified_agent_dispatch` records the first attempt and no other; changing that convention has to change them too. """ return f"{invocation_id}:attempt-{attempt}" def dispatch_mode(jobs: Sequence[WorkerJob]) -> str: backends = {job.backend for job in jobs} if len(backends) == 1: return next(iter(backends)) return BACKEND_MIXED def missing_completion_paths(job: WorkerJob) -> tuple[Path, ...]: return tuple(path for path in job.completion_paths if not path.is_file()) def dispatch_session_id(provider: str) -> str: """The session this dispatch mints for *provider*, or '' for one that has no such flag. Only the claude CLI reads `--session-id`; handing the value to another wrapper would start it with an argument it does not know. An empty value is the same as not issuing one — `WorkerJob.command` appends the flag only when it is set. This lives beside `worker_jobs_from_file` for the same reason `dispatch_result_path` does: both job constructors apply it, and a rule written twice is a rule that drifts. """ return generate_claude_session_id() if provider == "claude" else "" def worker_session_ids( team_state: Mapping[str, Any], worker_id: str | None = None ) -> list[str]: """The claude sessions this run's dispatches issued, in the order recorded. A worker started in a cmux pane is its own `claude -p` process and tags its session jsonl with neither `agentName` nor `teamName`, so nothing in the transcript says which worker wrote it. The id `dispatch_session_id` minted and `_dispatch_record` wrote is the only handle on it. Every id, not one: a retry opens a new session, so one worker legitimately owns several rows with different `sessionId` values, and taking only the first or last bills a single attempt of a retried worker. Pass ``worker_id`` for one worker's sessions, or omit it for the run's. This reads a `workerDispatches[]` row shape, so it belongs beside the code that writes that shape rather than in each of the two modules that read it. A non-list log yields nothing rather than raising: `session.observe_lead_session` calls this for its exclusion set and catches only OSError, so a `TypeError` from iterating a number here would take down `team dispatch` / `team await` / `report-finalize`. Observation not breaking its caller is a contract kept here, the same way `add_observed_session` keeps it for the keys it appends to. """ session_ids: list[str] = [] dispatches = team_state.get("workerDispatches") for record in dispatches if isinstance(dispatches, list) else []: if not isinstance(record, Mapping): continue if worker_id is not None and record.get("workerId") != worker_id: continue session_id = str(record.get("sessionId") or "").strip() if session_id and session_id not in session_ids: session_ids.append(session_id) return session_ids def dispatch_result_path( worker_id: str, worker_result_path: Path, manifest: Mapping[str, Any], project_root: Path, ) -> Path: """What `resultPath` means for this worker. For everyone but the report writer it is the worker-result file. The report writer authors three artifacts and its canonical result is the final-report data.json, not its own `.md` pointer — a distinction that is not cosmetic: `dispatch_core` hands `job.result_path` to report finalization as the data path, so a `.md` there is parsed as JSON and a complete report settles `error`. This lives beside `worker_jobs_from_file` because both job constructors must apply it. It was `dispatch_core`-private while the roster path applied it and the `--jobs-file` path took whatever the file said, which is how the two produced different jobs for the same worker. """ if worker_id != REPORT_WRITER_WORKER_ID: return worker_result_path if uses_report_contract_v3(manifest): return report_narrative_path(project_root, manifest) return final_report_data_path( resolve_required_path(project_root, manifest, "expectedReportRecordPath") ) def dispatch_completion_paths( worker_id: str, worker_result_path: Path, manifest: Mapping[str, Any], project_root: Path, ) -> tuple[Path, ...]: """Every artifact that must exist before this worker counts as done. The report writer's two are the report record and the worker-result pointer. The full reading copy is rendered on demand and is not a completion artifact. """ if worker_id != REPORT_WRITER_WORKER_ID: return (worker_result_path,) if uses_report_contract_v3(manifest): return (report_narrative_path(project_root, manifest), worker_result_path) data_json = final_report_data_path( resolve_required_path(project_root, manifest, "expectedReportRecordPath") ) return (data_json, worker_result_path) def validate_initial_prompts( manifest: Mapping[str, Any], jobs: Sequence[WorkerJob] ) -> None: records = [ PromptRecord( job.worker_id, job.dispatch_kind, job.prompt_path, metadata_path=(job.prompt_metadata_path if job.invocation_id else None), expected_duty_audience=(job.audience if job.invocation_id else None), ) for job in jobs ] errors = validate_initial_prompt_records( manifest=manifest, records=records, require_evidence_ledger=True, ) if errors: task_type = require_string(manifest, "taskType") raise DispatchError(f"{task_type} prompt contract: " + "; ".join(errors)) def validate_dispatch_prompts( manifest: Mapping[str, Any], active_context: Mapping[str, Any], jobs: Sequence[WorkerJob], ) -> None: if isinstance(manifest.get("agentContract"), Mapping): _validate_agent_invocations(manifest, jobs) initial_jobs = [ job for job in jobs if not job.dispatch_kind.startswith("reverify-r") ] if initial_jobs: validate_initial_prompts(manifest, initial_jobs) reverify_jobs = [ job for job in jobs if job.dispatch_kind.startswith("reverify-r") ] if not reverify_jobs: return task_type = require_string(manifest, "taskType") workflow = active_context.get("workflow") if not isinstance(workflow, Mapping): raise DispatchError("reverify prompt contract: active run workflow is missing") forbidden_actions = string_value(workflow.get("forbiddenActions")) if not forbidden_actions: raise DispatchError( "reverify prompt contract: workflow.forbiddenActions is missing" ) errors: list[str] = [] for job in reverify_jobs: try: text = job.prompt_path.read_text(encoding="utf-8") except OSError as exc: errors.append(f"{job.worker_id}: cannot read prompt {job.prompt_path}: {exc}") continue errors.extend( f"{job.worker_id}: {error}" # The job carries the model this dispatch will actually run, so the # hand-written `**Model:**` header is checkable here — and must be, # because a header naming a model the runtime does not serve kills # the worker with a provider 400 minutes later, where it reads as a # worker fault rather than the typo it is. for error in validate_reverify_prompt( text, task_type=task_type, forbidden_actions=forbidden_actions, expected_model=job.model_execution_value or None, ) ) if errors: raise DispatchError("reverify prompt contract: " + "; ".join(errors)) def _validate_agent_invocations( manifest: Mapping[str, Any], jobs: Sequence[WorkerJob], ) -> None: manifest_path_value = require_string(manifest, "runManifestPath") assignments = manifest.get("invocationAssignments") if not isinstance(assignments, Mapping): raise DispatchError("run manifest has no invocationAssignments object") errors: list[str] = [] for job in jobs: missing = [ name for name, value in ( ("invocationId", job.invocation_id), ("audience", job.audience), ("assignmentRef", job.assignment_ref), ("promptMetadataPath", str(job.prompt_metadata_path)), ("catalogDigest", job.catalog_digest), ("assignmentDigest", job.assignment_digest), ("dutyDigest", job.duty_digest), ("instructionDigest", job.instruction_digest), ("promptDigest", job.prompt_digest), ("enforcementMode", job.enforcement_mode), ) if not value or value == "." ] if missing: errors.append( f"{job.worker_id}: agent invocation metadata fields are missing: " + ", ".join(missing) ) continue if job.enforcement_mode != "core-pre-dispatch": errors.append( f"{job.worker_id}: code-owned dispatch requires core-pre-dispatch" ) assignment_payload = assignments.get(job.assignment_ref) try: assignment = agent_model_assignment_from_payload(assignment_payload) except AgentInvocationError as exc: errors.append(f"{job.worker_id}: {exc}") continue if assignment.model_execution_value != job.model_execution_value: errors.append( f"{job.worker_id}: model execution value does not match invocation assignment" ) if assignment.host_model_value != job.host_model_value: errors.append( f"{job.worker_id}: host model value does not match invocation assignment" ) manifest_path = resolve_project_path(job.project_root, manifest_path_value) verification = verify_agent_invocation( job.prompt_metadata_path, project_root=job.project_root, expected_run_manifest_path=manifest_path, expected_assignment=assignment, expected_invocation_id=job.invocation_id, expected_worker_id=job.worker_id, expected_assignment_ref=job.assignment_ref, expected_audience=job.audience, ) errors.extend(f"{job.worker_id}: {error}" for error in verification) try: metadata = load_json_object(job.prompt_metadata_path, "agent metadata") except DispatchError as exc: errors.append(f"{job.worker_id}: {exc}") continue digests = metadata.get("digests") if not isinstance(digests, Mapping) or dict(digests) != job.digests: errors.append(f"{job.worker_id}: job digests do not match agent metadata") if errors: raise DispatchError("agent invocation contract: " + "; ".join(errors)) def divergent_fields( pairs: Sequence[tuple[str, Any, Any]], *, left: str, right: str, ) -> list[str]: """어긋난 필드만 이름과 두 값으로 적는다. 이 두 검사는 각각 일곱 개와 다섯 개의 조건을 `or` 로 묶어 한 문장으로 보고했다. 그래서 재시도 프롬프트의 `dispatchKind` 하나가 틀린 경우와 provider 가 틀린 경우가 같은 문장으로 나왔고, 읽는 쪽이 어느 값을 봐야 할지 알 수 없었다. 판정에 쓴 값은 이미 손에 있으므로 감출 이유가 없다. """ return [ f"{name}: {left}={found!r} {right}={expected!r}" for name, found, expected in pairs if found != expected ] def _v2_job_authority_errors( job: WorkerJob, metadata: Mapping[str, Any], manifest: Mapping[str, Any], ) -> list[str]: try: identity = invocation_metadata_identity(metadata) except AgentInvocationError as exc: return [str(exc)] if identity is None: return ["jobs file v2 identity has v1 invocation metadata"] errors: list[str] = [] if job.invocation_ref != identity.invocation_ref: errors.append("jobs file invocationRef does not match invocation metadata") divergent = divergent_fields( ( ("participantRef", identity.participant_ref, job.participant_ref), ("roleExecutionRef", identity.role_execution_ref, job.role_execution_ref), ("executionLabel", identity.execution_label, job.execution_label), ("dutyId", identity.duty_id, job.duty_id), ("attempt", identity.attempt, job.attempt), ), left="metadata", right="job", ) if divergent: errors.append( "jobs file v2 identity does not match invocation metadata — " + "; ".join(divergent) ) prompt = metadata.get("prompt") metadata_prompt_value = ( string_value(prompt.get("path")) if isinstance(prompt, Mapping) else "" ) metadata_prompt_path = resolve_project_path( job.project_root, metadata_prompt_value ) if ( not metadata_prompt_value or job.prompt_path.resolve() != metadata_prompt_path.resolve() ): errors.append("jobs file promptPath does not match invocation metadata") role_executions = manifest.get("roleExecutions") execution = next(( row for row in role_executions if isinstance(row, Mapping) and row.get("roleExecutionRef") == identity.role_execution_ref ), None) if isinstance(role_executions, list) else None participants = manifest.get("participantAssignments") participant = next(( row for row in participants if isinstance(row, Mapping) and row.get("participantRef") == identity.participant_ref ), None) if isinstance(participants, list) else None assignment = metadata.get("modelAssignment") try: duty_role = role_for_duty(job.duty_id) except RoleCatalogError: duty_role = "" if isinstance(assignment, Mapping): try: parsed_assignment = agent_model_assignment_from_payload(assignment) except AgentInvocationError as exc: errors.append(str(exc)) else: errors.extend( "jobs file " + error for error in v2_role_assignment_authority_errors( manifest, role_execution_ref=identity.role_execution_ref, participant_ref=identity.participant_ref, assignment=parsed_assignment, ) ) missing = [ name for name, value in ( ("roleExecution", execution), ("participant", participant), ("modelAssignment", assignment), ) if not isinstance(value, Mapping) ] if missing: errors.append( "jobs file v2 identity does not match role execution authority: " f"the run manifest has no {', '.join(missing)} for this job" ) return errors # 어긋난 필드를 이름으로 댄다. 일곱 조건을 한 문장으로 접던 동안, dispatchKind # 하나가 틀린 경우와 provider 가 틀린 경우가 같은 문장으로 보고됐다 — # 재시도 프롬프트를 만들다 `initial-retry` 로 적은 리드가 원인을 좁히지 # 못했다. 판정에 쓴 값은 여기 다 있으므로 굳이 감출 이유가 없다. divergent = divergent_fields( ( ("participantRef", execution.get("participantRef"), job.participant_ref), ("executionLabel", execution.get("executionLabel"), job.execution_label), ("role", execution.get("role"), job.role), ("modelExecutionValue", assignment.get("modelExecutionValue"), job.model_execution_value), ("dispatchKind", metadata.get("dispatchKind"), job.dispatch_kind), ("provider", execution.get("provider"), job.provider), ), left="manifest", right="job", ) if execution.get("role") != duty_role: divergent.append( f"role: roleExecution={execution.get('role')!r} " f"duty {job.duty_id!r} requires {duty_role!r}" ) if divergent: errors.append( "jobs file v2 identity does not match role execution authority — " + "; ".join(divergent) ) return errors def _validate_v2_jobs_file_authority( manifest: Mapping[str, Any], jobs: Sequence[WorkerJob] ) -> None: if ( manifest.get("schemaVersion") != "2.0" or manifest.get("executionIdentityVersion") != 2 ): return _validate_agent_invocations(manifest, jobs) errors: list[str] = [] for job in jobs: metadata = load_json_object(job.prompt_metadata_path, "agent metadata") errors.extend( f"{job.worker_id}: {error}" for error in _v2_job_authority_errors(job, metadata, manifest) ) if errors: raise DispatchError("jobs file v2 authority: " + "; ".join(errors)) role_executions = { string_value(row.get("roleExecutionRef")): row for row in manifest.get("roleExecutions") or [] if isinstance(row, Mapping) } if any( isinstance(role_executions.get(job.role_execution_ref), Mapping) and role_executions[job.role_execution_ref].get("role") == "translator" for job in jobs ): raise DispatchError( "jobs file translator must use canonical reservation dispatch" ) def worker_jobs_from_file( project_root: Path, jobs_file: Path, *, manifest: Mapping[str, Any], active_context: Mapping[str, Any] | None = None, backend: str, idle_timeout_seconds: int | None, default_dispatch_kind: str, resolve_wrapper: Callable[[str], Path], default_provider: Callable[[str], str], ) -> list[WorkerJob]: payload = load_json_object( resolve_project_path(project_root, str(jobs_file)), "jobs file", ) dispatch_kind = string_value(payload.get("dispatchKind")) or default_dispatch_kind workers = payload.get("workers") if not isinstance(workers, list): raise DispatchError("jobs file workers must be an array") jobs = [ _worker_job_from_file( project_root, item, manifest=manifest, active_context=active_context, backend=backend, idle_timeout_seconds=idle_timeout_seconds, dispatch_kind=dispatch_kind, resolve_wrapper=resolve_wrapper, default_provider=default_provider, ) for item in workers if isinstance(item, Mapping) ] _validate_v2_jobs_file_authority(manifest, jobs) return jobs def _worker_job_from_file( project_root: Path, item: Mapping[str, Any], *, manifest: Mapping[str, Any], active_context: Mapping[str, Any] | None = None, backend: str, idle_timeout_seconds: int | None, dispatch_kind: str, resolve_wrapper: Callable[[str], Path], default_provider: Callable[[str], str], ) -> WorkerJob: identity = worker_execution_identity(item) worker_id = ( require_string(item, "workerId") if identity is None else v2_worker_state_key(item) ) provider = ( string_value(item.get("provider")) or default_provider(worker_id) if identity is None else require_string(item, "provider") ) prompt_path = resolve_project_path( project_root, require_string(item, "promptPath") ) worker_result_path = resolve_project_path( project_root, require_string(item, "workerResultPath") ) # The file's own `resultPath` / `completionPaths` are advisory: the same # derivation the roster path applies decides them, so a jobs-file dispatch # and a roster dispatch of one worker cannot disagree about which artifact # is the result. A file that names the report writer's `.md` as its result # used to reach report finalization as the data path. result_path = dispatch_result_path( worker_id, worker_result_path, manifest, project_root ) completion_paths = dispatch_completion_paths( worker_id, worker_result_path, manifest, project_root ) digests = item.get("digests") digest_values = digests if isinstance(digests, Mapping) else {} host_model_value = item.get("hostModelValue") if host_model_value is not None and not isinstance(host_model_value, str): raise DispatchError("jobs file hostModelValue must be a string or null") return WorkerJob( worker_id=worker_id, provider=provider, backend=backend, project_root=project_root, model_execution_value=require_string(item, "modelExecutionValue"), wrapper_path=resolve_wrapper(provider), prompt_path=prompt_path, result_path=result_path, worker_result_path=worker_result_path, completion_paths=completion_paths, # 파일이 적지 않았으면 run authority 에서 해소한다. 로스터 경로는 # 언제나 그렇게 하고(`dispatch_core._jobs_from_roster`), 예약 쪽 # (`agent_prompt_cli._run_worktree`)도 같은 seam 을 읽는다. 이 자리만 # 손으로 적은 값을 유일한 출처로 삼던 동안, 같은 run 이 어떤 경로로 # 디스패치됐는지에 따라 워커의 소스 루트가 달라졌다 — 그 값은 # writePolicy 의 `sourcePolicy.allowedRoot` 이므로, 예약본과 갈리면 # 같은 invocationRef 가 `invocationRef drift` 로 거부된다. worktree_path=( string_value(item.get("worktreePath")) or worktree_path(manifest, active_context or {}) ), role=require_string(item, "role"), idle_timeout_seconds=idle_timeout_seconds, dispatch_kind=dispatch_kind, # Minted here, not read from the file: the lead writes this file by # hand, and a session id it typed would be unfalsifiable — nothing can # tell an id that names no session from one that names the wrong one. session_id=dispatch_session_id(provider), invocation_id=string_value(item.get("invocationId")), audience=string_value(item.get("audience")), assignment_ref=string_value(item.get("assignmentRef")), prompt_metadata_path=resolve_project_path( project_root, string_value(item.get("promptMetadataPath")), ) if string_value(item.get("promptMetadataPath")) else Path(), catalog_digest=string_value(digest_values.get("catalogDigest")), assignment_digest=string_value(digest_values.get("assignmentDigest")), duty_digest=string_value(digest_values.get("dutyDigest")), instruction_digest=string_value(digest_values.get("instructionDigest")), prompt_digest=string_value(digest_values.get("promptDigest")), host_model_value=host_model_value, enforcement_mode=string_value(item.get("enforcementMode")), **(identity or {}), ) def worker_execution_identity( item: Mapping[str, Any], ) -> dict[str, Any] | None: identity_keys = { "executionIdentityVersion", "participantRef", "roleExecutionRef", "executionLabel", "dutyId", "invocationRef", } schema = item.get("schemaVersion") has_v2_fields = bool(identity_keys.intersection(item)) if schema in (None, 1, "1", "1.0"): if has_v2_fields: raise DispatchError("jobs file worker mixes v1 and v2 identity") return None if schema != "2.0" or item.get("executionIdentityVersion") != 2: raise DispatchError("jobs file worker has unsupported identity version") if "workerId" in item: raise DispatchError("jobs file worker mixes v1 and v2 identity") attempt = item.get("attempt") if not isinstance(attempt, int) or isinstance(attempt, bool) or attempt < 1: raise DispatchError("jobs file v2 attempt must be positive") return { "participant_ref": require_string(item, "participantRef"), "role_execution_ref": require_string(item, "roleExecutionRef"), "execution_label": require_string(item, "executionLabel"), "duty_id": require_string(item, "dutyId"), "invocation_ref": require_string(item, "invocationRef"), "attempt": attempt, } def v2_worker_state_key(item: Mapping[str, Any]) -> str: assignment_ref = require_string(item, "assignmentRef") worker_id = assignment_ref.rsplit("/", 1)[-1] if not worker_id: raise DispatchError("jobs file v2 assignmentRef has no worker projection") return worker_id def utc_now() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def dispatch_record_matches(record: object, prompt_path: Path, attempt: int) -> bool: """Whether this team-state row is the dispatch handed `prompt_path`. The prompt file identifies a dispatch: each round writes its own, and a second run of the same one is a retry, which `attempt` separates. Worker id and dispatch kind cannot carry that — `--dispatch-kind` is the lead's to pick and it has reused one across two rounds, which put the finished round's completion on the older namesake row and left the row that actually ran marked running, so an await re-collected that worker forever. """ return ( isinstance(record, dict) and record.get("promptPath") == str(prompt_path) and record.get("attempt") == attempt )