feat: persist prompt artifacts per run

Author: xiaojiou176

apps/orchestrator/src/cortexpilot_orch/contract/compiler.py

@@ -421,6 +421,53 @@ def build_role_binding_summary(contract: dict[str, Any]) -> dict[str, Any]:
     }
 
 
+def build_prompt_artifact(
+    contract: dict[str, Any],
+    *,
+    run_id: str = "",
+    task_id: str = "",
+) -> dict[str, Any]:
+    role_contract = contract.get("role_contract") if isinstance(contract.get("role_contract"), dict) else {}
+    if not role_contract:
+        role_contract = _build_role_contract(contract, _load_agent_registry())
+    assigned_agent = contract.get("assigned_agent") if isinstance(contract.get("assigned_agent"), dict) else {}
+    role = str(
+        assigned_agent.get("role")
+        or (role_contract.get("identity", {}) if isinstance(role_contract.get("identity"), dict) else {}).get("role")
+        or "WORKER"
+    ).strip().upper() or "WORKER"
+    role_contract = _merge_role_config_defaults(
+        role_contract,
+        _find_role_config_defaults(_load_role_config_registry(), role),
+    )
+    identity = role_contract.get("identity") if isinstance(role_contract.get("identity"), dict) else {}
+    runtime_binding_raw = role_contract.get("runtime_binding") if isinstance(role_contract.get("runtime_binding"), dict) else {}
+    runtime_binding = {
+        "runner": _normalize_optional_ref(runtime_binding_raw.get("runner")),
+        "provider": _normalize_optional_ref(runtime_binding_raw.get("provider")),
+        "model": _normalize_optional_ref(runtime_binding_raw.get("model")),
+    }
+    resolved_task_id = str(task_id or contract.get("task_id") or "").strip()
+    return {
+        "artifact_type": "prompt_artifact",
+        "version": "v1",
+        "source": "contract-derived",
+        "execution_authority": "task_contract",
+        "run_id": str(run_id or "").strip(),
+        "task_id": resolved_task_id,
+        "assigned_agent": {
+            "role": role,
+            "agent_id": str(identity.get("agent_id") or assigned_agent.get("agent_id") or "").strip(),
+        },
+        "purpose": str(role_contract.get("purpose") or "").strip(),
+        "system_prompt_ref": _normalize_optional_ref(role_contract.get("system_prompt_ref")),
+        "skills_bundle_ref": _normalize_optional_ref(role_contract.get("skills_bundle_ref")),
+        "mcp_bundle_ref": _normalize_optional_ref(role_contract.get("mcp_bundle_ref")),
+        "runtime_binding": runtime_binding,
+        "role_binding_summary": build_role_binding_summary(contract),
+    }
+
+
 def _build_role_contract(contract: dict[str, Any], registry: dict[str, Any] | None) -> dict[str, Any]:
     assigned_agent = contract.get("assigned_agent") if isinstance(contract.get("assigned_agent"), dict) else {}
     role = str(assigned_agent.get("role") or "WORKER").strip().upper() or "WORKER"

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

@@ -1,10 +1,11 @@
 from __future__ import annotations
 
 from collections.abc import Callable
+import json
 from pathlib import Path
 from typing import Any
 
-from cortexpilot_orch.contract.compiler import build_role_binding_summary
+from cortexpilot_orch.contract.compiler import build_prompt_artifact, build_role_binding_summary
 from cortexpilot_orch.store.run_store import RunStore
 
 
@@ -208,5 +209,21 @@ def persist_contract_state(
             )
     store.write_task_contract(run_id, task_id, contract)
     store.write_active_contract(run_id, contract)
+    prompt_artifact = build_prompt_artifact(contract, run_id=run_id, task_id=task_id)
+    prompt_artifact_path = store.write_artifact(
+        run_id,
+        "prompt_artifact.json",
+        json.dumps(prompt_artifact, ensure_ascii=False, indent=2),
+    )
+    store.append_event(
+        run_id,
+        {
+            "level": "INFO",
+            "event": "PROMPT_ARTIFACT_WRITTEN",
+            "run_id": run_id,
+            "task_id": task_id,
+            "meta": {"path": str(prompt_artifact_path.relative_to(store.run_dir(run_id)))},
+        },
+    )
     if ensure_evidence_bundle_fn is not None and failure_reason:
         ensure_evidence_bundle_fn(store, run_id, contract, failure_reason)

apps/orchestrator/tests/test_scheduler_bridge_runtime.py

@@ -155,3 +155,11 @@ def test_persist_contract_state_writes_role_binding_summary_to_manifest(tmp_path
 
     written = json.loads((store._runs_root / run_id / "manifest.json").read_text(encoding="utf-8"))
     assert written["role_binding_summary"] == build_role_binding_summary(contract)
+    prompt_artifact = json.loads(
+        (store._runs_root / run_id / "artifacts" / "prompt_artifact.json").read_text(encoding="utf-8")
+    )
+    assert prompt_artifact["artifact_type"] == "prompt_artifact"
+    assert prompt_artifact["execution_authority"] == "task_contract"
+    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)

docs/architecture/runtime-topology.md

@@ -72,6 +72,9 @@ flowchart LR
   `workflow_case_read_model` directly for operator inspection, but those UI
   cards remain read-only mirrors below `task_contract`.
 - Runtime artifacts (`manifest`, `events.jsonl`, reports) are generated per run.
+- Runs may now also persist `artifacts/prompt_artifact.json`, a contract-derived
+  snapshot of prompt/bundle/runtime-binding refs for that run. It is a
+  read-only audit artifact, not a second execution authority source.
 - Run detail views may now include derived decision packs such as
   `incident_pack.json`, while approval queues synthesize `approval_pack`
   summaries from run events plus manifest metadata. These are derived operator