feat: register planning artifacts in run manifests

Author: xiaojiou176

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

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import hashlib
 import json
 import threading
 from datetime import datetime, timezone
@@ -119,6 +120,29 @@ def _strip_intake_only_contract_fields(contract: dict[str, Any]) -> dict[str, An
     return sanitized
 
 
+def _artifact_ref_for_path(path: Path, *, rel_path: str, name: str, media_type: str = "application/json") -> dict[str, Any]:
+    payload = path.read_bytes()
+    return {
+        "name": name,
+        "path": rel_path,
+        "sha256": hashlib.sha256(payload).hexdigest(),
+        "media_type": media_type,
+        "size_bytes": len(payload),
+    }
+
+
+def _append_manifest_artifact(manifest: dict[str, Any], ref: dict[str, Any]) -> None:
+    artifacts = manifest.get("artifacts") if isinstance(manifest.get("artifacts"), list) else []
+    key = (str(ref.get("name") or ""), str(ref.get("path") or ""))
+    for item in artifacts:
+        if not isinstance(item, dict):
+            continue
+        if (str(item.get("name") or ""), str(item.get("path") or "")) == key:
+            return
+    artifacts.append(ref)
+    manifest["artifacts"] = artifacts
+
+
 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):
@@ -144,18 +168,33 @@ def _persist_planning_artifacts_for_run(
         return []
 
     run_store = RunStore(runs_root=runs_root)
+    run_dir = run_store.run_dir(run_id)
     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] = []
+    artifact_refs: list[dict[str, Any]] = []
     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))
+        artifact_path = run_store.write_artifact(run_id, filename, json.dumps(payload, ensure_ascii=False, indent=2))
         written.append(filename)
+        artifact_refs.append(
+            _artifact_ref_for_path(
+                artifact_path,
+                rel_path=f"artifacts/{filename}",
+                name=filename.removesuffix(".json"),
+            )
+        )
 
     if written:
+        manifest_path = run_dir / "manifest.json"
+        manifest = _read_json_file(manifest_path)
+        if manifest:
+            for ref in artifact_refs:
+                _append_manifest_artifact(manifest, ref)
+            manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
         run_store.append_event(
             run_id,
             {

apps/orchestrator/src/cortexpilot_orch/scheduler/scheduler_bridge_contract.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from collections.abc import Callable
+import hashlib
 import json
 from pathlib import Path
 from typing import Any
@@ -9,6 +10,29 @@
 from cortexpilot_orch.store.run_store import RunStore
 
 
+def _artifact_ref_for_path(path: Path, *, rel_path: str, name: str, media_type: str = "application/json") -> dict[str, Any]:
+    payload = path.read_bytes()
+    return {
+        "name": name,
+        "path": rel_path,
+        "sha256": hashlib.sha256(payload).hexdigest(),
+        "media_type": media_type,
+        "size_bytes": len(payload),
+    }
+
+
+def _append_manifest_artifact(manifest: dict[str, Any], ref: dict[str, Any]) -> None:
+    artifacts = manifest.get("artifacts") if isinstance(manifest.get("artifacts"), list) else []
+    key = (str(ref.get("name") or ""), str(ref.get("path") or ""))
+    for item in artifacts:
+        if not isinstance(item, dict):
+            continue
+        if (str(item.get("name") or ""), str(item.get("path") or "")) == key:
+            return
+    artifacts.append(ref)
+    manifest["artifacts"] = artifacts
+
+
 class ContractStateWriter:
     def __init__(
         self,
@@ -215,6 +239,16 @@ def persist_contract_state(
         "prompt_artifact.json",
         json.dumps(prompt_artifact, ensure_ascii=False, indent=2),
     )
+    if manifest is not None:
+        _append_manifest_artifact(
+            manifest,
+            _artifact_ref_for_path(
+                prompt_artifact_path,
+                rel_path="artifacts/prompt_artifact.json",
+                name="prompt_artifact",
+            ),
+        )
+        write_manifest_fn(store, run_id, manifest)
     store.append_event(
         run_id,
         {

apps/orchestrator/tests/test_main_pm_intake_helpers_branches.py

@@ -564,12 +564,16 @@ def execute_task(contract_path: Path, mock_mode: bool = False) -> str:
     worker_contracts = json.loads(
         (runs_root / run_id / "artifacts" / "planning_worker_prompt_contracts.json").read_text(encoding="utf-8")
     )
+    manifest = json.loads((runs_root / run_id / "manifest.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"
+    artifact_names = [item["name"] for item in manifest["artifacts"]]
+    assert "planning_wave_plan" in artifact_names
+    assert "planning_worker_prompt_contracts" in artifact_names
     assert intake_events[-1] == ("persist", {"event": "INTAKE_RUN", "run_id": run_id})
 
 

apps/orchestrator/tests/test_scheduler_bridge_runtime.py

@@ -163,3 +163,5 @@ def test_persist_contract_state_writes_role_binding_summary_to_manifest(tmp_path
     assert prompt_artifact["run_id"] == run_id
     assert prompt_artifact["task_id"] == "task-role-binding-summary"
     assert prompt_artifact["role_binding_summary"] == build_role_binding_summary(contract)
+    artifact_names = [item["name"] for item in written["artifacts"]]
+    assert "prompt_artifact" in artifact_names