"""Compact path-hint persistence and legacy context hydration.""" from __future__ import annotations import re from pathlib import Path from typing import Any, Mapping from okstra_project.dirs import ( DISCOVERY_RELATIVE, OKSTRA_RELATIVE, TASKS_RELATIVE, okstra_home, ) from .paths import runs_dir_of, task_manifest_file, task_timeline_file from .worker_artifacts import artifacts_from_context RUN_CONTEXT_KIND = "run-context" RUN_CONTEXT_SCHEMA_VERSION = "2.0" ACTIVE_CONTEXT_KIND = "active-run-context" def compact_run_context(ctx: Mapping[str, Any]) -> dict[str, Any]: """Persist only path clues that can rehydrate the legacy flat context.""" return { "schemaVersion": RUN_CONTEXT_SCHEMA_VERSION, "kind": RUN_CONTEXT_KIND, "identity": _compact_identity(ctx), "pathHints": { "project": _compact_project_hints(), "task": { "taskRoot": ctx.get("TASK_ROOT_RELATIVE_PATH", ""), }, "run": _compact_run_hints(ctx), "runtime": { "okstraHome": str(okstra_home()), }, }, "timestamps": { "runTimestampIso": ctx.get("RUN_TIMESTAMP_ISO", ""), "taskDate": ctx.get("TASK_DATE", ""), }, } def compact_active_run_context( ctx: Mapping[str, Any], payload: Mapping[str, Any], ) -> dict[str, Any]: """Persist active-run-context with path hints instead of repeated paths.""" run_context = compact_run_context(ctx) return { "schemaVersion": RUN_CONTEXT_SCHEMA_VERSION, "kind": ACTIVE_CONTEXT_KIND, "identity": run_context["identity"], "pathHints": run_context["pathHints"], "task": _compact_active_task(payload), "workflow": dict(_mapping(payload.get("workflow"))), "run": { "stage": ctx.get("RUN_STAGE", ""), "fixRunCarry": ctx.get("FIX_RUN_CONTEXT", ""), }, "agentContract": dict(_mapping(payload.get("agentContract"))), "inputs": _compact_active_inputs(payload), "workers": _compact_active_workers(payload), "executorWorktree": dict(_mapping(payload.get("executorWorktree"))), "verificationTarget": dict(_mapping(payload.get("verificationTarget"))), "lazyReadPlan": dict(_mapping(payload.get("lazyReadPlan"))), } def hydrate_run_context(payload: Mapping[str, Any]) -> dict[str, Any]: """Return the legacy flat context for compact run-context payloads.""" if payload.get("kind") != RUN_CONTEXT_KIND: return dict(payload) hints = payload.get("pathHints") identity = payload.get("identity") if not isinstance(hints, Mapping) or not isinstance(identity, Mapping): return dict(payload) return _hydrate_from_hints(identity, hints, payload.get("timestamps", {})) def hydrate_active_run_context(payload: Mapping[str, Any]) -> dict[str, Any]: """Return the legacy active-run-context shape for compact payloads.""" if payload.get("kind") != ACTIVE_CONTEXT_KIND or "pathHints" not in payload: return dict(payload) identity = _mapping(payload.get("identity")) hints = _mapping(payload.get("pathHints")) ctx = hydrate_run_context({ "kind": RUN_CONTEXT_KIND, "identity": identity, "pathHints": hints, "timestamps": {}, }) return { "schemaVersion": "1.0", "kind": ACTIVE_CONTEXT_KIND, "task": _hydrate_active_task(payload, ctx), "workflow": dict(_mapping(payload.get("workflow"))), "run": _hydrate_active_run(payload, ctx), "agentContract": dict(_mapping(payload.get("agentContract"))), "instructionSet": _hydrate_active_instruction_set(payload, ctx), "workers": _hydrate_active_workers(payload, ctx), "errorLogs": _hydrate_active_error_logs(ctx), "runtimeResources": { "codingPreflightDir": ctx.get("OKSTRA_CODING_PREFLIGHT_DIR", ""), "workerPreamblePathByAudience": { "analysis": ctx.get("ANALYSIS_WORKER_PREAMBLE_PATH", ""), "implementation-executor": ctx.get( "IMPLEMENTATION_WORKER_PREAMBLE_PATH", "" ), "implementation-verifier": ctx.get( "IMPLEMENTATION_WORKER_PREAMBLE_PATH", "" ), "report-writer": ctx.get("REPORT_WRITER_PREAMBLE_PATH", ""), }, "workerErrorContractPath": ctx.get("WORKER_ERROR_CONTRACT_PATH", ""), }, "executorWorktree": dict(_mapping(payload.get("executorWorktree"))), "verificationTarget": dict(_mapping(payload.get("verificationTarget"))), "sourceArtifacts": _hydrate_active_source_artifacts(ctx), "lazyReadPlan": dict(_mapping(payload.get("lazyReadPlan"))), } def _compact_identity(ctx: Mapping[str, Any]) -> dict[str, Any]: return { "projectId": ctx.get("PROJECT_ID", ""), "projectRoot": ctx.get("PROJECT_ROOT", ""), "workspaceRoot": ctx.get("WORKSPACE_ROOT", ""), "taskGroup": ctx.get("TASK_GROUP", ""), "taskId": ctx.get("TASK_ID", ""), "taskKey": ctx.get("TASK_KEY", ""), "taskType": ctx.get("TASK_TYPE", ""), "segments": { "taskGroup": ctx.get("TASK_GROUP_SEGMENT", ""), "taskId": ctx.get("TASK_ID_SEGMENT", ""), "taskType": ctx.get("TASK_TYPE_SEGMENT", ""), }, } def _compact_active_task(payload: Mapping[str, Any]) -> dict[str, Any]: task = _mapping(payload.get("task")) keep = ( "projectId", "taskGroup", "taskId", "taskKey", "taskType", "workCategory", "projectRoot", ) return {key: task.get(key, "") for key in keep} def _compact_active_inputs(payload: Mapping[str, Any]) -> dict[str, bool]: instruction_set = _mapping(payload.get("instructionSet")) return { "hasClarificationResponse": bool(instruction_set.get("clarificationResponsePath")), } def _compact_active_workers(payload: Mapping[str, Any]) -> list[dict[str, Any]]: workers = payload.get("workers") if not isinstance(workers, list): return [] return [_compact_active_worker(worker) for worker in workers if isinstance(worker, Mapping)] def _compact_active_worker(worker: Mapping[str, Any]) -> dict[str, Any]: keep = ( "workerId", "role", "agent", "agentLabel", "model", "provider", "runner", "modelExecutionValue", "attemptRequired", ) return {key: worker.get(key, "") for key in keep} def _compact_project_hints() -> dict[str, str]: return { "okstraRoot": str(OKSTRA_RELATIVE), "tasksRoot": str(TASKS_RELATIVE), "discoveryRoot": str(DISCOVERY_RELATIVE), } def _hydrate_active_task(payload: Mapping[str, Any], ctx: Mapping[str, str]) -> dict[str, str]: task = dict(_mapping(payload.get("task"))) task["taskRootPath"] = ctx.get("TASK_ROOT_RELATIVE_PATH", "") return task def _hydrate_active_run( payload: Mapping[str, Any], ctx: Mapping[str, str], ) -> dict[str, str]: # `stage` and `fixRunCarry` are run inputs / derived content, not paths, so # pathHints cannot rebuild them — they survive the round trip only by being # read back off the compact payload. run = _mapping(payload.get("run")) return { "stage": str(run.get("stage", "") or ""), "fixRunCarry": str(run.get("fixRunCarry", "") or ""), "runDirectoryPath": ctx.get("RUN_DIR_RELATIVE_PATH", ""), "runManifestPath": ctx.get("RUN_MANIFEST_RELATIVE_PATH", ""), "teamStatePath": ctx.get("TEAM_STATE_RELATIVE_PATH", ""), "promptSnapshotPath": ctx.get("RUN_PROMPT_SNAPSHOT_RELATIVE_PATH", ""), "finalReportRecordPath": ctx.get("FINAL_REPORT_RECORD_RELATIVE_PATH", ""), "convergenceStatePath": ctx.get("CONVERGENCE_STATE_RELATIVE_PATH", ""), "finalStatusPath": ctx.get("FINAL_STATUS_RELATIVE_PATH", ""), "validatorScriptPath": ctx.get("RUN_VALIDATOR_RELATIVE_PATH", ""), "resumeCommandPath": ctx.get("CLAUDE_RESUME_COMMAND_RELATIVE_PATH", ""), "workerPromptsDirectoryPath": ctx.get("RUN_PROMPTS_RELATIVE_PATH", ""), "workerResultsDirectoryPath": ctx.get("WORKER_RESULTS_RELATIVE_PATH", ""), } def _hydrate_active_instruction_set( payload: Mapping[str, Any], ctx: Mapping[str, str], ) -> dict[str, str]: instruction_set = ctx.get("INSTRUCTION_SET_RELATIVE_PATH", "") inputs = _mapping(payload.get("inputs")) clarification_response = "" if inputs.get("hasClarificationResponse"): clarification_response = f"{instruction_set}/clarification-response.md" target = _mapping(payload.get("verificationTarget")) return { "path": instruction_set, "analysisPacketPath": ctx.get("ANALYSIS_PACKET_RELATIVE_PATH", ""), "taskBriefPath": f"{instruction_set}/task-brief.md", "analysisProfilePath": f"{instruction_set}/analysis-profile.md", "analysisMaterialPath": f"{instruction_set}/analysis-material.md", "referenceExpectationsPath": ctx.get("REFERENCE_EXPECTATIONS_RELATIVE_PATH", ""), "clarificationResponsePath": clarification_response, "finalReportTemplatePath": ctx.get("FINAL_REPORT_TEMPLATE_RELATIVE_PATH", ""), "finalReportSchemaPath": ctx.get("FINAL_REPORT_SCHEMA_RELATIVE_PATH", ""), "verificationTargetPath": str(target.get("path") or ""), "verificationTargetDigest": str(target.get("digest") or ""), } def _hydrate_active_workers( payload: Mapping[str, Any], ctx: Mapping[str, str], ) -> list[dict[str, Any]]: workers = payload.get("workers") if not isinstance(workers, list): return [] return [ _hydrate_active_worker(worker, ctx) for worker in workers if isinstance(worker, Mapping) ] def _hydrate_active_worker( worker: Mapping[str, Any], ctx: Mapping[str, str], ) -> dict[str, Any]: worker_id = str(worker.get("workerId", "")) hydrated = dict(worker) paths = artifacts_from_context(ctx, relative=True).get(worker_id, {}) hydrated["promptPath"] = paths.get("promptPath", "") hydrated["resultPath"] = paths.get("resultPath", "") return hydrated def _hydrate_active_error_logs(ctx: Mapping[str, str]) -> dict[str, Any]: artifacts = artifacts_from_context(ctx, relative=True) return { "runErrorsLogPath": ctx.get("RUN_ERRORS_LOG_RELATIVE_PATH", ""), "sidecarsByWorkerId": { worker_id: paths["errorsSidecarPath"] for worker_id, paths in artifacts.items() }, } def _hydrate_active_source_artifacts(ctx: Mapping[str, str]) -> dict[str, str]: return { "taskManifestPath": ctx.get("TASK_MANIFEST_RELATIVE_PATH", ""), "runContextPath": ctx.get("RUN_CONTEXT_RELATIVE_PATH", ""), "runInputsPath": ctx.get("RUN_INPUTS_RELATIVE_PATH", ""), "historyTimelinePath": ctx.get("TIMELINE_RELATIVE_PATH", ""), } def _compact_run_hints(ctx: Mapping[str, Any]) -> dict[str, Any]: return { "runRoot": ctx.get("RUN_DIR_RELATIVE_PATH", ""), "stage": ctx.get("RUN_STAGE", ""), "fileSuffix": ctx.get("RUN_FILE_SUFFIX", ""), "sequences": { "manifests": ctx.get("RUN_MANIFESTS_SEQ", ""), "prompts": ctx.get("RUN_PROMPTS_SEQ", ""), "reports": ctx.get("RUN_REPORTS_SEQ", ""), "status": ctx.get("RUN_STATUS_SEQ", ""), "state": ctx.get("RUN_STATE_SEQ", ""), "sessions": ctx.get("RUN_SESSIONS_SEQ", ""), "workerResults": ctx.get("WORKER_RESULTS_SEQ", ""), }, } def _hydrate_from_hints( identity: Mapping[str, Any], hints: Mapping[str, Any], timestamps: Any, ) -> dict[str, Any]: paths = _build_path_set(identity, hints) filenames = _filename_fields(paths) relative = _relative_fields(paths["project_root"], paths) return { **_identity_fields(identity), **_runtime_fields(paths), **_absolute_fields(paths), **filenames, **_sequence_fields(paths), **relative, **_timestamp_fields(timestamps), } def _build_path_set(identity: Mapping[str, Any], hints: Mapping[str, Any]) -> dict[str, Any]: project_root = Path(str(identity.get("projectRoot", ""))) workspace_root = Path(str(identity.get("workspaceRoot", ""))) project_hints = _mapping(hints.get("project")) run_hints = _mapping(hints.get("run")) task_root = _project_path(project_root, _task_root_hint(hints)) run_root = _project_path(project_root, str(run_hints.get("runRoot", ""))) task_type_segment = _segment(identity, "taskType") sequences = _sequences(run_hints) suffixes = _suffixes(task_type_segment, sequences) paths = { "project_root": project_root, "workspace_root": workspace_root, "okstra_root": _project_path(project_root, str(project_hints.get("okstraRoot", OKSTRA_RELATIVE))), "tasks_root": _project_path(project_root, str(project_hints.get("tasksRoot", TASKS_RELATIVE))), "discovery_dir": _project_path(project_root, str(project_hints.get("discoveryRoot", DISCOVERY_RELATIVE))), "task_root": task_root, "run_dir": run_root, "runtime_home": Path(str(_mapping(hints.get("runtime")).get("okstraHome", okstra_home()))), "task_type_segment": task_type_segment, "run_stage": str(run_hints.get("stage", "")), "run_file_suffix": str(run_hints.get("fileSuffix", suffixes["reports"])), "sequences": sequences, "suffixes": suffixes, } paths.update(_task_path_set(task_root)) paths.update(_run_path_set(run_root, task_type_segment, sequences, suffixes)) paths.update(_discovery_path_set(paths["discovery_dir"])) return paths def _task_path_set(task_root: Path) -> dict[str, Path]: instruction_set = task_root / "instruction-set" history_dir = task_root / "history" recap_dir = task_root / "recap" return { "task_manifest": task_manifest_file(task_root), "task_index": task_root / "task-index.md", "instruction_set": instruction_set, "analysis_packet": instruction_set / "analysis-packet.md", "task_qa": task_root / "qa", "runs_dir": runs_dir_of(task_root), "history_dir": history_dir, "timeline_file": task_timeline_file(task_root), "recap_dir": recap_dir, "recap_log": recap_dir / "recap-log.jsonl", "final_report_template": instruction_set / "final-report-template.md", "final_report_schema": instruction_set / "final-report-schema.json", "reference_expectations": instruction_set / "reference-expectations.md", } def _run_path_set( run_dir: Path, task_type_segment: str, sequences: Mapping[str, str], suffixes: Mapping[str, str], ) -> dict[str, Path]: run_prompts = run_dir / "prompts" worker_results = run_dir / "worker-results" return { "run_manifests": run_dir / "manifests", "run_state": run_dir / "state", "run_prompts": run_prompts, "run_reports": run_dir / "reports", "run_status": run_dir / "status", "run_sessions": run_dir / "sessions", "run_logs": run_dir / "logs", "worker_results": worker_results, # Carry is implementation-only and stage-shared (flat under the # task-type run dir), so strip a `stage-` leaf before anchoring it — # mirrors paths.py `run_carry`. "run_carry": ( (run_dir.parent if re.fullmatch(r"stage-\d+", run_dir.name) else run_dir) / "carry" if task_type_segment == "implementation" else run_dir / "carry" ), **_run_files(run_dir, run_prompts, worker_results, task_type_segment, sequences, suffixes), } def _run_files( run_dir: Path, run_prompts: Path, worker_results: Path, task_type_segment: str, sequences: Mapping[str, str], suffixes: Mapping[str, str], ) -> dict[str, Path]: run_manifests = run_dir / "manifests" run_state = run_dir / "state" run_reports = run_dir / "reports" run_status = run_dir / "status" run_sessions = run_dir / "sessions" run_logs = run_dir / "logs" return { "run_manifest_file": run_manifests / f"run-manifest{suffixes['manifests']}.json", "run_context_file": run_manifests / f"run-context-{task_type_segment}-{sequences['manifests']}.json", "run_inputs_file": run_manifests / f"run-inputs-{task_type_segment}-{sequences['manifests']}.json", "run_prompt_snapshot": run_prompts / f"lead-execution-prompt{suffixes['prompts']}.md", "duty_contract_root": run_prompts / f"duty-contracts{suffixes['prompts']}", "lead_instructions": run_prompts / f"lead-instructions{suffixes['prompts']}.md", "lead_prompt_metadata": ( run_prompts / f"lead-execution-prompt{suffixes['prompts']}.md.meta.json" ), "invocation_reservation_root": run_prompts / ".agent-invocations", "claude_worker_prompt": run_prompts / f"claude-worker-prompt{suffixes['prompts']}.md", "codex_worker_prompt": run_prompts / f"codex-worker-prompt{suffixes['prompts']}.md", "antigravity_worker_prompt": run_prompts / f"antigravity-worker-prompt{suffixes['prompts']}.md", "report_writer_worker_prompt": run_prompts / f"report-writer-worker-prompt{suffixes['prompts']}.md", "final_report": run_reports / f"final-report{suffixes['reports']}.md", "final_status": run_status / f"final{suffixes['status']}.status", "team_state": run_state / f"team-state{suffixes['state']}.json", "active_run_context": run_state / f"active-run-context{suffixes['state']}.json", "lead_events": run_state / f"lead-events-{task_type_segment}-{sequences['state']}.jsonl", "convergence_state": run_state / f"convergence-{task_type_segment}-{sequences['state']}.json", "claude_resume_command": run_sessions / f"claude-resume{suffixes['sessions']}.sh", "claude_worker_result": worker_results / f"claude-worker{suffixes['worker_results']}.md", "codex_worker_result": worker_results / f"codex-worker{suffixes['worker_results']}.md", "antigravity_worker_result": worker_results / f"antigravity-worker{suffixes['worker_results']}.md", "report_writer_worker_result": worker_results / f"report-writer-worker{suffixes['worker_results']}.md", "report_writer_narrative": worker_results / f"report-writer-narrative{suffixes['worker_results']}.md", "approval_decisions": run_state / f"approval-decisions-{task_type_segment}-{sequences['state']}.json", "design_preparation": run_state / f"design-preparation-{task_type_segment}-{sequences['state']}.json", "plan_body_verification": run_state / f"plan-body-verification-{task_type_segment}-{sequences['state']}.json", "run_errors_log": run_logs / f"errors-{task_type_segment}-{sequences['state']}.jsonl", "claude_worker_errors_sidecar": worker_results / f"claude-worker-errors{suffixes['worker_results']}.json", "codex_worker_errors_sidecar": worker_results / f"codex-worker-errors{suffixes['worker_results']}.json", "antigravity_worker_errors_sidecar": worker_results / f"antigravity-worker-errors{suffixes['worker_results']}.json", "report_writer_errors_sidecar": worker_results / f"report-writer-worker-errors{suffixes['worker_results']}.json", } def _discovery_path_set(discovery_dir: Path) -> dict[str, Path]: return { "latest_task_file": discovery_dir / "latest-task.json", "task_catalog_file": discovery_dir / "task-catalog.json", } def _identity_fields(identity: Mapping[str, Any]) -> dict[str, str]: return { "PROJECT_ID": str(identity.get("projectId", "")), "PROJECT_ROOT": str(identity.get("projectRoot", "")), "WORKSPACE_ROOT": str(identity.get("workspaceRoot", "")), "TASK_GROUP": str(identity.get("taskGroup", "")), "TASK_ID": str(identity.get("taskId", "")), "TASK_KEY": str(identity.get("taskKey", "")), "TASK_TYPE": str(identity.get("taskType", "")), "TASK_GROUP_SEGMENT": _segment(identity, "taskGroup"), "TASK_ID_SEGMENT": _segment(identity, "taskId"), "TASK_TYPE_SEGMENT": _segment(identity, "taskType"), } def _runtime_fields(paths: Mapping[str, Any]) -> dict[str, str]: runtime_home = Path(paths["runtime_home"]) lead = runtime_home / "prompts" / "lead" return { "ANALYSIS_WORKER_PREAMBLE_PATH": str( runtime_home / "templates" / "worker-prompt-preamble.md" ), "IMPLEMENTATION_WORKER_PREAMBLE_PATH": str( runtime_home / "templates" / "implementation-worker-preamble.md" ), "REPORT_WRITER_PREAMBLE_PATH": str( runtime_home / "templates" / "report-writer-prompt-preamble.md" ), "WORKER_ERROR_CONTRACT_PATH": str( runtime_home / "templates" / "worker-error-contract.md" ), "OKSTRA_LEAD_CONTRACT_PATH": str(lead / "okstra-lead-contract.md"), "OKSTRA_CONTEXT_LOADER_PATH": str(lead / "context-loader.md"), "OKSTRA_TEAM_CONTRACT_PATH": str(lead / "team-contract.md"), "OKSTRA_CONVERGENCE_PATH": str(lead / "convergence.md"), "OKSTRA_PLAN_BODY_VERIFICATION_PATH": str(lead / "plan-body-verification.md"), "OKSTRA_REPORT_WRITER_PATH": str(lead / "report-writer.md"), "OKSTRA_CODING_PREFLIGHT_DIR": str(runtime_home / "prompts" / "coding-preflight"), } def _absolute_fields(paths: Mapping[str, Any]) -> dict[str, str]: return { **_absolute_project_task_fields(paths), **_absolute_run_fields(paths), **_absolute_worker_fields(paths), } def _absolute_project_task_fields(paths: Mapping[str, Any]) -> dict[str, str]: return { "OKSTRA_ROOT": str(paths["okstra_root"]), "OKSTRA_TASKS_ROOT": str(paths["tasks_root"]), "OKSTRA_DISCOVERY_DIR": str(paths["discovery_dir"]), "TASK_ROOT": str(paths["task_root"]), "TASK_MANIFEST_PATH": str(paths["task_manifest"]), "TASK_INDEX_PATH": str(paths["task_index"]), "INSTRUCTION_SET_PATH": str(paths["instruction_set"]), "ANALYSIS_PACKET_PATH": str(paths["analysis_packet"]), "TASK_QA_PATH": str(paths["task_qa"]), "RUNS_DIR": str(paths["runs_dir"]), "HISTORY_DIR": str(paths["history_dir"]), "TIMELINE_PATH": str(paths["timeline_file"]), "RECAP_DIR": str(paths["recap_dir"]), "RECAP_LOG_PATH": str(paths["recap_log"]), "FINAL_REPORT_TEMPLATE_PATH": str(paths["final_report_template"]), "FINAL_REPORT_SCHEMA_PATH": str(paths["final_report_schema"]), "REFERENCE_EXPECTATIONS_FILE": str(paths["reference_expectations"]), "OKSTRA_LATEST_TASK_FILE": str(paths["latest_task_file"]), "OKSTRA_TASK_CATALOG_FILE": str(paths["task_catalog_file"]), } def _absolute_run_fields(paths: Mapping[str, Any]) -> dict[str, str]: return { "RUN_DIR": str(paths["run_dir"]), "RUN_STAGE": str(paths["run_stage"]), "RUN_MANIFESTS_DIR": str(paths["run_manifests"]), "RUN_STATE_DIR": str(paths["run_state"]), "RUN_PROMPTS_DIR": str(paths["run_prompts"]), "RUN_REPORTS_DIR": str(paths["run_reports"]), "RUN_STATUS_DIR": str(paths["run_status"]), "RUN_SESSIONS_DIR": str(paths["run_sessions"]), "RUN_LOGS_DIR": str(paths["run_logs"]), "WORKER_RESULTS_PATH": str(paths["worker_results"]), "RUN_CARRY_PATH": str(paths["run_carry"]), "RUN_MANIFEST_PATH": str(paths["run_manifest_file"]), "RUN_CONTEXT_FILE": str(paths["run_context_file"]), "RUN_PROMPT_SNAPSHOT_FILE": str(paths["run_prompt_snapshot"]), "DUTY_CONTRACT_ROOT": str(paths["duty_contract_root"]), "LEAD_INSTRUCTIONS_PATH": str(paths["lead_instructions"]), "LEAD_PROMPT_METADATA_PATH": str(paths["lead_prompt_metadata"]), "INVOCATION_RESERVATION_ROOT": str(paths["invocation_reservation_root"]), "FINAL_REPORT_PATH": str(paths["final_report"]), "CONVERGENCE_STATE_PATH": str(paths["convergence_state"]), "FINAL_STATUS_PATH": str(paths["final_status"]), "TEAM_STATE_PATH": str(paths["team_state"]), "ACTIVE_RUN_CONTEXT_PATH": str(paths["active_run_context"]), "LEAD_EVENTS_PATH": str(paths["lead_events"]), "CLAUDE_RESUME_COMMAND_PATH": str(paths["claude_resume_command"]), "RUN_ERRORS_LOG_PATH": str(paths["run_errors_log"]), "RUN_VALIDATOR_PATH": str(paths["workspace_root"] / "validators" / "validate-run.py"), "LATEST_RUN_PATH": str(paths["run_dir"]), } def _absolute_worker_fields(paths: Mapping[str, Any]) -> dict[str, str]: return { "CLAUDE_WORKER_PROMPT_FILE": str(paths["claude_worker_prompt"]), "CODEX_WORKER_PROMPT_FILE": str(paths["codex_worker_prompt"]), "ANTIGRAVITY_WORKER_PROMPT_FILE": str(paths["antigravity_worker_prompt"]), "REPORT_WRITER_WORKER_PROMPT_FILE": str(paths["report_writer_worker_prompt"]), "CLAUDE_WORKER_RESULT_FILE": str(paths["claude_worker_result"]), "CODEX_WORKER_RESULT_FILE": str(paths["codex_worker_result"]), "ANTIGRAVITY_WORKER_RESULT_FILE": str(paths["antigravity_worker_result"]), "REPORT_WRITER_WORKER_RESULT_FILE": str(paths["report_writer_worker_result"]), "REPORT_WRITER_NARRATIVE_FILE": str(paths["report_writer_narrative"]), "APPROVAL_DECISIONS_PATH": str(paths["approval_decisions"]), "DESIGN_PREPARATION_PATH": str(paths["design_preparation"]), "PLAN_BODY_VERIFICATION_STATE_PATH": str(paths["plan_body_verification"]), "CLAUDE_WORKER_ERRORS_SIDECAR_PATH": str(paths["claude_worker_errors_sidecar"]), "CODEX_WORKER_ERRORS_SIDECAR_PATH": str(paths["codex_worker_errors_sidecar"]), "ANTIGRAVITY_WORKER_ERRORS_SIDECAR_PATH": str(paths["antigravity_worker_errors_sidecar"]), "REPORT_WRITER_WORKER_ERRORS_SIDECAR_PATH": str(paths["report_writer_errors_sidecar"]), } def _filename_fields(paths: Mapping[str, Any]) -> dict[str, str]: return { "RUN_MANIFEST_FILENAME": Path(paths["run_manifest_file"]).name, "RUN_PROMPT_SNAPSHOT_FILENAME": Path(paths["run_prompt_snapshot"]).name, "LEAD_INSTRUCTIONS_FILENAME": Path(paths["lead_instructions"]).name, "LEAD_PROMPT_METADATA_FILENAME": Path(paths["lead_prompt_metadata"]).name, "FINAL_REPORT_FILENAME": Path(paths["final_report"]).name, "FINAL_STATUS_FILENAME": Path(paths["final_status"]).name, "CLAUDE_RESUME_COMMAND_FILENAME": Path(paths["claude_resume_command"]).name, "RUN_FILE_SUFFIX": str(paths["run_file_suffix"]), } def _sequence_fields(paths: Mapping[str, Any]) -> dict[str, str]: sequences = paths["sequences"] return { "RUN_MANIFESTS_SEQ": sequences["manifests"], "RUN_PROMPTS_SEQ": sequences["prompts"], "RUN_REPORTS_SEQ": sequences["reports"], "RUN_STATUS_SEQ": sequences["status"], "RUN_STATE_SEQ": sequences["state"], "RUN_SESSIONS_SEQ": sequences["sessions"], "WORKER_RESULTS_SEQ": sequences["worker_results"], } def _relative_fields(project_root: Path, paths: Mapping[str, Any]) -> dict[str, str]: return { "OKSTRA_DISCOVERY_RELATIVE_PATH": _rel(project_root, paths["discovery_dir"]), "OKSTRA_LATEST_TASK_RELATIVE_PATH": _rel(project_root, paths["latest_task_file"]), "OKSTRA_TASK_CATALOG_RELATIVE_PATH": _rel(project_root, paths["task_catalog_file"]), "TASK_ROOT_RELATIVE_PATH": _rel(project_root, paths["task_root"]), "TASK_MANIFEST_RELATIVE_PATH": _rel(project_root, paths["task_manifest"]), "TASK_INDEX_RELATIVE_PATH": _rel(project_root, paths["task_index"]), "INSTRUCTION_SET_RELATIVE_PATH": _rel(project_root, paths["instruction_set"]), "ANALYSIS_PACKET_RELATIVE_PATH": _rel(project_root, paths["analysis_packet"]), "RUNS_RELATIVE_PATH": _rel(project_root, paths["runs_dir"]), "HISTORY_RELATIVE_PATH": _rel(project_root, paths["history_dir"]), "TIMELINE_RELATIVE_PATH": _rel(project_root, paths["timeline_file"]), "RECAP_DIR_RELATIVE_PATH": _rel(project_root, paths["recap_dir"]), "RECAP_LOG_RELATIVE_PATH": _rel(project_root, paths["recap_log"]), "RUN_DIR_RELATIVE_PATH": _rel(project_root, paths["run_dir"]), "RUN_MANIFESTS_RELATIVE_PATH": _rel(project_root, paths["run_manifests"]), "RUN_STATE_RELATIVE_PATH": _rel(project_root, paths["run_state"]), "RUN_PROMPTS_RELATIVE_PATH": _rel(project_root, paths["run_prompts"]), "RUN_REPORTS_RELATIVE_PATH": _rel(project_root, paths["run_reports"]), "RUN_STATUS_RELATIVE_PATH": _rel(project_root, paths["run_status"]), "RUN_SESSIONS_RELATIVE_PATH": _rel(project_root, paths["run_sessions"]), **_relative_file_fields(project_root, paths), } def _relative_file_fields(project_root: Path, paths: Mapping[str, Any]) -> dict[str, str]: return { "RUN_MANIFEST_RELATIVE_PATH": _rel(project_root, paths["run_manifest_file"]), "RUN_CONTEXT_RELATIVE_PATH": _rel(project_root, paths["run_context_file"]), "RUN_INPUTS_RELATIVE_PATH": _rel(project_root, paths["run_inputs_file"]), "RUN_PROMPT_SNAPSHOT_RELATIVE_PATH": _rel(project_root, paths["run_prompt_snapshot"]), "DUTY_CONTRACT_ROOT_RELATIVE_PATH": _rel( project_root, paths["duty_contract_root"] ), "LEAD_INSTRUCTIONS_RELATIVE_PATH": _rel( project_root, paths["lead_instructions"] ), "LEAD_PROMPT_METADATA_RELATIVE_PATH": _rel( project_root, paths["lead_prompt_metadata"] ), "INVOCATION_RESERVATION_ROOT_RELATIVE_PATH": _rel( project_root, paths["invocation_reservation_root"] ), "CLAUDE_WORKER_PROMPT_RELATIVE_PATH": _rel(project_root, paths["claude_worker_prompt"]), "CODEX_WORKER_PROMPT_RELATIVE_PATH": _rel(project_root, paths["codex_worker_prompt"]), "ANTIGRAVITY_WORKER_PROMPT_RELATIVE_PATH": _rel(project_root, paths["antigravity_worker_prompt"]), "REPORT_WRITER_WORKER_PROMPT_RELATIVE_PATH": _rel(project_root, paths["report_writer_worker_prompt"]), "FINAL_REPORT_RECORD_RELATIVE_PATH": _rel(project_root, paths["final_report"]), "CONVERGENCE_STATE_RELATIVE_PATH": _rel( project_root, paths["convergence_state"] ), "FINAL_STATUS_RELATIVE_PATH": _rel(project_root, paths["final_status"]), "TEAM_STATE_RELATIVE_PATH": _rel(project_root, paths["team_state"]), "ACTIVE_RUN_CONTEXT_RELATIVE_PATH": _rel(project_root, paths["active_run_context"]), "LEAD_EVENTS_RELATIVE_PATH": _rel(project_root, paths["lead_events"]), "WORKER_RESULTS_RELATIVE_PATH": _rel(project_root, paths["worker_results"]), "RUN_CARRY_RELATIVE_PATH": _rel(project_root, paths["run_carry"]), "FINAL_REPORT_TEMPLATE_RELATIVE_PATH": _rel(project_root, paths["final_report_template"]), "FINAL_REPORT_SCHEMA_RELATIVE_PATH": _rel(project_root, paths["final_report_schema"]), "REFERENCE_EXPECTATIONS_RELATIVE_PATH": _rel(project_root, paths["reference_expectations"]), "CLAUDE_RESUME_COMMAND_RELATIVE_PATH": _rel(project_root, paths["claude_resume_command"]), "RUN_VALIDATOR_RELATIVE_PATH": _rel(project_root, paths["workspace_root"] / "validators" / "validate-run.py"), "CLAUDE_WORKER_RESULT_RELATIVE_PATH": _rel(project_root, paths["claude_worker_result"]), "CODEX_WORKER_RESULT_RELATIVE_PATH": _rel(project_root, paths["codex_worker_result"]), "ANTIGRAVITY_WORKER_RESULT_RELATIVE_PATH": _rel(project_root, paths["antigravity_worker_result"]), "REPORT_WRITER_WORKER_RESULT_RELATIVE_PATH": _rel(project_root, paths["report_writer_worker_result"]), "REPORT_WRITER_NARRATIVE_RELATIVE_PATH": _rel(project_root, paths["report_writer_narrative"]), "APPROVAL_DECISIONS_RELATIVE_PATH": _rel(project_root, paths["approval_decisions"]), "DESIGN_PREPARATION_RELATIVE_PATH": _rel(project_root, paths["design_preparation"]), "PLAN_BODY_VERIFICATION_STATE_RELATIVE_PATH": _rel(project_root, paths["plan_body_verification"]), "RUN_ERRORS_LOG_RELATIVE_PATH": _rel(project_root, paths["run_errors_log"]), "CLAUDE_WORKER_ERRORS_SIDECAR_RELATIVE_PATH": _rel(project_root, paths["claude_worker_errors_sidecar"]), "CODEX_WORKER_ERRORS_SIDECAR_RELATIVE_PATH": _rel(project_root, paths["codex_worker_errors_sidecar"]), "ANTIGRAVITY_WORKER_ERRORS_SIDECAR_RELATIVE_PATH": _rel(project_root, paths["antigravity_worker_errors_sidecar"]), "REPORT_WRITER_WORKER_ERRORS_SIDECAR_RELATIVE_PATH": _rel(project_root, paths["report_writer_errors_sidecar"]), "LATEST_RUN_RELATIVE_PATH": _rel(project_root, paths["run_dir"]), } def _timestamp_fields(timestamps: Any) -> dict[str, str]: values = _mapping(timestamps) return { "RUN_TIMESTAMP_ISO": str(values.get("runTimestampIso", "")), "TASK_DATE": str(values.get("taskDate", "")), } def _task_root_hint(hints: Mapping[str, Any]) -> str: task_hints = _mapping(hints.get("task")) return str(task_hints.get("taskRoot", "")) def _segment(identity: Mapping[str, Any], name: str) -> str: segments = _mapping(identity.get("segments")) return str(segments.get(name, "")) def _mapping(value: Any) -> Mapping[str, Any]: return value if isinstance(value, Mapping) else {} def _sequences(run_hints: Mapping[str, Any]) -> dict[str, str]: raw = _mapping(run_hints.get("sequences")) return { "manifests": str(raw.get("manifests", "")), "prompts": str(raw.get("prompts", "")), "reports": str(raw.get("reports", "")), "status": str(raw.get("status", "")), "state": str(raw.get("state", "")), "sessions": str(raw.get("sessions", "")), "worker_results": str(raw.get("workerResults", "")), } def _suffixes(task_type_segment: str, sequences: Mapping[str, str]) -> dict[str, str]: return {key: f"-{task_type_segment}-{seq}" for key, seq in sequences.items()} def _project_path(project_root: Path, value: str) -> Path: path = Path(value) return path if path.is_absolute() else project_root / path def _rel(project_root: Path, target: Path) -> str: try: return str(target.resolve().relative_to(project_root.resolve())) except (ValueError, OSError): return str(target)