审计/链路追踪落地（可回放审计记录 FROZEN v1）（Refs #17）

issue #17 要求「每步 trace 可回放（谁/何模型/何工具/何审批）+ schema 冻结 + 测试」。
本仓已持久化事件流 + 任务谱系 + 每任务 usage(model_id) + 审批；本 PR 把它们规整为
统一、有序、可回放的审计记录并冻结 schema。

代码：
- orchestrator/audit.py（新，纯模块，无 Redis/WS/FastAPI/模型）：build_audit_trace(
  events, task_facts, approvals, lineage) 逐事件产审计记录（who/when/model_id/tool_count/
  approval{id,decision}/result/lineage），audit_id 由序位确定（非随机，可字节级复现）；
  replay(records) 产人读步骤行。无内容、无密钥、缺信号不伪造（model/tool 缺则 null）。
- orchestrator/main.py：build_audit_trace_for_run（装配器，从 list_events + 任务 usage +
  run.approvals 取数）+ 读接口 GET …/{id}/audit（三别名路由，复用既有鉴权）。

文档：docs/integration/audit-trace-schema.md → FROZEN v1：§3.1 回放装配、§5 冻结记录形；
诚实标注**有意排除**（prompt/代码原文、model 请求响应体刻意不留痕——无内容原则；
无 SK/MCP 工具层故无工具名谱系），按规则 #9 不伪造、不在本次扩展。

测试：scripts/test-audit-trace.py（纯模块 + 集成）：逐步 who/model/tool/approval 重建、
result 归类、有序回放、**断言无 secret_ref/azkv/明文泄漏**；接入 CI。

影响范围：仅 agent_swarm（orchestrator 新增只读审计接口 + 纯模块 + 文档 + 测试 + CI）。
- Manager：新增只读 GET …/{id}/audit；不改回调/契约/计费/审批链。
- 密钥/审计：审计记录只含非内容元数据 + secret_ref 永不写入；归因按 user/channelId，不引入 tenant。

Refs #17

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

.github/workflows/ci.yml

@@ -59,6 +59,10 @@ jobs:
         env: { REDIS_FAKE: "1" }
         run: python scripts/test-contract-events.py
 
