diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 057dc99..4357612 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -70,6 +70,10 @@ jobs: env: { REDIS_FAKE: "1" } run: python scripts/test-usage-cost-phase.py + - name: Review/rework client timeline events (#34) + env: { REDIS_FAKE: "1", MAX_REVIEW_CYCLES: "2" } + run: python scripts/test-review-timeline-events.py + - name: Audit / lineage trace (replayable) (#17) env: { REDIS_FAKE: "1" } run: python scripts/test-audit-trace.py diff --git a/docs/integration/event-schema.md b/docs/integration/event-schema.md index 0e0403f..7b7ae87 100644 --- a/docs/integration/event-schema.md +++ b/docs/integration/event-schema.md @@ -4,7 +4,7 @@ > > 冻结要点（实现见 `orchestrator/swarm_runtime.py`，契约测试 `scripts/test-contract-freeze.py`）： > 1. envelope 带 **per-swarm 严格递增 `sequence`**（从 1、无空洞，供客户端 `events?after=` 去重/续传，§2）。 -> 2. **客户端 13 类事件**（§4 标 ⭐）冻结为 `FROZEN_CLIENT_EVENT_TYPES`（`swarm_runtime.py`）。 +> 2. **客户端事件集**（§4 标 ⭐）冻结为 `FROZEN_CLIENT_EVENT_TYPES`（`swarm_runtime.py`）：#28 的 13 类核心 + #34 新增 4 类评审/返工（`review.started`/`review.decision_made`/`rework.requested`/`rework.completed`，**脱敏投影**，见 §4）= 17。 > 3. **artifact** 扁平字段 `{uri, checksum, task_id, size_bytes?, created_at}`（§3）。 > 4. **凭据红线**：`secret_ref`/`credential_ref`/`signing_secret_ref` 为 `azkv://` 引用（非明文），Swarm 侧 `_redact_sensitive` 透传引用、剔除明文 `*_token/_secret/_key/password`；客户端可见视图由 HM 再做递归脱敏。 > @@ -64,7 +64,7 @@ ## 4. 事件类型注册表（与 HM 一致） -⭐ = **冻结的客户端 13 类**（`FROZEN_CLIENT_EVENT_TYPES`，驾驶舱据此渲染；HM `agent_callback.go` 据此对齐注册表）。其余为 Swarm 也发的辅助/运行时事件（HM 仍消费，但非客户端必需）。 +⭐ = **冻结的客户端事件**（`FROZEN_CLIENT_EVENT_TYPES`，17 类 = #28 的 13 核心 + #34 的 4 评审/返工，驾驶舱据此渲染；HM `agent_callback.go` 据此对齐注册表）。`review.*`/`rework.*` 为**脱敏客户端投影**（仅 reviewer id/verdict/failed_criteria/affected_tasks/recommended_rework/root_cause + 安全标量；**不含** evidence/summary/rework_reason 自由文本）。其余为 Swarm 也发的辅助/运行时事件（HM 仍消费，但非客户端必需）。 | event_type | payload 必填字段（HM 校验） | 分类 | Swarm 是否发出 | |---|---|---|---| @@ -81,6 +81,10 @@ | ⭐ `swarm.completed` | `status` | swarm_terminal | ✅（终态，与 `deployment.status_changed` 并发） | | ⭐ `swarm.failed` | `status` | swarm_terminal | ✅（终态） | | ⭐ `swarm.stopped` | `status` | swarm_terminal | ✅（`stop_run`） | +| ⭐ `review.started` | `cycle`, `reviewer_count` | review | ✅（#34，脱敏投影；交叉评审一轮开始） | +| ⭐ `review.decision_made` | `accepted`, `rework_targets` | review | ✅（#34，脱敏投影；仲裁结论 + 逐评审者 verdict/criteria，无 evidence/summary） | +| ⭐ `rework.requested` | `task_id`, `root_cause` | review | ✅（#34，脱敏投影；无 rework_reason/evidence） | +| ⭐ `rework.completed` | `task_id`, `root_cause` | review | ✅（#34，脱敏投影；返工被重新接受） | | `deployment.status_changed` | `status` | deployment | ✅（HM 用于 AgentDeployment.Status；客户端用 swarm.* 终态） | | `task.heartbeat` | `task_id`, `agent_role` | swarm_task_flow | ✅ | | `task.blocked` | `task_id`, `reason` | swarm_task_flow | ✅ | diff --git a/docs/integration/frontend-event-api.md b/docs/integration/frontend-event-api.md index 66793eb..f0b2adf 100644 --- a/docs/integration/frontend-event-api.md +++ b/docs/integration/frontend-event-api.md @@ -1,6 +1,8 @@ # 前端事件 API（Frontend Event API） -> 状态：**DRAFT / 待对齐 Frontend & Product Team**。 +> 状态：**DRAFT / 待对齐 Frontend & Product Team**；**评审/返工时间线部分已推进（#34）**——`review.started`/`review.decision_made`/`rework.requested`/`rework.completed` 已定义为客户端可见事件（脱敏投影）并纳入冻结集（见 §4 / event-schema §4）。其余视图（graph/cost/SSE 传输）仍待 Frontend & Product 冻结。 +> +> **剩余跨仓项（#34 DoD）**：HM `agent_callback.go` 登记这 4 类；macOS/winos cockpit 渲染评审结论/证据摘要/返工目标/循环次数时间线（客户端仓子工单）；渲染只用脱敏摘要，不展示原文（对齐 security-boundary）。 > > 依据：`heicode-mananger/docs/integration/heicode-desktop-client-api.md`、`docs/integration/manager-side-contract-patches.md §6`。配套：[`event-schema.md`](./event-schema.md)、[`runtime-contract.md`](./runtime-contract.md)。 @@ -43,7 +45,7 @@ | UI 视图 | 数据来源（现有） | 缺口 | |---|---|---| | Task DAG / topology | `/tasks`（`depends_on`）+ `/workflow.phases` | 稳定 graph schema、增量更新事件 | -| Review loop / retry timeline | `task.retried` + `timeline.updated`（`Review cycle N`） | 独立 review/retry 事件类型与时间线 schema | +| Review loop / retry timeline | **`review.started` / `review.decision_made` / `rework.requested` / `rework.completed`**（#34，脱敏客户端投影，见 event-schema §4）+ `task.retried` | ✅ 事件类型已定义并纳入冻结集（17 类）；HM 注册 + cockpit 渲染（评审结论/证据摘要/返工目标/循环次数）落客户端仓子工单 | | Collaboration timeline | `handoff.requested/completed` + peer 路由 | peer 消息事件未对前端暴露 | | Token / cost monitor | `/metrics` + `budget.alert`（usage） | 统一 usage 推送（见 usage-billing §7） | | 实时推送 | `/logs` 轮询 | 统一 SSE/WS event stream（事件名、断线重连、刷新可恢复） | @@ -54,7 +56,7 @@ - 🔴 Swarm 未提供前端 SSE/WS 推送（仅 REST 轮询）。 - 🔴 蜂群 graph/timeline/cost 的统一前端 schema 未定义。 -- 🟡 review/retry/collaboration 缺独立事件类型（当前借 `task.retried`/`timeline.updated`）。 +- 🟢 review/rework 已有独立客户端事件（#34：`review.started`/`review.decision_made`/`rework.requested`/`rework.completed`，脱敏投影，已纳入冻结集）；collaboration 仍借 `handoff.*`/peer 路由（未单列）。 - 待对齐：Frontend / Product Team 确认蜂群三视图（graph / timeline / cost）入口与事件流形态；与 HM 的 `/api/user/tasks/{id}/events` 通道如何衔接。 ## 6. Swarm 侧冻结口径（#18：run/task/event API + Agent Registry/调度） diff --git a/docs/swarm/review-loop-protocol.md b/docs/swarm/review-loop-protocol.md index 78233cc..6b1f656 100644 --- a/docs/swarm/review-loop-protocol.md +++ b/docs/swarm/review-loop-protocol.md @@ -79,7 +79,7 @@ ### 3.2 状态与事件 - **状态**：`run.metadata["cross_review"]` 存 `AggregatedVerdict.to_dict()`，`run.metadata["rework_attributions"]` 存 `[ReworkAttribution.to_dict()]`，与 `review_cycles` 并存（`MAX_REVIEW_CYCLES` 预算复用）；拒绝则 `reopen_task` 返工目标、run 退回 `running`。 -- **事件不进 Manager 流**：`review.*` / `rework.*` 的 payload builder 已实现但**不经 `emit_event` 外发**（未在 Manager `agent_callback.go` 注册；与 `swarm.health`/`convergence.*` 同策略，避免向订阅全部的回调投递未登记事件）。重开通过既有 `timeline.updated` 反映。登记后方可启用 Manager 侧发送。 +- **事件已对客户端可见（#34）**：`review.started` / `review.decision_made` / `rework.requested` / `rework.completed` 现以**脱敏客户端投影**（`cross_review.*_client_payload`：仅 reviewer id / verdict / failed_criteria / affected_tasks / recommended_rework / root_cause + 安全标量；**丢弃** evidence / summary / rework_reason 自由文本）经 `emit_event` 由 `run_cross_review` 外发，并纳入 `FROZEN_CLIENT_EVENT_TYPES`（13→17）。HM `agent_callback.go` 需登记这 4 类，cockpit 据此渲染评审/返工时间线。内部仍另存 `run.metadata["cross_review"]` / `["rework_attributions"]`（含完整 evidence，供审计，不外发）。测试 `scripts/test-review-timeline-events.py`。 ### 3.4 不变量 diff --git a/orchestrator/cross_review.py b/orchestrator/cross_review.py index 72a1b8a..6fabc22 100644 --- a/orchestrator/cross_review.py +++ b/orchestrator/cross_review.py @@ -446,3 +446,88 @@ def rework_completed_payload( "summary": f"Rework {'completed' if succeeded else 'failed'} for " f"{attribution.target_task_id}", } + + +# ── Client-visible REDACTED projections (issue #34) ─────────────────────────── +# The payload builders above are the rich INTERNAL telemetry shape. The desktop cockpit +# review/rework timeline must show only non-content metadata, so the client-facing events carry a +# strict allowlist: reviewer identity/verdict/failed_criteria/affected_tasks/recommended_rework/ +# root_cause (+ safe scalars: cycle, counts, accepted, disagreement, method, status). They DROP all +# free text — `evidence`, `rework_reason`, and human-readable `summary` — to honor +# security-boundary (no prompt/code/content) and the #34 DoD. These four are the event types added +# to FROZEN_CLIENT_EVENT_TYPES; the emitter (`main.run_cross_review`) sends ONLY these projections. + +# The four review/rework event types promoted to the client-visible set (#34). +CLIENT_REVIEW_EVENT_TYPES = ( + "review.started", + "review.decision_made", + "rework.requested", + "rework.completed", +) + + +def review_started_client_payload(run_id: str, reviewer_agent_ids: List[str], *, cycle: int = 0) -> Dict[str, Any]: + """Redacted `review.started` (a cross-review round begins). Identity + count only.""" + return { + "swarm_id": run_id, + "phase": "Review", + "review_kind": "cross_review", + "cycle": cycle, + "reviewer_agent_ids": list(reviewer_agent_ids), + "reviewer_count": len(reviewer_agent_ids), + } + + +def review_decision_client_payload(run_id: str, verdict: "AggregatedVerdict", *, cycle: int = 0) -> Dict[str, Any]: + """Redacted `review.decision_made`: arbitration outcome + per-reviewer verdict/criteria only. + + Drops each ReviewDecision's free-text `evidence`/`summary` and the verdict `summary`. + """ + return { + "swarm_id": run_id, + "phase": "Review", + "review_kind": "cross_review", + "cycle": cycle, + "accepted": verdict.accepted, + "arbitration_method": verdict.method, + "disagreement": verdict.disagreement, + "pass_votes": verdict.pass_votes, + "fail_votes": verdict.fail_votes, + "rework_targets": list(verdict.rework_targets), + "reviewers": [ + { + "reviewer_agent_id": d.reviewer_agent_id, + "verdict": d.verdict, + "failed_criteria": list(d.failed_criteria), + "affected_tasks": list(d.affected_tasks), + "recommended_rework": list(d.recommended_rework), + } + for d in verdict.decisions + ], + } + + +def rework_requested_client_payload(run_id: str, attribution: ReworkAttribution, *, cycle: int = 0) -> Dict[str, Any]: + """Redacted `rework.requested`: target + root_cause + attribution ids only (no reason/evidence).""" + return { + "swarm_id": run_id, + "phase": "Review", + "cycle": cycle, + "task_id": attribution.target_task_id, + "root_cause": attribution.root_cause.value, + "source_task_id": attribution.source_task_id, + "introduced_by_agent_id": attribution.introduced_by_agent_id, + "detected_by_agent_id": attribution.detected_by_agent_id, + } + + +def rework_completed_client_payload(run_id: str, *, task_id: str, root_cause: str, cycle: int = 0) -> Dict[str, Any]: + """Redacted `rework.completed`: which task's rework converged + its root_cause.""" + return { + "swarm_id": run_id, + "phase": "Review", + "cycle": cycle, + "task_id": task_id, + "root_cause": root_cause, + "status": "completed", + } diff --git a/orchestrator/main.py b/orchestrator/main.py index 57b0ce9..926e22e 100644 --- a/orchestrator/main.py +++ b/orchestrator/main.py @@ -962,9 +962,25 @@ async def run_cross_review(run, tasks) -> bool: verdict = cross_review_mod.aggregate_reviews(decisions) run.metadata["cross_review"] = verdict.to_dict() + # Client-visible review timeline (#34): emit REDACTED projections (allowlist only — no + # evidence/summary/reason free text; see cross_review.*_client_payload). HM registers these 4 + # types; the cockpit renders the review/rework timeline from them. + reviewer_ids = [d.reviewer_agent_id for d in decisions] + await swarm_runtime.emit_event(run, "review.started", + payload=cross_review_mod.review_started_client_payload(run.swarm_id, reviewer_ids, cycle=cycles)) + await swarm_runtime.emit_event(run, "review.decision_made", + payload=cross_review_mod.review_decision_client_payload(run.swarm_id, verdict, cycle=cycles)) + task_owner = {t.task_id: t.assigned_agent_id for t in tasks if t.assigned_agent_id} valid_targets = [tid for tid in verdict.rework_targets if tid in run.task_ids] if verdict.accepted or not valid_targets: + # A prior cycle's rework was redone and is now accepted → mark those targets complete. + if cycles > 0: + for a in (run.metadata.get("rework_attributions") or []): + await swarm_runtime.emit_event(run, "rework.completed", + payload=cross_review_mod.rework_completed_client_payload( + run.swarm_id, task_id=a.get("target_task_id"), + root_cause=a.get("root_cause"), cycle=cycles)) run.metadata["reviews"] = [] # consumed await swarm_runtime.save_run(run) return False @@ -986,6 +1002,11 @@ async def run_cross_review(run, tasks) -> bool: "status": "running", "retry_tasks": reopened, "disagreement": verdict.disagreement, }) + # Client-visible rework timeline (#34): one redacted rework.requested per reopened target. + for a in attributions: + if a.target_task_id in reopened: + await swarm_runtime.emit_event(run, "rework.requested", + payload=cross_review_mod.rework_requested_client_payload(run.swarm_id, a, cycle=cycles + 1)) logger.info(f"Cross-review reopened {reopened} on run {run.swarm_id} (cycle {cycles + 1})") return True diff --git a/orchestrator/swarm_runtime.py b/orchestrator/swarm_runtime.py index 689d8e1..f183e36 100644 --- a/orchestrator/swarm_runtime.py +++ b/orchestrator/swarm_runtime.py @@ -98,6 +98,13 @@ FROZEN_CLIENT_EVENT_TYPES = ( "swarm.completed", "swarm.failed", "swarm.stopped", + # Review/rework timeline (agent_swarm#34) — emitted as REDACTED client projections from + # run_cross_review (cross_review.*_client_payload): reviewer id/verdict/criteria/targets/ + # root_cause only, no evidence/summary/reason free text. + "review.started", + "review.decision_made", + "rework.requested", + "rework.completed", ) diff --git a/scripts/test-contract-freeze.py b/scripts/test-contract-freeze.py index 077cfbb..23035bd 100644 --- a/scripts/test-contract-freeze.py +++ b/scripts/test-contract-freeze.py @@ -68,13 +68,15 @@ async def test_sequence_monotonic(): async def test_frozen_event_types_roundtrip(): - check("frozen set has exactly 13 client event types", len(FROZEN_CLIENT_EVENT_TYPES) == 13) + # The #28 freeze defined these 13 core client types; #34 later added 4 review/rework types + # (review.started/decision_made, rework.requested/completed). Assert the 13 core are present + # (subset), not an exact count, so adding client events doesn't break this contract test. expected = { "task.created", "task.claimed", "task.running", "task.completed", "task.failed", "handoff.created", "approval.requested", "approval.approved", "approval.rejected", "artifact.created", "swarm.completed", "swarm.failed", "swarm.stopped", } - check("frozen set matches the agreed 13", set(FROZEN_CLIENT_EVENT_TYPES) == expected) + check("the 13 #28-core client types are all in the frozen set", expected.issubset(set(FROZEN_CLIENT_EVENT_TYPES))) run = await new_run() for et in FROZEN_CLIENT_EVENT_TYPES: diff --git a/scripts/test-review-timeline-events.py b/scripts/test-review-timeline-events.py new file mode 100644 index 0000000..7bef1b2 --- /dev/null +++ b/scripts/test-review-timeline-events.py @@ -0,0 +1,131 @@ +"""Review/rework client-timeline event tests (issue #34). + +The runtime cross-review (#11) was internal-only. #34 promotes a REDACTED client projection of +review.started / review.decision_made / rework.requested / rework.completed to the frozen +client-visible set so the cockpit can render the review/rework timeline. + +Verifies: + * the 4 types are in FROZEN_CLIENT_EVENT_TYPES; + * run_cross_review emits review.started + review.decision_made + rework.requested on a failing + cross-review round (real site); + * emitted payloads carry ONLY the allowlist — NO evidence / summary / rework_reason free text + (security-boundary / #34 redaction); + * the client-projection builders drop content fields. + +Hermetic: REDIS_FAKE, no model key, no callback url. + +Run from agent_swarm_v6 (install deps first): + pip install -r orchestrator/requirements.txt + REDIS_FAKE=1 MAX_REVIEW_CYCLES=2 python scripts/test-review-timeline-events.py +""" +import asyncio +import json +import os +import sys +from pathlib import Path + +os.environ["REDIS_FAKE"] = "1" +os.environ["MAX_REVIEW_CYCLES"] = "2" +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from orchestrator import cross_review as cr +from orchestrator.swarm_runtime import FROZEN_CLIENT_EVENT_TYPES + +failures = [] + +SECRET_EVIDENCE = "SECRET-EVIDENCE-prompt-and-code-content" +SECRET_SUMMARY = "SECRET-SUMMARY-free-text" + + +def check(name, cond): + print(("PASS" if cond else "FAIL"), "-", name) + if not cond: + failures.append(name) + + +def test_frozen_set_and_builders(): + for et in cr.CLIENT_REVIEW_EVENT_TYPES: + check(f"{et} in frozen client set", et in FROZEN_CLIENT_EVENT_TYPES) + check("frozen set now has 17 client types", len(FROZEN_CLIENT_EVENT_TYPES) == 17) + + # decision projection drops evidence/summary/confidence/weight + decisions = [ + cr.ReviewDecision(verdict="pass", reviewer_agent_id="r1", evidence=[SECRET_EVIDENCE], summary=SECRET_SUMMARY), + cr.ReviewDecision(verdict="fail", reviewer_agent_id="r2", failed_criteria=["tests missing"], + recommended_rework=["t1"], evidence=[SECRET_EVIDENCE], summary=SECRET_SUMMARY), + ] + verdict = cr.aggregate_reviews(decisions) + p = cr.review_decision_client_payload("s1", verdict, cycle=0) + blob = json.dumps(p) + check("decision projection has no evidence/summary content", SECRET_EVIDENCE not in blob and SECRET_SUMMARY not in blob) + check("decision projection reviewers expose only allowlist keys", + all(set(r.keys()) == {"reviewer_agent_id", "verdict", "failed_criteria", "affected_tasks", "recommended_rework"} + for r in p["reviewers"])) + + # rework.requested projection drops rework_reason/evidence + attr = cr.ReworkAttribution(target_task_id="t1", rework_reason=SECRET_SUMMARY, + root_cause=cr.ReworkCategory.TEST, evidence=[SECRET_EVIDENCE]) + rp = cr.rework_requested_client_payload("s1", attr, cycle=1) + rblob = json.dumps(rp) + check("rework projection has no reason/evidence content", SECRET_SUMMARY not in rblob and SECRET_EVIDENCE not in rblob) + check("rework projection exposes root_cause", rp.get("root_cause") == "test" and rp.get("task_id") == "t1") + + +async def test_emitted_from_run_cross_review(): + 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": "review me"}, + "callback": {"url": "", "subscribed_events": []}, + "metadata": {"manager_deployment_id": "m-rev"}} + run, _ = await swarm_runtime.get_or_create_run(body=body, idempotency_key=None, correlation_id="c-rev") + t1 = await task_queue.create_task(task_id=f"{run.swarm_id}-t1", description="impl", + agent_role="impl", required_capabilities=["python"], enqueue=True) + await swarm_runtime.attach_task(run, t1.task_id) + # Must be terminal for reopen_task to send it back for rework. + await task_queue.complete_task(t1.task_id, result=json.dumps({"summary": "done"})) + + # Two reviews: 1 pass / 1 fail (tie -> safe-reject) with the fail recommending rework on t1. + run.metadata["reviews"] = [ + cr.ReviewDecision(verdict="pass", reviewer_agent_id="rev-1", + evidence=[SECRET_EVIDENCE], summary=SECRET_SUMMARY).to_dict(), + cr.ReviewDecision(verdict="fail", reviewer_agent_id="rev-2", failed_criteria=["missing tests"], + recommended_rework=[t1.task_id], evidence=[SECRET_EVIDENCE], summary=SECRET_SUMMARY).to_dict(), + ] + await swarm_runtime.save_run(run) + + reopened = await orch.run_cross_review(run, [t1]) + check("run_cross_review reopened (failing round)", reopened is True) + + raw = await redis_client.lrange(f"{swarm_runtime.EVENT_KEY_PREFIX}{run.swarm_id}", 0, -1) + events = [json.loads(r) for r in raw] + types = [e["event_type"] for e in events] + check("review.started emitted", "review.started" in types) + check("review.decision_made emitted", "review.decision_made" in types) + check("rework.requested emitted", "rework.requested" in types) + + full = json.dumps(events) + check("NO evidence/summary content leaked in any emitted event", + SECRET_EVIDENCE not in full and SECRET_SUMMARY not in full) + check("no 'evidence'/'rework_reason' keys in emitted review/rework payloads", + all("evidence" not in (e.get("payload") or {}) and "rework_reason" not in (e.get("payload") or {}) + for e in events if e["event_type"].startswith(("review.", "rework.")))) + dm = [e for e in events if e["event_type"] == "review.decision_made"][0] + check("review.decision_made carries rework_targets + reviewers", "rework_targets" in dm["payload"] and "reviewers" in dm["payload"]) + + +async def main(): + test_frozen_set_and_builders() + await test_emitted_from_run_cross_review() + print() + if failures: + print(f"{len(failures)} review-timeline check(s) FAILED: {failures}") + sys.exit(1) + print("all review-timeline event checks passed") + + +if __name__ == "__main__": + asyncio.run(main())