feat: persist planning artifacts for runs

Author: xiaojiou176

apps/orchestrator/src/cortexpilot_orch/api/main_pm_intake_helpers.py

@@ -17,8 +17,9 @@
 from cortexpilot_orch.config import load_config
 from cortexpilot_orch.contract.compiler import build_role_binding_summary, sync_role_contract
 from cortexpilot_orch.observability.logger import log_event
-from cortexpilot_orch.planning.intake import IntakeService
+from cortexpilot_orch.planning.intake import IntakeService, _build_wave_plan, _build_worker_prompt_contracts
 from cortexpilot_orch.store.intake_store import IntakeStore
+from cortexpilot_orch.store.run_store import RunStore
 
 
 _TRUTHY_VALUES = {"1", "true", "yes", "y", "on"}
@@ -118,6 +119,55 @@ def _strip_intake_only_contract_fields(contract: dict[str, Any]) -> dict[str, An
     return sanitized
 
 
+def _safe_read_intake_store_payload(store: object, method_name: str, intake_id: str) -> dict[str, Any]:
+    reader = getattr(store, method_name, None)
+    if not callable(reader):
+        return {}
+    try:
+        payload = reader(intake_id)
+    except Exception:
+        return {}
+    return payload if isinstance(payload, dict) else {}
+
+
+def _persist_planning_artifacts_for_run(
+    *,
+    intake_id: str,
+    run_id: str,
+    runs_root: Path,
+) -> list[str]:
+    intake_store = IntakeStore()
+    intake_payload = _safe_read_intake_store_payload(intake_store, "read_intake", intake_id)
+    response_payload = _safe_read_intake_store_payload(intake_store, "read_response", intake_id)
+    plan_bundle = response_payload.get("plan_bundle") if isinstance(response_payload.get("plan_bundle"), dict) else None
+    if not intake_payload or not isinstance(plan_bundle, dict):
+        return []
+
+    run_store = RunStore(runs_root=runs_root)
+    artifacts_to_write: list[tuple[str, Any]] = [
+        ("planning_wave_plan.json", _build_wave_plan(plan_bundle)),
+        ("planning_worker_prompt_contracts.json", _build_worker_prompt_contracts(plan_bundle, intake_payload)),
+    ]
+    written: list[str] = []
+    for filename, payload in artifacts_to_write:
+        if payload in ({}, [], None):
+            continue
+        run_store.write_artifact(run_id, filename, json.dumps(payload, ensure_ascii=False, indent=2))
+        written.append(filename)
+
+    if written:
+        run_store.append_event(
+            run_id,
+            {
+                "level": "INFO",
+                "event": "PLANNING_ARTIFACTS_WRITTEN",
+                "run_id": run_id,
+                "meta": {"intake_id": intake_id, "artifacts": written},
+            },
+        )
+    return written
+
+
 def configure_pm_session_aggregation(
     *,
     runs_root_fn: Callable[[], Path],
@@ -597,10 +647,16 @@ def _execute_in_background() -> None:
         )
 
     IntakeStore().append_event(intake_id, {"event": "INTAKE_RUN", "run_id": run_id})
+    planning_artifacts = _persist_planning_artifacts_for_run(
+        intake_id=intake_id,
+        run_id=run_id,
+        runs_root=runs_root,
+    )
     return {
         "ok": True,
         "run_id": run_id,
         "contract_path": str(contract_path),
         "strict_acceptance": bool(runtime_options.get("strict_acceptance", False)),
         "role_binding_summary": build_role_binding_summary(contract),
+        "planning_artifacts": planning_artifacts,
     }

apps/orchestrator/tests/test_main_pm_intake_helpers_branches.py

@@ -463,6 +463,116 @@ def execute_task(contract_path: Path, mock_mode: bool = False) -> str:
     assert observed_contract["runtime_options"]["strict_acceptance"] is True
 
 
+def test_run_intake_persists_planning_artifacts_into_run_bundle(monkeypatch, tmp_path: Path) -> None:
+    runs_root = tmp_path / "runs"
+    runtime_contract_root = tmp_path / ".runtime-cache" / "cortexpilot" / "contracts"
+    intake_payload = {
+        "objective": "Ship one planning artifact bridge",
+        "constraints": ["truthful-public-surface"],
+        "search_queries": ["command tower planning artifact"],
+    }
+    response_payload = {
+        "plan_bundle": {
+            "bundle_id": "bundle-1",
+            "objective": "Ship one planning artifact bridge",
+            "owner_agent": {"role": "PM", "agent_id": "pm-1"},
+            "plans": [
+                {
+                    "plan_id": "worker-1",
+                    "assigned_agent": {"role": "WORKER", "agent_id": "worker-1"},
+                    "spec": "Persist the planning artifact into the run bundle.",
+                    "allowed_paths": ["apps/orchestrator"],
+                    "acceptance_tests": [{"name": "pytest", "cmd": "python3 -m pytest -q", "must_pass": True}],
+                    "mcp_tool_set": ["codex"],
+                    "required_outputs": [{"name": "task_result.json", "type": "report"}],
+                }
+            ],
+        }
+    }
+    intake_events: list[tuple[str, dict[str, object]]] = []
+
+    class _Store:
+        def append_event(self, intake_id: str, payload: dict[str, object]) -> None:
+            intake_events.append((intake_id, payload))
+
+        def read_intake(self, intake_id: str) -> dict[str, object]:
+            assert intake_id == "persist"
+            return intake_payload
+
+        def read_response(self, intake_id: str) -> dict[str, object]:
+            assert intake_id == "persist"
+            return response_payload
+
+    monkeypatch.setattr(helpers, "IntakeStore", lambda: _Store())
+    monkeypatch.setattr(
+        helpers,
+        "load_config",
+        lambda: types.SimpleNamespace(
+            repo_root=tmp_path,
+            runs_root=runs_root,
+            contract_root=tmp_path / "contracts",
+            runtime_contract_root=runtime_contract_root,
+        ),
+    )
+
+    class _BuildOK:
+        def build_contract(self, intake_id: str) -> dict[str, object]:
+            assert intake_id == "persist"
+            return {
+                "task_id": "task-persist",
+                "owner_agent": {"role": "PM", "agent_id": "pm-1"},
+                "assigned_agent": {"role": "WORKER", "agent_id": "worker-1"},
+                "inputs": {"spec": "repro", "artifacts": []},
+                "required_outputs": [{"name": "task_result.json", "type": "json", "acceptance": "ok"}],
+                "allowed_paths": ["apps/orchestrator"],
+                "forbidden_actions": [],
+                "acceptance_tests": [{"name": "pytest", "cmd": "python3 -m pytest -q", "must_pass": True}],
+                "tool_permissions": {
+                    "filesystem": "workspace-write",
+                    "shell": "on-request",
+                    "network": "deny",
+                    "mcp_tools": ["codex"],
+                },
+                "mcp_tool_set": ["codex"],
+                "timeout_retry": {"timeout_sec": 60, "max_retries": 0, "retry_backoff_sec": 0},
+                "rollback": {"strategy": "git_reset_hard", "baseline_ref": "HEAD"},
+                "evidence_links": [],
+                "log_refs": {"run_id": "", "paths": {}},
+            }
+
+    class _Orchestrator:
+        @staticmethod
+        def execute_task(contract_path: Path, mock_mode: bool = False) -> str:
+            del mock_mode
+            payload = json.loads(contract_path.read_text(encoding="utf-8"))
+            store = RunStore(runs_root=runs_root)
+            run_id = store.create_run(str(payload.get("task_id") or "task"))
+            store.write_manifest(run_id, {"run_id": run_id, "task_id": payload.get("task_id"), "status": "RUNNING", "repo": {}})
+            return run_id
+
+    result = helpers.run_intake(
+        "persist",
+        {"mock": True},
+        intake_service_cls=_BuildOK,
+        orchestration_service=_Orchestrator(),
+        error_detail_fn=lambda code: {"code": code},
+        current_request_id_fn=lambda: "req-persist",
+    )
+
+    run_id = result["run_id"]
+    wave_plan = json.loads((runs_root / run_id / "artifacts" / "planning_wave_plan.json").read_text(encoding="utf-8"))
+    worker_contracts = json.loads(
+        (runs_root / run_id / "artifacts" / "planning_worker_prompt_contracts.json").read_text(encoding="utf-8")
+    )
+
+    assert result["planning_artifacts"] == ["planning_wave_plan.json", "planning_worker_prompt_contracts.json"]
+    assert wave_plan["wave_id"] == "bundle-1"
+    assert wave_plan["objective"] == "Ship one planning artifact bridge"
+    assert worker_contracts[0]["prompt_contract_id"] == "worker-1"
+    assert worker_contracts[0]["continuation_policy"]["on_blocked"] == "spawn_independent_temporary_unblock_task"
+    assert intake_events[-1] == ("persist", {"event": "INTAKE_RUN", "run_id": run_id})
+
+
 def test_build_role_binding_summary_marks_skills_and_mcp_registry_refs_as_registry_backed() -> None:
     summary = build_role_binding_summary(
         {