+      - name: Audit / lineage trace (replayable) (#17)
+        env: { REDIS_FAKE: "1" }
+        run: python scripts/test-audit-trace.py
+
       - name: Benchmark metric formulas (v2.1)
         env: { REDIS_FAKE: "1" }
         run: python scripts/test-benchmark-metrics.py

docs/integration/audit-trace-schema.md

@@ -1,6 +1,8 @@
 # 审计与链路追踪 Schema（Audit / Lineage / Trace）
 
-> 状态：**DRAFT / 待对齐 Audit & Compliance Team**。
+> 状态：**FROZEN v1（可回放审计记录已落地）** —— 回应 issue #17。每步 trace（谁/何时/何模型/何工具/何审批/结果/lineage）可由 `orchestrator/audit.py` 的纯装配器从已持久化的事件流 + 任务谱系 + 审批重建并按序回放；读接口 `GET …/{id}/audit`。契约测试 `scripts/test-audit-trace.py`。
+>
+> **诚实边界（规则 #9，本次不扩展）**：审计记录**只含非内容元数据**——prompt 原文/代码/对话**刻意不留痕**（与回调脱敏同源原则）；model trace = `model_id` + token 计数（不留请求/响应体）；tool trace = 文件/Git 资源动作计数（本运行时**无 SK/MCP 工具层**，故无工具名谱系）。这些为有意取舍/后续层，不在本次冻结内。
 >
 > 依据：`heicode-mananger/docs/heicode.md §五/§九`、`docs/heicode-runtime-auth-newapi-secret-design.md §三`。配套：[`event-schema.md`](./event-schema.md)、[`security-boundary.md`](./security-boundary.md)。
 
@@ -40,36 +42,50 @@
 | 回调投递审计 | `callback_attempts`、`/diagnostics` |
 | 用量归因 | `usage` + `X-Agent-*` 归因头（见 usage-billing） |
 | 脱敏 | `_redact_sensitive`（保留 `secret_ref`，脱敏明文密钥） |
+| **可回放审计装配 + 回放** | `orchestrator/audit.py`：`build_audit_trace(events, task_facts, approvals, lineage)` 纯函数把事件流 + 任务（含 `usage.model_id`/工具计数）+ 审批规整为有序审计记录（§5 冻结形）；`replay(records)` 产人读步骤行。读接口 `GET …/{id}/audit`（`build_audit_trace_for_run`）。**纯净、确定性、无内容/无密钥**。 |
 
-## 4. 缺口
+### 3.1 回放装配（issue #17）
+
+`audit.build_audit_trace` 对每条事件产一条审计记录，按存储顺序排列，`audit_id` 由序位确定（`aud_000001…`，非随机，**同一 run 字节级可复现**）。逐步可答：
+- **谁**：`actor.agent_instance_id`（事件携带）/ `actor.agent_role`（任务谱系回填）。
+- **何模型**：`model.model_id`（取自任务 `usage.model_id`，缺则 `null`，不伪造）。
+- **何工具**：`tool.count`（取自任务 `tool_calls`/`tool_count`；无工具层时为 `null`）。
+- **何审批**：`approval.{approval_id, decision}`（取自 payload + `run.approvals`，**绝不含 `secret_ref`**）。
+- **结果**：`result` 由 event_type 归类（completed→success / failed→failed / blocked|approval.rejected→blocked / stopped→stopped / 其余→info）。
+- **lineage**：四 ID 全量。每条 `redacted: true`。
+
+## 4. 覆盖与边界
 
 | 追踪项 | 状态 |
 |---|---|
 | Task lineage / execution trace | ✅ 已实现（事件流 + 任务字段） |
-| Approval trace | ✅ 已实现 |
-| Prompt trace（每次模型输入提示留痕） | 🔴 未实现（仅记 `model_id`/用量，不留 prompt 原文） |
-| Model trace（请求/响应、参数、provider） | 🟡 部分（`model_id`/tokens；无完整请求响应留痕） |
-| Tool trace（工具调用谱系） | 🔴 未实现（无 SK/MCP 工具计量，`sk_tool.*` 仅在 schema 预留） |
-| 统一审计查询字段 / 留存策略 | 🟡 事件可查；标准查询字段与留存期未定义 |
-| 交付回放（replay） | 🟡 事件流可顺序回放生命周期；无独立 replay 接口/快照格式 |
+| Approval trace | ✅ 已实现（审计记录 `approval.{id,decision}`） |
+| Model trace（`model_id` + token 计数） | ✅ 已实现（审计记录 `model.model_id`；无请求/响应体——见下「有意排除」） |
+| Tool trace（计数） | ✅ 计数已实现（`tool.count`）；工具名谱系 N/A（无 SK/MCP 工具层） |
+| 统一审计记录 schema + 回放 | ✅ 已冻结（§5）+ 可回放（`audit.build_audit_trace`/`replay`，`GET …/{id}/audit`） |
+| **有意排除（非缺口）** | Prompt/代码/对话**原文**——刻意不留痕（无内容原则）；model 请求/响应体——同上；SK/MCP 工具名谱系——无工具层。这些属取舍/后续层，按规则 #9 不伪造、不在本次冻结。 |
+| 留存期 / 合规归档策略 | 🟡 由 Audit & Compliance Team 定（留存期、外部归档），非本仓编排器。 |
 
-## 5. 建议审计记录 Schema（草案，待 Audit Team 冻结）
+## 5. 审计记录 Schema（FROZEN v1 —— `orchestrator/audit.py` 产出形）
 
 ```jsonc
 {
-  "audit_id": "aud_...",
+  "audit_id": "aud_000001",                  // 序位确定（非随机），同一 run 可复现
   "occurred_at": "2026-06-08T...Z",
-  "actor": { "user_id": "...", "channel_id": "..." },      // 谁（按 user/channelId，非 tenant）
-  "subject": { "agent_role": "...", "agent_instance_id": "...", "task_id": "..." }, // 哪个子 Agent / 任务
-  "action": "task.completed | approval.decided | resource.accessed | ...",
-  "resource": { "type": "git|database|storage|model", "ref": "non-secret-id", "secret_ref": "azkv://..." },
-  "approval_id": "approval_... | null",
+  "action": "task.completed",                // = event_type
+  "actor": { "agent_role": "impl", "agent_instance_id": "agent-7" },  // 谁
+  "task_id": "swarm-...-t | null",
+  "model": { "model_id": "gpt-x" } | null,    // 何模型（缺则 null，不伪造）
+  "tool":  { "count": 2 } | null,             // 何工具（计数；无工具层为 null）
+  "approval": { "approval_id": "approval_...", "decision": "approved|rejected" } | null, // 何审批（无 secret_ref）
+  "result": "success | failed | blocked | stopped | info",            // 结果
   "lineage": { "manager_deployment_id": "...", "deployment_id": "...", "swarm_id": "...", "correlation_id": "..." },
-  "result": "success | failed | blocked",
   "redacted": true
 }
 ```
 
+> 归因主轴为 `user.id`/`channelId`（见 usage-billing），**不引入 tenant**；actor 在审计记录中以 `agent_role`/`agent_instance_id` 表达执行主体，用户主体由 lineage→Manager 侧关联。
+
 ## 6. 待对齐对象
 
-Audit & Compliance Team：prompt/model/tool lineage 是否强制留痕及留存期、统一审计查询字段、交付回放快照格式、日志留存与合规要求；与 `telemetry-architecture`（benchmark 目录）数据源对齐。
+Audit & Compliance Team：留存期与外部归档策略、合规要求；prompt/model 请求体留痕是否在他处（受隐私/安全约束，本仓默认不留）；与 `telemetry-architecture`（benchmark 目录）数据源对齐。

orchestrator/audit.py

@@ -0,0 +1,173 @@
+"""Audit / lineage trace assembler (issue #17).
+
+Reconstructs a **replayable, ordered** audit trail for one swarm run from data the runtime
+already persists — the event stream, task lineage, per-task model usage, and approvals — and
+normalizes it into a single frozen record shape that answers, per step:
+**who / when / which model / which tool / which approval / result / lineage**.
+
+Design (mirrors convergence.py / cross_review.py):
+  * **Pure** — no Redis / WebSocket / FastAPI / model calls. The caller fetches events/tasks/
+    approvals and passes them in; this module only transforms. That makes it hermetically testable
+    and keeps the audit logic free of I/O side effects.
+  * **No content, no secrets** (rule: audit records carry only non-content metadata). Prompt text,
+    code, and conversation are deliberately NOT recorded — only `model_id` + token counts + tool
+    counts. Every record is marked `redacted: true`. `secret_ref` is never copied in.
+  * **Deterministic** — `audit_id` is derived from the ordered position, not a random id, so the
+    same run replays byte-identically.
+
+Frozen record schema: see docs/integration/audit-trace-schema.md §5 (FROZEN v1).
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
+
+
+# event_type -> coarse result classification (the "结果如何" column).
+_RESULT_BY_EVENT = {
+    "task.completed": "success",
+    "swarm.completed": "success",
+    "approval.approved": "success",
+    "artifact.created": "success",
+    "task.failed": "failed",
+    "swarm.failed": "failed",
+    "task.blocked": "blocked",
+    "approval.rejected": "blocked",
+    "swarm.stopped": "stopped",
+}
+
+
+def classify_result(event_type: str) -> str:
+    """Map an event_type to a coarse result. Unknown -> 'info' (no fabrication)."""
+    return _RESULT_BY_EVENT.get(event_type, "info")
+
+
+@dataclass
+class TaskAuditFacts:
+    """Per-task facts the caller extracts from task state + parsed result.usage.
+
+    model_id / tool_count come from the agent-reported `usage` block; left None/0 when the
+    agent did not report them (never fabricated).
+    """
+    agent_role: Optional[str] = None
+    assigned_agent_id: Optional[str] = None
+    model_id: Optional[str] = None
+    tool_count: int = 0
+    source: Optional[str] = None
+    parent_task_id: Optional[str] = None
+    root_task_id: Optional[str] = None
+
+
+@dataclass
+class AuditRecord:
+    audit_id: str
+    occurred_at: Optional[str]
+    action: str                                  # = event_type
+    actor: Dict[str, Optional[str]]              # who: agent_role / agent_instance_id
+    task_id: Optional[str]
+    model: Optional[Dict[str, Any]]              # which model: {model_id} (+counts), or None
+    tool: Optional[Dict[str, Any]]               # which tool: {count} (no SK/MCP layer yet), or None
+    approval: Optional[Dict[str, Any]]           # which approval: {approval_id, decision}, or None
+    result: str
+    lineage: Dict[str, Optional[str]]
+    redacted: bool = True
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "audit_id": self.audit_id,
+            "occurred_at": self.occurred_at,
+            "action": self.action,
+            "actor": self.actor,
+            "task_id": self.task_id,
+            "model": self.model,
+            "tool": self.tool,
+            "approval": self.approval,
+            "result": self.result,
+            "lineage": self.lineage,
+            "redacted": self.redacted,
+        }
+
+
+def build_audit_trace(
+    events: List[Dict[str, Any]],
+    task_facts: Dict[str, TaskAuditFacts],
+    approvals: Dict[str, Dict[str, Any]],
+    *,
+    lineage: Dict[str, Optional[str]],
+) -> List[Dict[str, Any]]:
+    """Assemble an ordered, replayable audit trail.
+
+    `events`     : the persisted event envelopes, in stored (chronological) order.
+    `task_facts` : task_id -> TaskAuditFacts (who/model/tool/source), extracted by the caller.
+    `approvals`  : run.approvals (approval_id -> {decision, ...}); only id+decision are surfaced.
+    `lineage`    : the 4-id lineage (manager_deployment_id/deployment_id/swarm_id/correlation_id).
+
+    Returns a list of audit-record dicts (frozen schema), one per event, in order.
+    """
+    records: List[Dict[str, Any]] = []
+    for idx, ev in enumerate(events):
+        event_type = ev.get("event_type") or "unknown"
+        task_id = ev.get("task_id")
+        facts = task_facts.get(task_id) if task_id else None
+        payload = ev.get("payload") or {}
+
+        # who: prefer the event's instance id, fall back to the task's assigned agent / role.
+        actor = {
+            "agent_role": (facts.agent_role if facts else None) or payload.get("agent_role"),
+            "agent_instance_id": ev.get("agent_instance_id")
+            or (facts.assigned_agent_id if facts else None),
+        }
+
+        # which model / tool — only when the task reported them (no fabrication).
+        model = None
+        tool = None
+        if facts:
+            if facts.model_id:
+                model = {"model_id": facts.model_id}
+            if facts.tool_count:
+                tool = {"count": facts.tool_count}
+
+        # which approval — for approval.* events, surface id + decision (never the secret_ref).
+        approval = None
+        approval_id = payload.get("approval_id")
+        if approval_id:
+            decision = (approvals.get(approval_id) or {}).get("decision") or payload.get("decision")
+            approval = {"approval_id": approval_id, "decision": decision}
+
+        records.append(
+            AuditRecord(
+                audit_id=f"aud_{idx + 1:06d}",
+                occurred_at=ev.get("occurred_at"),
+                action=event_type,
+                actor=actor,
+                task_id=task_id,
+                model=model,
+                tool=tool,
+                approval=approval,
+                result=classify_result(event_type),
+                lineage=dict(lineage),
+            ).to_dict()
+        )
+    return records
+
+
+def replay(records: List[Dict[str, Any]]) -> List[str]:
+    """Render the ordered trail as human-readable step lines (for replay / diagnostics).
+
+    Pure formatting over the assembled records — no re-derivation, so the replay cannot diverge
+    from the audited records.
+    """
+    lines: List[str] = []
+    for r in records:
+        who = r["actor"].get("agent_instance_id") or r["actor"].get("agent_role") or "system"
+        model = (r.get("model") or {}).get("model_id")
+        appr = r.get("approval") or {}
+        bits = [r["audit_id"], r.get("occurred_at") or "-", who, r["action"], f"->{r['result']}"]
+        if r.get("task_id"):
+            bits.append(f"task={r['task_id']}")
+        if model:
+            bits.append(f"model={model}")
+        if appr.get("approval_id"):
+            bits.append(f"approval={appr['approval_id']}:{appr.get('decision')}")
+        lines.append(" ".join(str(b) for b in bits))
+    return lines

orchestrator/main.py

@@ -32,6 +32,7 @@ from .dispatch_score import (
     normalize_capability_match, normalize_tau, normalize_load,
 )
 from . import convergence as convergence_mod
+from . import audit as audit_mod
 from . import autonomous_tasks as autonomous_mod
 from . import task_competition as competition_mod
 from . import cross_review as cross_review_mod
@@ -2010,6 +2011,50 @@ async def get_swarm_diagnostics(deployment_id: str, request: Request):
     return {"success": True, "data": await build_runtime_diagnostics(run)}
 
 
+async def build_audit_trace_for_run(run) -> Dict[str, Any]:
+    """Assemble the replayable audit/lineage trail for a run (issue #17).
+
+    Pulls the persisted event stream, per-task model/tool facts (from each task's reported
+    `usage`), and approvals, and hands them to the pure `audit.build_audit_trace`. Returns the
+    ordered records + a human-readable replay. Carries no prompt content or secrets.
+    """
+    events = (await swarm_runtime.list_events(run.swarm_id, limit=500)).get("events", [])
+    task_facts: Dict[str, audit_mod.TaskAuditFacts] = {}
+    for tid in run.task_ids:
+        task = await task_queue.get_task(tid)
+        if not task:
+            continue
+        usage = (parse_task_result(task) or {}).get("usage") or {}
+        task_facts[tid] = audit_mod.TaskAuditFacts(
+            agent_role=task.agent_role,
+            assigned_agent_id=getattr(task, "assigned_agent_id", None),
+            model_id=usage.get("model_id") or (task.context or {}).get("model_id"),
+            tool_count=infer_tool_count(parse_task_result(task)),
+            source=task.source,
+            parent_task_id=task.parent_task_id,
+            root_task_id=task.root_task_id,
+        )
+    lineage = {
+        "manager_deployment_id": run.manager_deployment_id,
+        "deployment_id": run.deployment_id,
+        "swarm_id": run.swarm_id,
+        "correlation_id": run.correlation_id,
+    }
+    records = audit_mod.build_audit_trace(events, task_facts, run.approvals or {}, lineage=lineage)
+    return {"records": records, "replay": audit_mod.replay(records), "count": len(records)}
+
+
+@app.get("/api/agent/swarm/deployments/{deployment_id}/audit")
+@app.get("/api/swarms/{deployment_id}/audit")
+@app.get("/api/agnet/deployments/{deployment_id}/audit")
+async def get_swarm_audit(deployment_id: str, request: Request):
+    """Return the replayable audit/lineage trace for a swarm deployment (issue #17)."""
+    run, auth_error = await get_runtime_run_or_404(request, deployment_id)
+    if auth_error:
+        return auth_error
+    return {"success": True, "data": await build_audit_trace_for_run(run)}
+
+
 @app.post("/api/agent/swarm/deployments/{deployment_id}/stop")
 @app.post("/api/swarms/{deployment_id}/stop")
 @app.post("/api/agnet/deployments/{deployment_id}/stop")

scripts/test-audit-trace.py

@@ -0,0 +1,137 @@
+"""Audit / lineage trace tests (issue #17).
+
+Covers the pure assembler (orchestrator/audit.py) and the orchestrator wiring
+(build_audit_trace_for_run) that reconstructs a replayable per-step trail —
+who / when / which model / which tool / which approval / result — from the
+event stream + task lineage + approvals, carrying NO prompt content or secrets.
+
+Run from agent_swarm_v6 (install deps first — needs fakeredis):
+    pip install -r orchestrator/requirements.txt
+    REDIS_FAKE=1 python scripts/test-audit-trace.py
+"""
+import asyncio
+import json
+import os
+import sys
+from pathlib import Path
+
+os.environ["REDIS_FAKE"] = "1"
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from orchestrator import audit as audit_mod
+from orchestrator.audit import TaskAuditFacts, build_audit_trace, replay, classify_result
+
+failures = []
+
+
+def check(name, cond):
+    print(("PASS" if cond else "FAIL"), "-", name)
+    if not cond:
+        failures.append(name)
+
+
+def test_pure():
+    events = [
+        {"event_type": "task.created", "task_id": "t1", "occurred_at": "T0", "payload": {"title": "x"}},
+        {"event_type": "task.claimed", "task_id": "t1", "agent_instance_id": "agent-7", "occurred_at": "T1", "payload": {}},
+        {"event_type": "task.completed", "task_id": "t1", "agent_instance_id": "agent-7", "occurred_at": "T2", "payload": {}},
+        {"event_type": "approval.requested", "occurred_at": "T3", "payload": {"approval_id": "ap1", "operation": "git.push", "risk_level": "high"}},
+        {"event_type": "approval.rejected", "occurred_at": "T4", "payload": {"approval_id": "ap1"}},
+        {"event_type": "swarm.failed", "occurred_at": "T5", "payload": {"status": "failed"}},
+    ]
+    task_facts = {
+        "t1": TaskAuditFacts(agent_role="impl", assigned_agent_id="agent-7",
+                             model_id="gpt-x", tool_count=3, source="agent_proposed"),
+    }
+    approvals = {"ap1": {"decision": "rejected", "reason": "blocked by policy"}}
+    lineage = {"manager_deployment_id": "m1", "deployment_id": "d1", "swarm_id": "s1", "correlation_id": "c1"}
+
+    recs = build_audit_trace(events, task_facts, approvals, lineage=lineage)
+    check("one record per event, in order", len(recs) == len(events))
+    check("audit_id is deterministic/sequential", [r["audit_id"] for r in recs][:2] == ["aud_000001", "aud_000002"])
+    check("every record redacted + carries full lineage",
+          all(r["redacted"] is True and r["lineage"]["swarm_id"] == "s1" for r in recs))
+
+    # who
+    claimed = recs[1]
+    check("who: actor resolves agent_instance_id", claimed["actor"]["agent_instance_id"] == "agent-7")
+    check("who: actor resolves agent_role from task facts", claimed["actor"]["agent_role"] == "impl")
+    # which model / tool
+    check("which model: surfaced from task usage", claimed["model"] == {"model_id": "gpt-x"})
+    check("which tool: count surfaced from task usage", claimed["tool"] == {"count": 3})
+    # which approval
+    appr_evt = recs[4]
+    check("which approval: id + decision surfaced", appr_evt["approval"] == {"approval_id": "ap1", "decision": "rejected"})
+    # result classification
+    check("result: completed -> success", recs[2]["result"] == "success")
+    check("result: approval.rejected -> blocked", appr_evt["result"] == "blocked")
+    check("result: swarm.failed -> failed", recs[5]["result"] == "failed")
+    check("result: unknown -> info (no fabrication)", classify_result("some.weird.event") == "info")
+
+    # no content / no secrets anywhere in the serialized trail
+    blob = json.dumps(recs)
+    check("no secret_ref / credential leaked into audit records",
+          "secret_ref" not in blob and "azkv://" not in blob and "credential" not in blob)
+
+    # replay is ordered and human-readable
+    lines = replay(recs)
+    check("replay yields one line per record, ordered", len(lines) == len(recs) and lines[0].startswith("aud_000001"))
+    check("replay surfaces who/model/approval", "agent-7" in lines[1] and "model=gpt-x" in lines[1]
+          and "approval=ap1:rejected" in lines[4])
+
+
+async def test_integration():
+    from orchestrator.redis_client import redis_client
+    from orchestrator.swarm_runtime import swarm_runtime
+    from orchestrator.task_queue import task_queue
+    from orchestrator import main as orch
+
+    await redis_client.connect()
+    body = {"mode": "swarm", "orchestration_plan": {"objective": "audit it"},
+            "callback": {"url": "", "subscribed_events": []},
+            "metadata": {"manager_deployment_id": "m-audit"}}
+    run, _ = await swarm_runtime.get_or_create_run(body=body, idempotency_key=None, correlation_id="c-audit")
+
+    task = await task_queue.create_task(
+        task_id=f"{run.swarm_id}-t", description="impl", agent_role="impl",
+        required_capabilities=["python"], enqueue=True,
+    )
+    # Agent-reported result carrying model usage + tool calls + a would-be secret (must NOT leak).
+    await task_queue.complete_task(task.task_id, result=json.dumps({
+        "summary": "done",
+        "usage": {"model_id": "gpt-x", "model_tokens": 10},
+        "tool_calls": [{"name": "write_file"}, {"name": "run_git"}],
+        "secret_ref": "azkv://vault/secrets/x",
+    }))
+    await swarm_runtime.attach_task(run, task.task_id)
+    run.approvals["ap1"] = {"approval_id": "ap1", "decision": "approved"}
+    await swarm_runtime.save_run(run)
+
+    await swarm_runtime.emit_event(run, "task.claimed", task_id=task.task_id,
+                                   agent_instance_id="agent-1", payload={"agent_role": "impl"})
+    await swarm_runtime.emit_event(run, "task.completed", task_id=task.task_id, agent_instance_id="agent-1")
+
+    data = await orch.build_audit_trace_for_run(run)
+    recs = data["records"]
+    check("integration: trace assembled from real run", data["count"] >= 1 and bool(recs))
+    by_type = {r["action"]: r for r in recs}
+    check("integration: model_id reconstructed from task usage",
+          by_type.get("task.completed", {}).get("model") == {"model_id": "gpt-x"})
+    check("integration: tool count reconstructed (2 tool_calls)",
+          by_type.get("task.completed", {}).get("tool") == {"count": 2})
+    check("integration: no secret leaked into audit trace",
+          "secret_ref" not in json.dumps(data) and "azkv://" not in json.dumps(data))
+
+
+async def main():
+    test_pure()
+    await test_integration()
+    print()
+    if failures:
+        print(f"{len(failures)} audit-trace check(s) FAILED: {failures}")
+        sys.exit(1)
+    print("all audit-trace checks passed")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())