"""Post-hoc checks for neutral lead lifecycle and live-activity contracts. Design: docs/superpowers/specs/2026-06-10-blocking-contract-posthoc-conformance-design.md | Check | Contract | Evidence | |-------|----------|----------| | 1. lead PROGRESS checkpoints | prompts/lead/okstra-lead-contract.md "Progress reporting (BLOCKING)" | selected adapter evidence source | | 2. worker-provider heartbeat | worker provider contract | audit sidecar `- PROGRESS: ` lines | | 3. implementation entry guard | prompts/lead/okstra-lead-contract.md "Entry guard (BLOCKING)" + prompts/launch.template.md "Host Orchestration Rules" (only when the run staged the rules file) | selected adapter evidence source | Evidence rules prevent false passes: - `claude-jsonl` accepts only assistant text and file-read tool-use records (a `Read` call, or a shell command naming the file), excluding injected skill text and sidechain records. - Session evidence is scoped to the current run window so a previous run cannot satisfy the neutral lead lifecycle contract. - Checkpoints emitted after validation begins (`phase-7-teardown`, `complete`) cannot be observed and are not required. """ from __future__ import annotations import json import os import re import sys from dataclasses import dataclass, field from datetime import datetime from pathlib import Path from typing import Any, Mapping, NamedTuple # scripts/ (repo) and python/ (installed under ~/.okstra/lib) are not packages; # insert whichever exists so okstra_ctl is importable directly. _VALIDATORS_DIR = Path(__file__).resolve().parent for _ssot_dir in (_VALIDATORS_DIR.parent / "scripts", _VALIDATORS_DIR.parent / "python"): if _ssot_dir.is_dir() and str(_ssot_dir) not in sys.path: sys.path.insert(0, str(_ssot_dir)) from okstra_ctl.paths import RunRef # noqa: E402 from okstra_ctl.worker_heartbeat import ( # noqa: E402 HEARTBEAT_LINE_RE, IN_STAGE_PREFIX, max_gap_seconds_after, ) from okstra_ctl.lead_events import ( # noqa: E402 LeadEvent, LeadEventParseError, read_lead_events, ) from okstra_ctl.agent.activity import ACTIVITY_FIELDS # noqa: E402 from okstra_ctl.domain.host import HostNotRegistered # noqa: E402 from okstra_ctl.final_report_paths import final_report_data_path # noqa: E402 from okstra_ctl.registry.host_registry import default_host_registry # noqa: E402 from okstra_ctl.wrapper_status import read_wrapper_status # noqa: E402 _DISPATCHED_STATUSES = {"completed", "timeout", "error", "in-progress"} _ATTEMPTED_STATUSES = {"completed", "timeout", "error"} _LIVENESS_AUDIT_HEARTBEAT = "audit-heartbeat" _LIVENESS_WRAPPER_STATUS = "wrapper-status" _REPORT_WRITER_HEARTBEAT_STAGES = { "started", "required-reading-complete", "synthesis-start", "data-json-write-start", "render-start", "write-result-start", } # lead 의 체크포인트 라인 — assistant text 블록 안에서 line-anchored 로만 인정. # lead 는 계약상 raw 로 emit 해야 하지만 실측(dev-9902)에서 `` `PROGRESS: ...` `` # 인라인 코드나 코드펜스로 감싸 emit 하는 경우가 있어, 라인 양끝의 backtick 과 # 선행 들여쓰기를 허용한다. phase 는 backtick 을 포함하지 않는다. _PROGRESS_LINE_RE = re.compile( r"^[ \t]*`*[ \t]*PROGRESS:[ \t]+(?P[^\s`]+)(?P[^`\n]*)`*[ \t]*$", re.MULTILINE, ) _ACTIVITY_ID_RE = re.compile(r"^A-(\d{3,})$") _TARGETED_REVERIFICATION_REF_RE = re.compile( r"^plan-body-verification:round-(\d+)$" ) _CLARIFICATION_ID_RE = re.compile(r"^C-\d{3,}$") _WORKER_FIELD_RE = re.compile( r"(?:^|\s)worker=(?P.+?)(?=\s+[a-zA-Z][a-zA-Z0-9_.-]*=|$)" ) _COMPLETED_STATUS_RE = re.compile(r"(?:^|\s)status=completed(?:\s|$)") # heartbeat 라인 shape 과 cadence 예산은 okstra_ctl.worker_heartbeat 정본을 쓴다 — # `okstra worker-liveness` 가 run 도중 같은 판정을 내리므로 정의가 갈리면 안 된다. _HEARTBEAT_LINE_RE = HEARTBEAT_LINE_RE _IN_STAGE_PREFIX = IN_STAGE_PREFIX _LEAD_CONTRACT_CITE = "prompts/lead/okstra-lead-contract.md 'Entry guard (BLOCKING)'" _HOST_RULES_CITE = "prompts/launch.template.md 'Host Orchestration Rules'" _CONVERGENCE_CITE = "prompts/lead/convergence.md" _PLAN_BODY_CITE = "prompts/lead/plan-body-verification.md" class _EntryGuardRead(NamedTuple): """implementation entry guard 가 Read 를 요구하는 파일 한 건.""" basename: str # (이 Read 가 앞서야 할 PROGRESS 체크포인트, 실패 메시지에 쓸 시점 표현). # None 이면 순서 제약이 없고 run 윈도우 안의 존재만 검사한다. anchor: tuple[str, str] | None cite: str # True 면 run 이 이 파일을 instruction-set 에 스테이징했을 때만 요구한다. staged_per_run: bool = False # 이 행이 도는 task-type. None 이면 task-type 을 가리지 않는다. task_types: frozenset[str] | None = None # 이 PROGRESS phase 가 run 에 실제로 나타났을 때만 요구한다. None 이면 # 무조건 요구한다. task-type 목록을 손으로 적는 것보다 정확하다 — 수렴은 # 여러 phase 에서 돌고, 어느 run 이 실제로 돌렸는지는 그 run 의 체크포인트가 # 말해 준다. required_when: str | None = None _IMPLEMENTATION_ONLY = frozenset({"implementation"}) _ENTRY_GUARD_READS = ( _EntryGuardRead( "_implementation-executor.md", ("phase-6-synthesis", "at Phase 5"), _LEAD_CONTRACT_CITE, task_types=_IMPLEMENTATION_ONLY, ), _EntryGuardRead( "_implementation-verifier.md", ("phase-6-synthesis", "at Phase 5"), _LEAD_CONTRACT_CITE, task_types=_IMPLEMENTATION_ONLY, ), _EntryGuardRead( "_implementation-deliverable.md", ("phase-7-persist", "at Phase 6"), _LEAD_CONTRACT_CITE, task_types=_IMPLEMENTATION_ONLY, ), # lead 프로파일 sidecar 가 아니라, run 이 instruction-set 에 스테이징한 host # orchestration 규칙이다. 세 가지가 sidecar 와 다르다. # # anchor 없음: 이 파일은 `render-bundle` 이 만든다(run.py `_write_instruction_set_sources`). # 그 앞에서 발화하는 gate(Step 5.1 waiver 제안)에는 아직 파일이 없고, 뒤에 # 오는 gate 는 run 의 PROGRESS 체크포인트와 시간 관계가 고정돼 있지 않다. # 증거가 뒷받침하는 주장은 "run 윈도우 안에 Read 가 최소 한 번" 뿐이다. # staged_per_run: 파일이 없는 번들 — 이 변경 전에 렌더된 run, 또는 payload 에 # prompts/host-orchestration/.md 가 없는 설치본 — 은 읽을 수 없는 # 파일 때문에 막히면 안 된다. 규칙이 없는 task-type 에 conformance 가 규칙을 # 요구하지 않는다는 원칙과 같은 판단이다. _EntryGuardRead( "host-orchestration-rules.md", None, _HOST_RULES_CITE, staged_per_run=True, task_types=_IMPLEMENTATION_ONLY, ), # 수렴과 계획-본문 검증은 규칙 문서가 lazy 로 남는다 — Phase 1 intake 가 # 읽지 않으므로, 읽지 않은 lead 는 그 라운드를 자기 기억으로 돌린다. 그 # 두 문서에 lead 대상 MUST 의 절반 이상이 들어 있다. # # task-type 이 아니라 체크포인트로 조건을 건다: 라운드를 실제로 돌린 run 만 # 그 문서를 요구받는다. _EntryGuardRead( "convergence.md", ("phase-5.5-convergence", "before the first convergence round"), _CONVERGENCE_CITE, required_when="phase-5.5-convergence", ), _EntryGuardRead( "plan-body-verification.md", ("phase-5.5.9-plan-verify", "before the first plan-verify round"), _PLAN_BODY_CITE, required_when="phase-5.5.9-plan-verify", ), ) # 환경으로 선택되는 어댑터라 lead 의 런타임이 이 파일을 가르쳐 주지 않는다. # 읽지 않은 lead 는 자기 런타임이 아는 방식 — cmux 경로에서는 okstra 가 소유한 # 디스패치를 host 네이티브로 가로채는 방식 — 으로 되돌아간다. CMUX_ADAPTER_BASENAME = "cmux.md" CMUX_ADAPTER_NAME = "cmux" _CMUX_ADAPTER_CITE = "prompts/lead/adapters/cmux.md" # Read 증거는 basename 으로 거른다 — 절대 경로는 레이어(repo / runtime / 설치본) # 마다 다르지만 basename 은 동일하다. 목록이 갈리지 않도록 기대치에서 파생한다. _TRACKED_READ_BASENAMES = ( *(row.basename for row in _ENTRY_GUARD_READS), CMUX_ADAPTER_BASENAME, ) # 셸 read 도 증거다. 계약이 요구하는 것은 "이 run 안에서 그 파일의 내용을 실제로 # 적재했는가" 이고, `cat` / `sed -n` 는 Read 도구와 동일하게 그것을 만족한다. # 도구 이름만 인정하면, 호스트가 "가능하면 Bash 로 파일을 읽으라"고 지시하는 모드 # (Claude Code auto 모드)에서는 계약을 그대로 이행한 lead 가 매번 # `contract-violated` 로 끝난다 — 두 지시가 정면으로 충돌한다. # # basename 이 경로 토큰으로 등장할 때만 인정한다. 트리 전체를 훑는 명령 # (`grep -rn "..." prompts/`)은 그 파일을 지목하지 않았으므로 매칭되지 않고, # 파일을 인자로 준 명령은 매칭된다 — 후자는 실제로 그 파일을 읽는다. _SHELL_READ_TOOL_NAME = "Bash" _SHELL_READ_PATTERNS = { base: re.compile(rf"(?:^|[\s'\"`=(/]){re.escape(base)}(?:$|[\s'\"`);:,])") for base in _TRACKED_READ_BASENAMES } def _read_evidence_basenames(block: dict) -> tuple[str, ...]: """tool_use 블록 하나가 증거로 인정되는 tracked basename 들.""" name = block.get("name") payload = block.get("input") or {} if name == "Read": base = Path(str(payload.get("file_path") or "")).name return (base,) if base in _TRACKED_READ_BASENAMES else () if name == _SHELL_READ_TOOL_NAME: command = str(payload.get("command") or "") if not command: return () return tuple( base for base, pattern in _SHELL_READ_PATTERNS.items() if pattern.search(command) ) return () @dataclass class SessionConformanceResult: errors: list[str] = field(default_factory=list) # Conformance findings that are reported without failing the run: the # PROGRESS narration lines. A missing checkpoint line means the run is # harder to follow after the fact, not that any of its work is wrong, and # by the time this validator runs the line can no longer be emitted — the # session that would have written it has ended. Checks that verify a # *claim* (a user confirmation the report says it obtained) stay in # `errors`. advisories: list[str] = field(default_factory=list) @property def ok(self) -> bool: return not self.errors @dataclass class _LeadEvidence: progress: list[tuple[str, str, str]] = field(default_factory=list) # (ts, phase-id, line) sidecar_reads: dict[str, list[str]] = field(default_factory=dict) # basename -> [ts] activities: list[LeadEvent] = field(default_factory=list) scanned_files: list[Path] = field(default_factory=list) window: tuple[str | None, str | None] = (None, None) def _ensure_token_usage_importable() -> None: """okstra_token_usage 패키지를 레이아웃별(repo/scripts, runtime/python, OKSTRA_PYTHONPATH)로 해소 — validate-run.py `_import_token_usage` 와 동일 후보.""" here = Path(__file__).resolve().parent candidates = [here.parent / "scripts", here.parent / "python"] env_pp = os.environ.get("OKSTRA_PYTHONPATH", "").strip() if env_pp: candidates.append(Path(env_pp)) for candidate in candidates: if candidate.is_dir() and (candidate / "okstra_token_usage").is_dir(): if str(candidate) not in sys.path: sys.path.insert(0, str(candidate)) break def _norm(value: str) -> str: return re.sub(r"[^a-z0-9]", "", (value or "").lower()) def _is_report_writer(worker: dict) -> bool: return "reportwriter" in _norm(str(worker.get("role", ""))) or "reportwriter" in _norm( str(worker.get("workerId", "")) ) def _worker_role(value: str) -> str | None: """역할 필드 전체에서 Unicode 대소문자 차이만 제거한다.""" return value.casefold() if value else None def _worker_roles(worker: dict) -> set[str]: """명단 `role` 과 `workerId` 를 활동 `agent` 가 쓸 수 있는 별칭으로 펼친다. 활동은 `codex` / `codex-worker` 를 쓰고 명단은 `Codex worker` 를 쓴다. 역할 문자열만 보면 둘은 다른 사람이 된다. """ roles: set[str] = set() for raw in (worker.get("role"), worker.get("workerId")): role = _worker_role(str(raw or "")) if not role: continue compact = role.replace(" ", "-") roles.add(role) roles.add(compact) if role.endswith(" worker"): roles.add(role[: -len(" worker")]) if compact.endswith("-worker"): roles.add(compact[: -len("-worker")]) return roles def _analysis_workers(team_state: Mapping[str, Any]) -> list[dict]: workers = [ worker for worker in (team_state.get("workers") or []) if isinstance(worker, dict) ] return [worker for worker in workers if not _is_report_writer(worker)] def _scan_one_jsonl( path: Path, since: str | None, until: str | None ) -> tuple[list[tuple[str, str, str]], dict[str, list[str]], str | None]: """jsonl 한 파일에서 (progress, sidecar reads, agentName) 을 추출한다.""" from okstra_token_usage.paths import ts_in_window progress: list[tuple[str, str, str]] = [] reads: dict[str, list[str]] = {} agent_name: str | None = None try: fh = path.open(encoding="utf-8") except OSError: return progress, reads, agent_name with fh: for raw in fh: try: rec = json.loads(raw) except (json.JSONDecodeError, UnicodeDecodeError): continue if agent_name is None and rec.get("agentName"): agent_name = rec["agentName"] if rec.get("type") != "assistant" or rec.get("isSidechain"): continue ts = rec.get("timestamp") or "" if ts and not ts_in_window(ts, since, until): continue msg = rec.get("message") or {} for block in msg.get("content") or []: if not isinstance(block, dict): continue if block.get("type") == "text": for m in _PROGRESS_LINE_RE.finditer(block.get("text") or ""): line = f"PROGRESS: {m.group('phase')}{m.group('rest')}".rstrip() progress.append((ts, m.group("phase"), line)) elif block.get("type") == "tool_use": for base in _read_evidence_basenames(block): reads.setdefault(base, []).append(ts) return progress, reads, agent_name def _collect_lead_evidence( team_state: dict, team_state_path: Path, run_manifest: Mapping[str, Any], project_root: Path, task_type: str, suffix: str | None, projects_root: Path | None, ) -> tuple[_LeadEvidence | None, str | None]: """lead 후보 jsonl 을 스캔해 증거를 모은다. 후보 = {기록된 lead.sessionId} ∪ {team 태그는 있으나 agentName 이 없는 세션}. 후자는 `claude --resume` 으로 lead 세션이 fork 된 경우(새 sessionId, agentName 없음)를 흡수한다 — worker 세션은 agentName 이 있어 자연 배제된다. """ from okstra_token_usage.claude import find_claude_team_sessions from okstra_token_usage.collect import ( resolve_run_window, resolve_team_needles_with_source, ) from okstra_token_usage.paths import claude_project_dir, find_session_jsonl since, until = resolve_run_window(team_state_path, team_state, relax_start=False) lead_sid = (team_state.get("lead") or {}).get("sessionId") or "" team_needles, _needle_source = resolve_team_needles_with_source( team_state, project_root, since, until, projects_root=projects_root ) sessions = find_claude_team_sessions( project_root, team_needles, lead_sid, projects_root=projects_root ) # `claude --resume` 로 fork 되거나 leadSessionIds[] 에 기록된 lead 세대는 # team 태그 needle 로 안 잡힐 수 있어 명시적으로 후보에 추가한다. # # 기록된 id 는 프로젝트 디렉터리 밖에 있을 수 있다 — 리드 세션의 cwd 가 # 대상 프로젝트 루트와 다르면(`okstra preflight --cwd <다른 프로젝트>`) # transcript 는 리드 cwd 로 인코딩된 디렉터리에 놓인다. 그래서 id 로 찾는다. proj_dir = claude_project_dir(project_root, projects_root) recorded_ids = [lead_sid, *(team_state.get("leadSessionIds") or [])] for sid in recorded_ids: if not isinstance(sid, str) or not sid: continue candidate = find_session_jsonl(sid, project_root, projects_root) if candidate is not None: sessions.setdefault(sid, candidate) evidence = _LeadEvidence(window=(since, until)) for sid, path in sorted(sessions.items()): progress, reads, agent_name = _scan_one_jsonl(path, since, until) if agent_name and sid != lead_sid: continue # agentName 이 찍힌 세션은 worker — lead 후보에서 제외 evidence.scanned_files.append(path) evidence.progress.extend(progress) for base, ts_list in reads.items(): evidence.sidecar_reads.setdefault(base, []).extend(ts_list) if not evidence.scanned_files: return None, ( f"lead session jsonl not found under {proj_dir} nor anywhere under " f"{(projects_root or proj_dir.parent)} by recorded session id " f"(lead.sessionId={lead_sid or ''}) — selected adapter evidence " "source `claude-jsonl` cannot verify the PROGRESS checkpoint / " "implementation entry-guard conformance cannot be verified, which " "fails the run (same principle as the token-usage accuracy contract)." ) evidence.progress.sort() for ts_list in evidence.sidecar_reads.values(): ts_list.sort() if _is_activity_contract_v1_planning(run_manifest): activities, events_path, error = _read_scoped_lead_activities( team_state, run_manifest, project_root, task_type, suffix ) if error: return None, error evidence.activities.extend(activities) if events_path is not None: evidence.scanned_files.append(events_path) return evidence, None def _resolve_lead_events_path( team_state: Mapping[str, Any], run_manifest: Mapping[str, Any], project_root: Path, ) -> tuple[Path | None, str | None]: raw = run_manifest.get("leadEventsPath") or team_state.get("leadEventsPath") or ( (team_state.get("artifacts") or {}).get("leadEventsPath") if isinstance(team_state.get("artifacts"), dict) else "" ) if not raw: return None, ( "artifact lead event log path missing from team-state " "(`leadEventsPath`) — selected adapter evidence source `artifact-only` " "cannot verify conformance." ) path = Path(str(raw)) if not path.is_absolute(): path = project_root / path if not path.is_file(): return None, ( f"artifact lead event log not found: {path} — selected adapter evidence " "source `artifact-only` cannot verify conformance." ) return path, None def _event_matches_run( event: LeadEvent, team_state: Mapping[str, Any], run_manifest: Mapping[str, Any], task_type: str, run_seq: str, ) -> bool: task_key = str(run_manifest.get("taskKey") or team_state.get("taskKey") or "") expected_runtime = str( run_manifest.get("leadRuntime") or team_state.get("leadRuntime") or "claude-code" ) if event.lead_runtime != expected_runtime: return False if task_key and event.task_key != task_key: return False return event.task_type == task_type and event.run_seq == run_seq def _run_sequence(run_manifest: Mapping[str, Any], suffix: str | None) -> str: sequences = run_manifest.get("runSequencesByCategory") if isinstance(sequences, Mapping): value = sequences.get("manifests") if isinstance(value, str) and value: return value if suffix and "-" in suffix: return suffix.rsplit("-", 1)[1] return "" def _read_scoped_lead_activities( team_state: Mapping[str, Any], run_manifest: Mapping[str, Any], project_root: Path, task_type: str, suffix: str | None, ) -> tuple[list[LeadEvent], Path | None, str | None]: run_seq = _run_sequence(run_manifest, suffix) if not run_seq: return [], None, ( "activity contract cannot scope lead events because the run sequence " "is missing from the run manifest and team-state filename." ) events_path, error = _resolve_lead_events_path( team_state, run_manifest, project_root ) if error: return [], None, error try: events = read_lead_events(events_path) except LeadEventParseError as exc: return [], events_path, ( "artifact lead event log is malformed — selected adapter evidence " f"source cannot verify conformance: {exc}" ) activities = [ event for event in events if event.event_type == "activity" and _event_matches_run( event, team_state, run_manifest, task_type, run_seq ) ] return activities, events_path, None # 리드가 `progress` 행을 빠뜨려도 같은 사실이 activity 원장에 남아 있는 # 체크포인트. 값은 `okstra agent-activity append --kind` 가 받는 것과 같고, # 왼쪽이 그 kind, 오른쪽이 그것이 증명하는 PROGRESS phase 다. _ACTIVITY_PROGRESS_PHASE = { "worker-dispatched": "phase-4-dispatch", "worker-completed": "phase-5-collect", "verification-round-completed": "phase-5.5-convergence", } def _progress_from_activity(event) -> tuple[str, str, str] | None: """activity 한 행이 증명하는 PROGRESS 체크포인트. `PROGRESS:` 는 사용자가 보는 대화 텍스트이고, artifact-only 호스트에서는 리드가 그것을 `leadEventsPath` 에 손으로 append 해야 검증에 잡힌다. 리드가 그 행을 빠뜨리면 워커를 실제로 띄우고 거둔 run 도 체크포인트 전건 누락으로 보고됐다 — 그 사실이 `okstra agent-activity append` 가 쓴 activity 행에 이미 기록돼 있는데도(실측: dev-10784 error-analysis, advisory 22건 중 9건이 activity 원장으로 확인 가능한 항목이었다). 원장에 있는 것은 원장에서 읽는다. 나머지 체크포인트(intake·prompts·team-create·synthesis·persist)는 대응하는 activity kind 가 없다. 그것들은 여전히 리드가 남겨야 하고, 없으면 없다고 보고된다. """ details = event.details phase = _ACTIVITY_PROGRESS_PHASE.get(str(details.get("kind") or "")) if phase is None: return None agent = str(details.get("agent") or "").strip() if not agent or agent == "okstra-lead": # 리드 자신을 주체로 적은 행은 워커를 지목하지 않는다. 워커별 체크포인트 # 검사는 이름을 대조하므로, 이름 없는 행으로 그 검사를 만족시키지 않는다. return (event.timestamp, phase, f"PROGRESS: {phase}") return (event.timestamp, phase, f"PROGRESS: {phase} worker={agent}") def _progress_line_from_event(event) -> tuple[str, str, str] | None: details = event.details phase = details.get("phase") if not isinstance(phase, str) or not phase: line_candidate = details.get("line") if isinstance(line_candidate, str): match = _PROGRESS_LINE_RE.search(line_candidate) if match: phase = match.group("phase") if not isinstance(phase, str) or not phase: return None line = details.get("line") if not isinstance(line, str) or not line: message = details.get("message") tail = f" {message}" if isinstance(message, str) and message else "" line = f"PROGRESS: {phase}{tail}" worker = details.get("worker") if isinstance(worker, str) and worker.strip(): from okstra_ctl.lead_progress import render_progress_line line = render_progress_line(phase, [("worker", worker), ("detail", line)]) return (event.timestamp, phase, line) def _sidecar_read_from_event(event) -> tuple[str, str] | None: details = event.details if event.event_type != "sidecar-read" and details.get("kind") != "implementation-sidecar": return None basename = details.get("basename") if not isinstance(basename, str) or not basename: raw_path = details.get("path") or details.get("filePath") basename = Path(str(raw_path or "")).name if basename not in _TRACKED_READ_BASENAMES: return None return (basename, event.timestamp) def _collect_artifact_lead_evidence( team_state: dict, run_manifest: Mapping[str, Any], project_root: Path, task_type: str, suffix: str | None, ) -> tuple[_LeadEvidence | None, str | None]: if not suffix or "-" not in suffix: return None, ( "artifact lead event log cannot be scoped because team-state filename " "does not expose a run artifact suffix." ) events_path, error = _resolve_lead_events_path( team_state, run_manifest, project_root ) if error: return None, error try: events = read_lead_events(events_path) except LeadEventParseError as exc: return None, ( "artifact lead event log is malformed — selected adapter evidence " f"source `artifact-only` cannot verify conformance: {exc}" ) run_seq = _run_sequence(run_manifest, suffix) evidence = _LeadEvidence(scanned_files=[events_path]) from_activities: list[tuple[str, str, str]] = [] for event in events: if not _event_matches_run( event, team_state, run_manifest, task_type, run_seq ): continue if event.event_type == "activity": evidence.activities.append(event) derived = _progress_from_activity(event) if derived is not None: from_activities.append(derived) elif event.event_type in ("progress", "progress-checkpoint"): progress = _progress_line_from_event(event) if progress is not None: evidence.progress.append(progress) elif event.event_type in ("artifact-read", "sidecar-read"): read = _sidecar_read_from_event(event) if read is not None: basename, timestamp = read evidence.sidecar_reads.setdefault(basename, []).append(timestamp) if not evidence.progress: # 리드가 `progress` 행을 하나도 남기지 않은 run 에서만 원장으로 대신 # 읽는다. 한 줄이라도 남긴 run 에서는 그 줄들이 검사 대상이다 — 거기서 # 원장을 섞으면 "PROGRESS 라인의 `worker=` 가 로스터 역할과 일치하는가" # 라는 서술 정확성 검사가 대조할 것을 잃는다. evidence.progress.extend(from_activities) evidence.progress.sort() for ts_list in evidence.sidecar_reads.values(): ts_list.sort() return evidence, None def _conformance_evidence_source( team_state: dict, ) -> tuple[str | None, str | None]: """리드의 증거가 어디 있는지 — 리드 회계가 정한다. `dispatchMode` 는 **워커**를 어떻게 띄우는지를 말하는 필드다. 그 값이 리드 세션의 전사 여부를 바꾸지는 않는다: cmux 판으로 워커를 내보낸 Claude Code 리드도 자기 전사에 Read 와 PROGRESS 를 그대로 남긴다. 어댑터 자신도 그렇게 선언한다 — `prompts/lead/adapters/cmux.md:18` 의 `sessionAccounting | unchanged`. 종전에는 이 함수가 어댑터 메타데이터를 보기 전에 `dispatchMode` 로 단락해 `artifact-only` 를 확정했다. 그러면 수집기가 `lead-events-*.jsonl` 만 보는데 거기에는 `okstra agent-activity append` 가 쓴 활동 레코드만 들어가고 리드 툴셋에는 PROGRESS 나 파일 읽기를 그리로 보낼 경로가 없다 — 실측(`fontsninja-nlpvibe` stage 10): artifact-only 로 progress 0·reads 0, 같은 run 을 리드 전사로 읽으면 progress 21·reads 5(entry guard 가 요구하는 네 파일 전부). 즉 어떤 리드가 무엇을 해도 통과할 수 없는 검사였다. """ adapter = team_state.get("leadAdapter") if isinstance(adapter, dict): accounting = str(adapter.get("sessionAccounting", "")).strip() if accounting in {"claude-jsonl", "artifact-only"}: return accounting, None if accounting: return None, f"unsupported sessionAccounting: `{accounting}`" legacy_runtime = str(team_state.get("leadRuntime", "") or "claude-code") try: accounting = default_host_registry().resolve(legacy_runtime).descriptor.session_accounting except HostNotRegistered: accounting = "claude-jsonl" if accounting == "artifact-only": return "artifact-only", None return "claude-jsonl", None def _convergence_rounds_ran(run_dir: Path, suffix: str | None) -> bool: """이 run 의 convergence state artifact 가 실제 round 를 1회 이상 돌았는지. auto-disable(`totalRounds: 0`)·artifact 부재는 phase-5.5 라인을 요구하지 않는다.""" if not suffix: return False path = run_dir / "state" / f"convergence-{suffix}.json" try: doc = json.loads(path.read_text()) except (OSError, json.JSONDecodeError): return False return isinstance(doc, dict) and (doc.get("totalRounds") or 0) >= 1 def _verification_error_workers(run_dir: Path, suffix: str | None) -> set[str]: """이 run 의 convergence 에서 `verification-error` 를 낸 워커들.""" if not suffix: return set() path = run_dir / "state" / f"convergence-{suffix}.json" try: doc = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return set() if not isinstance(doc, dict): return set() workers: set[str] = set() for finding in doc.get("findings") or []: if not isinstance(finding, dict): continue for entry in finding.get("rounds") or []: if not isinstance(entry, dict): continue for worker, vote in (entry.get("votes") or {}).items(): if isinstance(vote, dict) and vote.get("verdict") == "verification-error": workers.add(str(worker)) return workers def _contract_violation_agents(run_dir: Path, suffix: str | None) -> set[str]: """run 에러 로그에 `contract-violation` 으로 기록된 agent 들.""" if not suffix: return set() path = run_dir / "logs" / f"errors-{suffix}.jsonl" try: raw = path.read_text(encoding="utf-8") except OSError: return set() agents: set[str] = set() for line in raw.splitlines(): line = line.strip() if not line: continue try: row = json.loads(line) except json.JSONDecodeError: continue if isinstance(row, dict) and row.get("errorType") == "contract-violation": agent = row.get("agent") or row.get("workerId") if agent: agents.add(str(agent)) return agents def _check_verification_errors_are_logged( run_dir: Path, suffix: str | None, advisories: list[str] ) -> None: """`verification-error` 를 낸 워커는 에러 로그에도 남아야 한다. 계약(okstra-lead-contract.md §"Phase 5.5: Convergence loop")은 재검증 배치가 `verification-error` 로 끝나면 lead 가 위반마다 한 건씩 `okstra error-log append-observed --error-type contract-violation` 으로 기록하라고 요구한다. 기록이 없으면 워커 실패가 감사 흔적 없이 지나가고, 그 run 의 합의는 실제보다 적은 검증 위에 서 있다. advisory 로 낸다. 이 파일의 구분을 따른다 — 검증기가 도는 시점에는 그 기록을 더 이상 쓸 수 없고(세션이 끝났다), 빠진 기록은 run 을 나중에 따라가기 어렵게 만들지 그 run 의 작업이 틀렸다는 뜻은 아니다. """ failed = _verification_error_workers(run_dir, suffix) if not failed: return logged = _contract_violation_agents(run_dir, suffix) missing = sorted(failed - logged) if missing: advisories.append( "convergence verification-error not recorded in the run error log for: " + ", ".join(missing) + " — prompts/lead/okstra-lead-contract.md 'Phase 5.5: Convergence loop' " "requires one `okstra error-log append-observed --error-type " "contract-violation` event per violation." ) def _plan_body_rounds_ran(run_dir: Path, suffix: str | None) -> int: """이 run 의 plan-body 검증이 실제로 돈 라운드 수. 각 라운드는 새 워커 배치를 띄우므로 라운드마다 배치 경계가 하나씩 생긴다. Phase 6 이후에 도는 구간이라 기존 배치-정리 강제(수렴 라운드 1 직전 / report-writer 디스패치 직전)의 바깥이었고, 그래서 self-fix 를 여러 라운드 돈 run 은 매 라운드의 완료 워커가 그대로 남았다.""" if not suffix: return 0 try: doc = json.loads(_plan_body_state_path(run_dir, suffix).read_text()) except (OSError, json.JSONDecodeError): return 0 if not isinstance(doc, dict): return 0 # 계약은 이 값을 중첩 `planBodyVerification`(자가수정 루프가 끝난 뒤의 최종 # 상태)에 둔다. 최상위는 라운드별 이력이라 라운드 수의 자리가 아니다. projection = doc.get("planBodyVerification") rounds = projection.get("roundCount") if isinstance(projection, dict) else None return rounds if isinstance(rounds, int) and rounds > 0 else 0 def _plan_body_state_path(run_dir: Path, suffix: str) -> Path: """이 run 의 plan-body 상태 파일. `suffix` 는 team-state 이름에서 나온 **state** seq 다. `okstra_ctl/paths.py` 가 이 파일을 같은 seq 로 정의하므로 둘이 같은 곳을 가리킨다 — 리드가 리포트 seq 로 쓰면 여기서 못 찾고, 라운드 수가 0 으로 읽힌다. """ return run_dir / "state" / f"plan-body-verification-{suffix}.json" def _ids_reported_as_asked(report_path: Path) -> list[str]: """리포트가 "사용자에게 물었다"고 기록한 열린 승인 차단 행의 id.""" doc = _read_report_data(report_path) asked: list[str] = [] for row in doc.get("clarificationItems") or []: if not isinstance(row, dict): continue if row.get("blocks") != "approval" or row.get("status") != "open": continue if not str(row.get("userConfirmation") or "").startswith("asked-"): continue row_id = str(row.get("id") or "").strip() if row_id: asked.append(row_id) return asked def _is_activity_contract_v1_planning( run_manifest: Mapping[str, Any], ) -> bool: return ( run_manifest.get("activityContractVersion") == 1 and run_manifest.get("taskType") == "implementation-planning" ) def _activity_index(events: list[LeadEvent]) -> dict[str, list[LeadEvent]]: indexed: dict[str, list[LeadEvent]] = {} for event in events: kind = str(event.details.get("kind") or "") indexed.setdefault(kind, []).append(event) return indexed def _read_report_data(report_path: Path) -> Mapping[str, Any]: """`--report` 가 data.json · markdown · html 이어도 같은 레코드를 연다. Phase 7 는 data.json 을 넘긴다. `.md` 만 받던 동안 투영과 self-fix 횟수가 빈 객체에서 나와 `projected=` / `selfFixRoundsApplied=0` 이 됐다. """ try: data = json.loads(final_report_data_path(report_path).read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return {} return data if isinstance(data, Mapping) else {} def _progress_workers( evidence: _LeadEvidence, phase: str, *, completed_only: bool = False, ) -> set[str]: workers: set[str] = set() for _timestamp, event_phase, line in evidence.progress: if event_phase != phase: continue if completed_only and _COMPLETED_STATUS_RE.search(line) is None: continue worker = _progress_worker(line) if worker is not None: workers.add(worker) return workers def _progress_worker(line: str) -> str | None: match = _WORKER_FIELD_RE.search(line) return match.group("worker") if match else None def _activity_has_agent(events: list[LeadEvent], roles: set[str]) -> bool: return any( _worker_role(str(event.details.get("agent") or "")) in roles for event in events ) def _roster_roles_for_value(value: str, workers: list[dict]) -> set[str]: candidate = _worker_role(value) for worker in workers: roles = _worker_roles(worker) if candidate in roles: return roles return {candidate} if candidate else set() def _check_activity_id_order(events: list[LeadEvent], errors: list[str]) -> None: activity_ids = [str(event.details.get("activityId") or "") for event in events] invalid = [ value or "" for value in activity_ids if not _ACTIVITY_ID_RE.fullmatch(value) ] if invalid: errors.append( "activity contract: activityId values must use A-NNN format; " f"invalid={invalid}." ) duplicates = sorted( {value for value in activity_ids if activity_ids.count(value) > 1} ) if duplicates: errors.append( "activity contract: activityId values must be unique; " f"duplicates={duplicates}." ) if invalid: return numbers = [ int(_ACTIVITY_ID_RE.fullmatch(value).group(1)) for value in activity_ids ] if any(current <= previous for previous, current in zip(numbers, numbers[1:])): errors.append( "activity contract: activityId values must be in strict A-NNN order; " f"recorded={activity_ids}." ) def _check_projected_agent_activity( events: list[LeadEvent], report_data: Mapping[str, Any], errors: list[str], ) -> None: expected = [ {field: event.details.get(field) for field in ACTIVITY_FIELDS} for event in events ] raw_projected = report_data.get("agentActivity") projected = ( [ {field: row.get(field) for field in ACTIVITY_FIELDS} for row in raw_projected if isinstance(row, Mapping) ] if isinstance(raw_projected, list) else raw_projected ) if projected == expected: return mismatch = "length" if isinstance(projected, list) and len(projected) == len(expected): mismatch = next( ( f"index {index}" for index, (actual, wanted) in enumerate(zip(projected, expected)) if actual != wanted ), "content", ) errors.append( "activity contract: final-report data.json agentActivity must exactly " "match canonical activity events in order and core fields; " f"mismatch={mismatch}, expected={len(expected)}, " f"projected={len(projected) if isinstance(projected, list) else ''}." ) def _check_activity_worker_pairs( evidence: _LeadEvidence, indexed: dict[str, list[LeadEvent]], analysis_workers: list[dict], errors: list[str], ) -> None: expected = ( ("phase-4-dispatch", "worker-dispatched", False), ("phase-5-collect", "worker-completed", True), ) for phase, kind, completed_only in expected: workers = _progress_workers( evidence, phase, completed_only=completed_only ) for worker in sorted(workers): roles = _roster_roles_for_value(worker, analysis_workers) if _activity_has_agent(indexed.get(kind, []), roles): continue errors.append( f"activity contract: missing `{kind}` for worker `{worker}` " f"recorded by `PROGRESS: {phase}`." ) def _check_activity_worker_agents( indexed: dict[str, list[LeadEvent]], analysis_workers: list[dict], errors: list[str], ) -> None: accepted = { role for worker in analysis_workers for role in _worker_roles(worker) } roster = [str(worker.get("role") or "") for worker in analysis_workers] for kind in ("worker-dispatched", "worker-completed"): for event in indexed.get(kind, []): agent = str(event.details.get("agent") or "") if _worker_role(agent) in accepted: continue errors.append( f"activity contract: `{kind}` agent `{agent or ''}` must " f"match an actual manifest roster role; roster={roster}." ) def _allowed_automatic_rounds(self_fix_rounds: int, *, gating: bool = True) -> int: """이 run 이 돌아도 되는 자동 plan-body 라운드 수. 상한이 2 로 고정돼 있던 동안 다른 규칙과 정면으로 충돌했다. `validators/validate-run.py` `_validate_verdict_rounds_outlive_self_fix` 는 자가수정 뒤 **모든** 항목이 그 rewrite 이후 라운드의 판정을 들고 있어야 한다고 요구한다. 그런데 자가수정 직후의 표적 재검증은 고친 항목만 덮으므로, 나머지 항목을 덮는 배치가 하나 더 필요하다 — 실측 run 에서 114항목 중 18개가 표적 라운드, 남은 96개가 그다음 라운드였다. 상한 2 는 그 마지막 배치를 금지하고, 금지에 따르면 96개가 고쳐지기 전 본문에 대한 판정을 게이트에 싣는다. 기준선은 2다 — 초기 검증과 재검증 한 번. 자가수정이 없어도 두 번째 배치가 필요할 수 있다: 동수이거나 상대 워커가 판정을 못 낸 항목은 `needs-reverify` 로 그 라운드에 실린다. 자가수정은 여기에 배치를 하나씩 더한다. 표적 재검증은 기준선의 두 번째 배치가 받고, 늘어나는 것은 그 라운드가 덮지 못한 잔여 항목을 일소하는 배치다. ``gating=false`` 자문 경로는 추출과 1라운드만 남긴다. self-fix 와 일소 배치는 돌리지 않는다. """ if not gating: return 1 return 2 + max(self_fix_rounds, 0) def _plan_body_state_document(run_dir: Path, suffix: str | None) -> dict | None: if not suffix: return None try: doc = json.loads(_plan_body_state_path(run_dir, suffix).read_text()) except (OSError, json.JSONDecodeError): return None return doc if isinstance(doc, dict) else None def _plan_body_round_history_len(run_dir: Path, suffix: str | None) -> int | None: doc = _plan_body_state_document(run_dir, suffix) if doc is None: return None history = doc.get("roundHistory") if not isinstance(history, list): return None return len(history) def _self_fix_event_expectation( run_dir: Path, suffix: str | None, self_fix_rounds: int, ) -> int: """selfFixRoundsApplied 는 마지막 라운드 번호다. 활동 건수는 그룹 라운드 수. 라운드 3 에 원인 그룹 4개를 한 번에 쓰면 필드 값은 3, 이벤트는 1이다. """ doc = _plan_body_state_document(run_dir, suffix) if doc is None: return self_fix_rounds projection = doc.get("planBodyVerification") groups = ( projection.get("selfFixGroups") if isinstance(projection, dict) else None ) if not isinstance(groups, list) or not groups: return self_fix_rounds rounds = { row.get("round") for row in groups if isinstance(row, dict) and isinstance(row.get("round"), int) } return len(rounds) if rounds else self_fix_rounds def _self_fix_rounds_from_state(run_dir: Path, suffix: str | None) -> int | None: """이번 런 상태 파일의 자가수정 횟수. 없으면 None — 호출자가 리포트로 폴백. 리포트 칸은 이어진 seq 의 누적이라, 이번 창의 `self-fix-applied` 건수와 비교하면 어긋난다. """ doc = _plan_body_state_document(run_dir, suffix) if doc is None: return None projection = doc.get("planBodyVerification") value = ( projection.get("selfFixRoundsApplied") if isinstance(projection, dict) else None ) if isinstance(value, int) and value >= 0: return value value = doc.get("selfFixRoundsApplied") return value if isinstance(value, int) and value >= 0 else None def _is_plan_body_verification_activity(event: LeadEvent) -> bool: """`verification-round-completed` 가 계획 본문 배치인지. 같은 kind 로 적대 재검증 라운드도 남는다. 그 건을 본문 라운드에 넣으면 recorded 가 roundCount 보다 커진다. 요약이 `adversarial reverify` 이면 본문이 아니다. 픽스처 요약(`verification-round-completed for …`)은 본문이다. """ summary = str(event.details.get("summary") or "").lower() if "adversarial reverify" in summary or "adversarial re-verify" in summary: return False if "critic-gap" in summary: return False return True def _plan_body_verification(report_data: Mapping[str, Any]) -> Mapping[str, Any] | None: planning = report_data.get("implementationPlanning") verification = ( planning.get("planBodyVerification") if isinstance(planning, Mapping) else None ) return verification if isinstance(verification, Mapping) else None def _self_fix_rounds_applied(report_data: Mapping[str, Any]) -> int: plan_verification = _plan_body_verification(report_data) value = ( plan_verification.get("selfFixRoundsApplied") if plan_verification is not None else 0 ) return value if isinstance(value, int) and value >= 0 else 0 def _plan_body_gating(report_data: Mapping[str, Any]) -> bool: plan_verification = _plan_body_verification(report_data) if plan_verification is None: return True return plan_verification.get("gating") is not False def _matching_user_reverification_round( event: LeadEvent | None, clarification_id: str, plan_item_ids: set[str], verification_rounds: int, ) -> int | None: if event is None or event.details.get("outcome") != "resolved": return None evidence_refs = { ref for ref in (event.details.get("evidenceRefs") or []) if isinstance(ref, str) } clarification_refs = { ref for ref in evidence_refs if _CLARIFICATION_ID_RE.fullmatch(ref) } event_item_ids = { item_id for item_id in (event.details.get("planItemIds") or []) if isinstance(item_id, str) } round_numbers = { int(match.group(1)) for ref in evidence_refs for match in [_TARGETED_REVERIFICATION_REF_RE.fullmatch(ref)] if match is not None } if clarification_refs != {clarification_id} or event_item_ids != plan_item_ids: return None if len(round_numbers) != 1: return None round_number = next(iter(round_numbers)) return round_number if round_number <= verification_rounds else None def _resolved_correctness_reverification_rounds( indexed: dict[str, list[LeadEvent]], report_data: Mapping[str, Any], verification_rounds: int, ) -> set[int]: evaluated_by_id = { str(event.details.get("activityId") or ""): event for event in indexed.get("user-decision-evaluated", []) } rounds: set[int] = set() for row in report_data.get("clarificationItems") or []: if not isinstance(row, Mapping) or row.get("status") != "resolved": continue context = row.get("approvalContext") if not isinstance(context, Mapping): continue if context.get("classification") != "correctness-critical": continue resolution = context.get("resolution") if not isinstance(resolution, Mapping): continue clarification_id = str(row.get("id") or "") plan_item_ids = { item_id for item_id in (context.get("planItemIds") or []) if isinstance(item_id, str) } if not plan_item_ids: continue check_refs = [ ref for ref in (resolution.get("checkRefs") or []) if isinstance(ref, str) ] matched_rounds = [ _matching_user_reverification_round( evaluated_by_id.get(check_ref), clarification_id, plan_item_ids, verification_rounds, ) for check_ref in check_refs ] if matched_rounds and all( round_number is not None for round_number in matched_rounds ): unique_rounds = set(matched_rounds) if len(unique_rounds) == 1: rounds.update(unique_rounds) return rounds def _check_activity_round_counts( indexed: dict[str, list[LeadEvent]], report_data: Mapping[str, Any], run_dir: Path, suffix: str | None, errors: list[str], ) -> None: verification_rounds = _plan_body_rounds_ran(run_dir, suffix) recorded_verifications = len([ event for event in indexed.get("verification-round-completed", []) if _is_plan_body_verification_activity(event) ]) user_reverification_rounds = _resolved_correctness_reverification_rounds( indexed, report_data, verification_rounds, ) automatic_rounds = verification_rounds - len(user_reverification_rounds) state_self_fix = _self_fix_rounds_from_state(run_dir, suffix) self_fix_rounds = ( state_self_fix if state_self_fix is not None else _self_fix_rounds_applied(report_data) ) allowed_rounds = _allowed_automatic_rounds( self_fix_rounds, gating=_plan_body_gating(report_data), ) if automatic_rounds > allowed_rounds: errors.append( f"activity contract: at most {allowed_rounds} plan verification " f"batch(es) are allowed for {self_fix_rounds} self-fix round(s); " f"roundCount={verification_rounds}, " f"humanReverificationRounds={sorted(user_reverification_rounds)}." ) history_len = _plan_body_round_history_len(run_dir, suffix) if ( recorded_verifications != automatic_rounds and history_len != verification_rounds ): errors.append( "activity contract: `verification-round-completed` count must match " f"automatic plan-body rounds={automatic_rounds} from " f"roundCount={verification_rounds} in " f"{_plan_body_state_path(run_dir, suffix or '')}; " f"recorded={recorded_verifications}." ) recorded_self_fixes = len(indexed.get("self-fix-applied", [])) if recorded_self_fixes > 1: errors.append( "activity contract: automatic self-fix is limited to one rewrite; " "resolve remaining items through lead decisions or user confirmation." ) expected_self_fixes = _self_fix_event_expectation( run_dir, suffix, self_fix_rounds, ) if recorded_self_fixes != expected_self_fixes: errors.append( "activity contract: `self-fix-applied` count must match " f"selfFixRoundsApplied={expected_self_fixes}; recorded={recorded_self_fixes}." ) def _activity_references(event: LeadEvent, reference: str) -> bool: refs = event.details.get("evidenceRefs") return isinstance(refs, list) and reference in refs def _check_activity_user_decisions( indexed: dict[str, list[LeadEvent]], report_data: Mapping[str, Any], errors: list[str], ) -> None: rows = report_data.get("clarificationItems") if not isinstance(rows, list): return required = indexed.get("user-decision-required", []) evaluated = indexed.get("user-decision-evaluated", []) for row in rows: if not isinstance(row, Mapping) or row.get("blocks") != "approval": continue clarification_id = str(row.get("id") or "") confirmation = str(row.get("userConfirmation") or "") if confirmation.startswith("asked-") and not any( _activity_references(event, clarification_id) for event in required ): errors.append( "activity contract: missing `user-decision-required` referencing " f"approval clarification `{clarification_id}`." ) if row.get("status") == "resolved" and not any( _activity_references(event, clarification_id) for event in evaluated ): errors.append( "activity contract: missing `user-decision-evaluated` referencing " f"resolved approval clarification `{clarification_id}`." ) def _check_activity_contract( evidence: _LeadEvidence, team_state: Mapping[str, Any], run_manifest: Mapping[str, Any], report_path: Path, run_dir: Path, suffix: str | None, errors: list[str], ) -> None: if not _is_activity_contract_v1_planning(run_manifest): return indexed = _activity_index(evidence.activities) workers = [ worker for worker in (team_state.get("workers") or []) if isinstance(worker, dict) ] analysis_workers = _analysis_workers(team_state) report_data = _read_report_data(report_path) _check_activity_id_order(evidence.activities, errors) _check_projected_agent_activity(evidence.activities, report_data, errors) _check_activity_worker_pairs(evidence, indexed, analysis_workers, errors) _check_activity_worker_agents(indexed, workers, errors) _check_activity_round_counts(indexed, report_data, run_dir, suffix, errors) _check_activity_user_decisions(indexed, report_data, errors) def _check_user_confirm_checkpoints( by_phase: dict[str, list[tuple[str, str]]], asked_ids: list[str], activities: list, errors: list[str], ) -> None: """"물어봤다"는 리포트의 기록에는 물어본 흔적이 세션에 남아 있어야 한다. 승인을 막는 행은 run 하나를 통째로 대기 상태로 만든다. 그런 행을 세우기 전에 사용자에게 그 자리에서 묻는 것이 계약인데, 그 이행 여부를 리포트의 자기 신고 말고는 확인할 길이 없었다. 한 번은 리드가 차단이 생길 것을 미리 예측해 말해 놓고도 묻지 않은 채 행으로 만들었고, run 은 사용자가 이미 답해 둔 질문 앞에서 self-fix 예산을 전부 썼다. 체크포인트가 그 격차를 좁힌다 — 세션에 질문이 남지 않았다면 리드는 묻지 않은 것이다.""" if not asked_ids: return detail = "prompts/lead/okstra-lead-contract.md 'User confirmation before an approval blocker'" asked_lines = " ".join(line for _ts, line in by_phase.get("user-confirm", [])) # 원장이 더 강한 증거다. `okstra agent-activity append --kind # user-decision-required --evidence-ref C-NNN` 이 남긴 행은 리드가 그 # 질문을 실제로 냈다는 구조화된 기록이고, 대화에 같은 문장을 냈는지보다 # 확실하다. 실측(dev-10784 error-analysis): 리드가 C-001·C-002 를 묻고 # 두 행을 정확히 남겼는데, 이 검사가 대화 텍스트만 보아 둘 다 누락으로 # 보고했다. asked_in_ledger = { ref for event in activities if event.details.get("kind") == "user-decision-required" for ref in (event.details.get("evidenceRefs") or []) if isinstance(ref, str) } for row_id in asked_ids: if row_id in asked_lines or row_id in asked_in_ledger: continue errors.append( f"PROGRESS checkpoint missing: `user-confirm {row_id}` — the final " f"report records `{row_id}` as an approval blocker the user was " "asked about, but no such line exists in this run's evidence. Either " "ask before writing the row, or record what actually happened " f"(`deferred-no-interactive-session`) ({detail})." ) def _check_user_confirm_activities( evidence: _LeadEvidence, report_path: Path, errors: list[str], ) -> None: """artifact-only 호스트에서 "물어봤다"를 확인하는 자리. `_check_user_confirm_checkpoints` 와 같은 질문에 답하되, 근거를 대화 텍스트가 아니라 activity 원장의 `user-decision-required` 행에서 찾는다. 리드는 `okstra agent-activity append --kind user-decision-required --evidence-ref C-NNN` 으로 그 행을 남긴다. """ asked_ids = _ids_reported_as_asked(report_path) if not asked_ids: return required = [ event for event in evidence.activities if event.details.get("kind") == "user-decision-required" ] for row_id in asked_ids: if any(_activity_references(event, row_id) for event in required): continue errors.append( f"activity contract: the final report records `{row_id}` as an " "approval blocker the user was asked about, but no " "`user-decision-required` activity references it. Ask before " "writing the row, then record it with `okstra agent-activity " f"append --kind user-decision-required --evidence-ref {row_id}`." ) def _phase_mentions_worker( lines: list[tuple[str, str]], roles: set[str] ) -> bool: return any( _worker_role(worker) in roles for _ts, line in lines if (worker := _progress_worker(line)) is not None ) def _check_worker_checkpoint_lines( by_phase: dict[str, list[tuple[str, str]]], analysis_workers: list[dict], errors: list[str], ) -> None: """phase-4-dispatch / phase-5-collect 의 per-worker 라인 (SKILL.md: once per worker).""" for worker in analysis_workers: role = str(worker.get("role", "")).strip() or "" status = str(worker.get("status", "")).strip() roles = _worker_roles(worker) if status in _ATTEMPTED_STATUSES and not _phase_mentions_worker( by_phase.get("phase-4-dispatch", []), roles ): errors.append( f"PROGRESS checkpoint missing: no `phase-4-dispatch worker=` " f"line names worker `{role}` — one line per dispatched worker, " "prompts/lead/okstra-lead-contract.md 'Progress reporting (BLOCKING)'." ) if status == "completed" and not _phase_mentions_worker( by_phase.get("phase-5-collect", []), roles ): errors.append( f"PROGRESS checkpoint missing: no `phase-5-collect worker=` " f"line names completed worker `{role}` — one line per collected " "result, prompts/lead/okstra-lead-contract.md 'Progress reporting (BLOCKING)'." ) def _check_batch_cleanup_checkpoints( by_phase: dict[str, list[tuple[str, str]]], convergence_ran: bool, report_writer_dispatched: bool, errors: list[str], ) -> None: """phase-batch-cleanup: 각 배치 경계 직전에 이전 배치 pane/teammate 정리가 실제로 일어났는지 (prompts/profiles/_common-contract.md 'Phase-start cleanup'). R1=convergence round 1 직전, R2=report-writer dispatch 직전(수렴이 있었으면 마지막 라운드 이후 별도 1회). ISO-8601 ts 는 lexicographic 비교가 곧 시간순.""" detail = "prompts/lead/okstra-lead-contract.md 'Run-scoped worker-resource lifecycle'" cleanup_ts = sorted(ts for ts, _line in by_phase.get("phase-batch-cleanup", [])) conv_ts = sorted(ts for ts, _line in by_phase.get("phase-5.5-convergence", [])) synth_ts = sorted(ts for ts, _line in by_phase.get("phase-6-synthesis", [])) if convergence_ran and conv_ts and not any(ts <= conv_ts[0] for ts in cleanup_ts): errors.append( "PROGRESS checkpoint missing: no `phase-batch-cleanup` line before the " f"first `phase-5.5-convergence` round ({conv_ts[0]}) — the prior analysis " f"batch's panes/teammates must be cleared first ({detail})." ) if report_writer_dispatched and synth_ts: lower = conv_ts[-1] if (convergence_ran and conv_ts) else "" if not any(lower <= ts <= synth_ts[0] for ts in cleanup_ts): errors.append( "PROGRESS checkpoint missing: no `phase-batch-cleanup` line before " f"`phase-6-synthesis` ({synth_ts[0]}) — the prior batch must be cleared " f"before dispatching report-writer ({detail})." ) def _check_plan_verify_cleanup_checkpoints( by_phase: dict[str, list[tuple[str, str]]], plan_body_rounds: int, errors: list[str], ) -> None: """plan-body 라운드도 배치 경계다 — 라운드마다 새 검증 워커를 띄운다. Phase 6 뒤에 도는 구간이라 위 두 지점의 바깥이었고, self-fix 를 다섯 라운드 돈 run 은 라운드마다 완료 워커를 남겨 유휴 세션이 쌓였다. 라운드 2 이상은 직전 라운드의 워커가 반드시 존재하므로, 각 라운드 직전에 정리가 있어야 한다(라운드 1의 앞 경계는 report-writer 디스패치 정리가 이미 덮는다).""" if plan_body_rounds < 2: return detail = "prompts/lead/plan-body-verification.md §\"Round protocol\" step 7" round_ts = sorted(ts for ts, _line in by_phase.get("phase-5.5.9-plan-verify", [])) if not round_ts: errors.append( "PROGRESS checkpoint missing: `phase-5.5.9-plan-verify` — the state " f"file records {plan_body_rounds} plan-body rounds, each of which " f"dispatches a worker batch and must announce itself ({detail})." ) return cleanup_ts = sorted(ts for ts, _line in by_phase.get("phase-batch-cleanup", [])) for index, dispatched_at in enumerate(round_ts[1:], start=1): previous = round_ts[index - 1] if not any(previous <= ts <= dispatched_at for ts in cleanup_ts): errors.append( "PROGRESS checkpoint missing: no `phase-batch-cleanup` line " f"between plan-body rounds ({previous} → {dispatched_at}) — the " f"previous round's completed verifiers must be cleared before " f"the next round dispatches ({detail})." ) _STAGE_ANNOUNCE_RE = re.compile(r"\bstage=(\d+)\b") def _announced_stage(by_phase: Mapping[str, list[tuple[str, str]]]) -> int | None: for _ts, line in by_phase.get("phase-5-stage", []): matched = _STAGE_ANNOUNCE_RE.search(line) if matched: return int(matched.group(1)) return None def _stage_carry_persisted( run_dir: Path, by_phase: Mapping[str, list[tuple[str, str]]] ) -> bool: """이 stage 의 carry 사이드카가 디스크에 있는가 — 완료 라인의 전제. `phase-5-stage-complete` 는 executor 의 `### Stage Carry Evidence` 를 파싱한 뒤에만 낼 수 있고, 계약은 executor 가 carry 증거 없이 끝나면(FAIL 또는 non-result) 그 라인을 생략하라고 한다. 그런데 검사는 명부의 `completed` 만 봐서, 계약대로 생략한 리드가 권고를 받았다(실측 2026-09-08, fontsninja-v3-site dev-10627-2 implementation stage-1: executor 가 상위 게이트 부재로 carry 를 의도적으로 보류). FAIL 이면 사이드카를 쓰지 않는다는 같은 계약을 `validate-run.py` `_validate_stage_carry_sidecar_exists` 가 집행하므로, 그 파일의 존재를 완료 라인의 조건으로 쓴다. stage 번호는 run 디렉터리(`stage-`)에서, 없으면 `phase-5-stage stage=` 공지에서 읽는다. """ try: ref = RunRef.from_run_dir(run_dir) except ValueError: return True stage = ref.stage if ref.stage is not None else _announced_stage(by_phase) if stage is None: return any(ref.carry_dir.glob("stage-*.json")) return ref.carry(stage).exists() def _check_progress_checkpoints( evidence: _LeadEvidence, team_state: dict, run_dir: Path, suffix: str | None, report_path: Path, task_type: str, errors: list[str], advisories: list[str] | None = None, ) -> None: # The narration lines are advisory (see `SessionConformanceResult`); the # user-confirmation cross-check below stays in `errors`. narration = advisories if advisories is not None else errors by_phase: dict[str, list[tuple[str, str]]] = {} for ts, phase, line in evidence.progress: by_phase.setdefault(phase, []).append((ts, line)) def require(phase: str, condition: bool, detail: str) -> None: if condition and phase not in by_phase: narration.append( f"PROGRESS checkpoint missing: `{phase}` ({detail}) — " "prompts/lead/okstra-lead-contract.md 'Progress reporting (BLOCKING)'." ) intake = by_phase.get("phase-1-intake", []) if not any("complete" not in line.lower() for _ts, line in intake): narration.append( "PROGRESS checkpoint missing: `phase-1-intake reading task bundle` " "(start-of-Phase-1 line) — prompts/lead/okstra-lead-contract.md 'Progress reporting (BLOCKING)'." ) if not any("complete" in line.lower() for _ts, line in intake): narration.append( "PROGRESS checkpoint missing: `phase-1-intake complete` " "(after all intake reads) — prompts/lead/okstra-lead-contract.md 'Progress reporting (BLOCKING)'." ) workers = [w for w in (team_state.get("workers") or []) if isinstance(w, dict)] analysis_workers = _analysis_workers(team_state) any_dispatched = any( str(w.get("status", "")).strip() in _DISPATCHED_STATUSES for w in workers ) require("phase-2-prompts", bool(workers), "before any Write to assigned prompt paths") require( "phase-3-team-create", any_dispatched, "in Phase 3 after recording teamCreate in team-state — the " "`using implicit team` line, or the " "`skipped (concurrent-run)` variant in the concurrent-run path", ) _check_worker_checkpoint_lines(by_phase, analysis_workers, narration) convergence_ran = _convergence_rounds_ran(run_dir, suffix) require( "phase-5.5-convergence", convergence_ran, "at the start of each convergence round (state artifact records totalRounds >= 1)", ) report_writer = next((w for w in workers if _is_report_writer(w)), None) report_writer_dispatched = ( report_writer is not None and str(report_writer.get("status", "")).strip() in _DISPATCHED_STATUSES ) require( "phase-6-synthesis", report_writer_dispatched, "at the start of Phase 6 (report-writer dispatch)", ) if task_type == "implementation": # 이 run 이 실행하는 stage 의 공지/완료 라인 — implementation 전용 두 # 체크포인트. 완료 라인은 carry 증거를 파싱한 뒤에만 낼 수 있으므로 # implementer 가 완주한 run 에만 요구한다. require( "phase-5-stage", any_dispatched, "immediately before the Executor dispatch — `stage= title= " "steps=<count>` from the approved plan's Stage Map", ) implementer_completed = any( "implementer" in _norm(str(w.get("workerId") or "") + str(w.get("role") or "")) and str(w.get("status", "")).strip() == "completed" for w in workers ) require( "phase-5-stage-complete", implementer_completed and _stage_carry_persisted(run_dir, by_phase), "immediately after parsing the Executor's `### Stage Carry Evidence` " "block — `stage=<N> steps=<done>/<count>` from its stepResults", ) require("phase-7-persist", True, "at the start of Phase 7") _check_batch_cleanup_checkpoints( by_phase, convergence_ran, report_writer_dispatched, narration ) _check_plan_verify_cleanup_checkpoints( by_phase, _plan_body_rounds_ran(run_dir, suffix), narration ) _check_user_confirm_checkpoints( by_phase, _ids_reported_as_asked(report_path), evidence.activities, errors ) def _parse_iso(ts: str) -> datetime | None: try: return datetime.fromisoformat(ts.replace("Z", "+00:00")) except ValueError: return None def _check_heartbeat_sidecar(path: Path, worker_id: str, errors: list[str]) -> None: rel = path.name if not path.is_file(): errors.append(f"registered audit sidecar missing: {path}") return try: content = path.read_text(encoding="utf-8") except OSError as exc: errors.append(f"worker audit sidecar unreadable: {rel} ({exc})") return entries = [(m.group("stage"), m.group("ts")) for m in _HEARTBEAT_LINE_RE.finditer(content)] if not entries: errors.append( f"`{rel}` has no `- PROGRESS: <stage> <ISO-8601-UTC>` heartbeat lines — " "the in-process worker MUST write `started` immediately and append one " "line per stage at <= 5-minute cadence (agents/workers/claude-worker.md " "'Heartbeat', prompts/lead/okstra-lead-contract.md Common Mistakes)." ) return if entries[0][0] != "started": errors.append( f"`{rel}`: first heartbeat stage must be `started` " f"(found `{entries[0][0]}`) — the sidecar is written BEFORE the " "per-file reads, with a `- PROGRESS: started <ISO>` line." ) if worker_id == "report-writer": for stage, _raw_ts in entries: # `in-stage:<stage>` 는 계약이 장기 단계에 요구하는 진행 라인이다 — # allowlist 가 이를 배제하면 워커는 cadence 를 지킬 방법이 없다. base = stage[len(_IN_STAGE_PREFIX):] if stage.startswith(_IN_STAGE_PREFIX) else stage if base not in _REPORT_WRITER_HEARTBEAT_STAGES: errors.append( f"`{rel}`: heartbeat stage `{stage}` is not allowed for report-writer." ) # result 파일이 존재하면 마지막 단계 마커도 있어야 한다. timeout 으로 중단된 # worker(result 없음)에는 요구하지 않는다 — hang 이전 구간의 cadence 만 본다. result_file = path.with_name(rel.replace("-audit-", "-")) if result_file.exists() and not any(s == "write-result-start" for s, _ in entries): errors.append( f"`{rel}`: heartbeat is missing the `write-result-start` stage line " f"although the worker result `{result_file.name}` exists — every " "stage must append its own line (agents/workers/claude-worker.md 'Heartbeat')." ) # implementer(Executor)는 지금 실행 중인 plan step 을 하트비트로 알린다 # (_implementation-executor.md 'Per-step heartbeat'). result 를 남기고 완주한 # executor 의 사이드카에 step 라인이 하나도 없으면 이 run 의 유일한 스텝 단위 # 진행 신호가 없었던 것이다. cli-wrapper executor 는 wrapper-status 경로라 # 이 함수에 들어오지 않는다. if ( "implementer" in _norm(worker_id) and result_file.exists() and not any( stage.removeprefix(_IN_STAGE_PREFIX).startswith("step-") for stage, _ in entries ) ): errors.append( f"`{rel}`: implementer heartbeat has no `- PROGRESS: step-<k> <ISO>` " "line although the worker result exists — the executor MUST announce " "each plan step as it starts " "(prompts/profiles/_implementation-executor.md 'Per-step heartbeat')." ) prev: datetime | None = None prev_stage = "" for stage, raw_ts in entries: ts = _parse_iso(raw_ts) if ts is None: errors.append( f"`{rel}`: heartbeat line for stage `{stage}` has an unparseable " f"ISO-8601 timestamp `{raw_ts}`." ) continue if prev is not None: # 예산은 구간을 여는 단계에서 고른다 — 이 공백이 재는 것은 직전에 # 선언된 단계의 작업 시간이다. budget = max_gap_seconds_after(prev_stage) gap = (ts - prev).total_seconds() if gap < 0: errors.append( f"`{rel}`: heartbeat timestamps regress at stage `{stage}` ({raw_ts})." ) elif gap > budget: errors.append( f"`{rel}`: heartbeat gap after stage `{prev_stage}` is {int(gap)}s " f"(budget {budget}s) — emit " "`- PROGRESS: in-stage:<stage> <ISO>` during long stages." ) prev = ts prev_stage = stage def _check_worker_liveness( project_root: Path, team_state: Mapping[str, Any], errors: list[str] ) -> None: """Validate the liveness artifact selected by each dispatch record.""" dispatches = team_state.get("workerDispatches") if not isinstance(dispatches, list): return for record in dispatches: if not isinstance(record, Mapping): continue mode = record.get("livenessMode") worker_id = record.get("workerId") if not isinstance(worker_id, str) or not worker_id: continue if mode not in (_LIVENESS_AUDIT_HEARTBEAT, _LIVENESS_WRAPPER_STATUS): errors.append( f"worker `{worker_id}` dispatch has missing or unknown livenessMode: {mode!r}." ) continue if mode == _LIVENESS_AUDIT_HEARTBEAT: raw_path = record.get("auditSidecarPath") if not isinstance(raw_path, str) or not raw_path: errors.append( f"worker `{worker_id}` audit-heartbeat dispatch has no registered audit sidecar path." ) continue path = Path(raw_path) if not path.is_absolute(): path = project_root / path _check_heartbeat_sidecar(path, worker_id, errors) elif mode == _LIVENESS_WRAPPER_STATUS: raw_path = record.get("statusSidecarPath") if not isinstance(raw_path, str) or not raw_path: errors.append( f"worker `{worker_id}` wrapper-status dispatch has no registered status sidecar path." ) continue path = Path(raw_path) if not path.is_absolute(): path = project_root / path status = read_wrapper_status(path) if status is None: errors.append(f"registered wrapper status sidecar unreadable: {path}") def _instruction_set_dir(run_dir: Path, suffix: str | None, project_root: Path) -> Path | None: """이 run 의 instruction-set 디렉터리. 해소 불가면 None (경로를 모르면 스테이징 여부도 판정할 수 없고, 모르는 것을 근거로 run 을 떨어뜨리지 않는다).""" if not suffix: return None manifest = run_dir / "manifests" / f"run-manifest-{suffix}.json" try: data = json.loads(manifest.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError, UnicodeDecodeError): return None raw = data.get("instructionSetPath") if isinstance(data, dict) else None if not isinstance(raw, str) or not raw: return None path = Path(raw) if not path.is_absolute(): path = project_root / path return path if path.is_dir() else None def _check_cmux_adapter_read( evidence: _LeadEvidence, team_state: dict, errors: list[str] ) -> None: """검사 4 — cmux 어댑터 읽음. 모든 task-type 에 적용된다. 다른 어댑터는 lead 의 런타임이 고르지만 이것은 환경이 고른다. 그래서 읽지 않은 lead 에게는 이 경로가 존재한다는 사실 자체가 닿지 않고, 자기 런타임이 아는 host 네이티브 디스패치로 되돌아간다 — cmux 경로에서 okstra 가 소유한 바로 그 일이다.""" adapter = team_state.get("leadAdapter") name = str(adapter.get("name", "")).strip() if isinstance(adapter, dict) else "" if name != CMUX_ADAPTER_NAME: return if evidence.sidecar_reads.get(CMUX_ADAPTER_BASENAME): return source, _error = _conformance_evidence_source(team_state) if source == "artifact-only": # grok 같은 artifact-only 호스트는 Read 도구 기록이 없다. 면제 사유는 # 리드가 읽기를 남기지 못한다는 것이지 워커를 어떻게 띄웠는지가 아니다. return errors.append( f"cmux adapter: no read of `{CMUX_ADAPTER_BASENAME}` (a `Read` call or a " f"shell command naming it) found in the " "selected adapter evidence source within this run's window — the cmux " "adapter is selected by environment, not by lead runtime, so it MUST be " f"read before dispatch ({_CMUX_ADAPTER_CITE})." ) def _entry_guard_rows( task_type: str, seen_phases: set[str], instruction_set: Path | None ) -> list[_EntryGuardRead]: """이 run 에 실제로 걸리는 entry-guard 행.""" rows = [] for row in _ENTRY_GUARD_READS: if row.task_types is not None and task_type not in row.task_types: continue if row.required_when is not None and row.required_when not in seen_phases: # 그 라운드를 돌지 않은 run 이다. 돌지 않은 일의 규칙을 요구하지 않는다. continue if row.staged_per_run and not ( instruction_set and (instruction_set / row.basename).is_file() ): continue rows.append(row) return rows def _check_entry_guard_reads( evidence: _LeadEvidence, errors: list[str], instruction_set: Path | None, task_type: str, evidence_source: str = "claude-jsonl", ) -> None: """검사 3 — entry guard. fresh-read 규칙(이전 run 기억으로 갈음 불가)은 run 윈도우 스코핑이 보장한다: 이번 윈도우 안의 Read 만 증거로 인정된다. anchor 가 있는 행은 존재 + 순서를, anchor 가 `None` 인 행은 존재만 본다. `staged_per_run` 행은 이 run 의 instruction-set 에 실제로 있을 때만, `required_when` 행은 그 PROGRESS 체크포인트가 이 run 에 있을 때만 요구한다.""" if evidence_source == "artifact-only": # `_check_cmux_adapter_read` 와 같은 사유다: artifact-only 호스트의 # 리드는 파일을 읽어도 그 사실을 남길 도구 기록이 없다. 그 자리에서 # "읽은 흔적이 없다"는 읽지 않았다는 뜻이 아니라 기록 형식이 없다는 # 뜻이고, 그 둘을 구분하지 못하는 검사는 판정하지 않는다. return seen_phases = {phase for _ts, phase, _line in evidence.progress} anchor_phases = {row.anchor[0] for row in _ENTRY_GUARD_READS if row.anchor} anchors: dict[str, str] = {} for ts, phase, _line in evidence.progress: # progress 는 ts 정렬 — 첫 항목이 최초 발생 if phase in anchor_phases and ts: anchors.setdefault(phase, ts) for row in _entry_guard_rows(task_type, seen_phases, instruction_set): ts_list = evidence.sidecar_reads.get(row.basename) or [] if not ts_list: demand = ( f"it MUST be read fresh {row.anchor[1]}" if row.anchor else "the staged copy MUST be read at least once inside that window" ) errors.append( f"entry guard: no read of `{row.basename}` (a `Read` call " f"or a shell command naming it) found in " f"the selected adapter evidence source within this run's window — " f"{demand} ({row.cite})." ) continue if not row.anchor: continue anchor_phase, when = row.anchor anchor_ts = anchors.get(anchor_phase) if anchor_ts and min(ts_list) >= anchor_ts: errors.append( f"entry guard: `{row.basename}` was first read at " f"{min(ts_list)}, not before the first `PROGRESS: {anchor_phase}` " f"line ({anchor_ts}) — it must be loaded {when}, before " f"that checkpoint ({row.cite})." ) def validate_session_conformance( *, team_state: dict, team_state_path: Path, run_manifest: Mapping[str, Any], project_root: Path, report_path: Path, task_type: str, claude_projects_dir: Path | None = None, ) -> SessionConformanceResult: """Run post-hoc checks for the neutral lead and activity contracts. `claude_projects_dir` injects the Claude project root for tests and diagnostics. Heartbeat validation runs before selecting the adapter evidence source. """ result = SessionConformanceResult() _ensure_token_usage_importable() try: from okstra_token_usage.collect import run_artifact_suffix except ImportError as exc: # pragma: no cover — 설치본은 항상 패키지를 동반 result.errors.append(f"okstra_token_usage import failed — {exc}") return result run_dir = report_path.parent.parent suffix = run_artifact_suffix(team_state_path) _check_worker_liveness(project_root, team_state, result.errors) evidence_source, source_error = _conformance_evidence_source(team_state) if source_error: result.errors.append(source_error) return result if evidence_source == "artifact-only": evidence, error = _collect_artifact_lead_evidence( team_state, run_manifest, project_root, task_type, suffix ) else: evidence, error = _collect_lead_evidence( team_state, team_state_path, run_manifest, project_root, task_type, suffix, claude_projects_dir, ) if error: result.errors.append(error) return result _check_progress_checkpoints( evidence, team_state, run_dir, suffix, report_path, task_type, result.errors, result.advisories, ) _check_verification_errors_are_logged(run_dir, suffix, result.advisories) _check_activity_contract( evidence, team_state, run_manifest, report_path, run_dir, suffix, result.errors, ) _check_cmux_adapter_read(evidence, team_state, result.errors) _check_entry_guard_reads( evidence, result.errors, _instruction_set_dir(run_dir, suffix, project_root), task_type, evidence_source, ) return result