From dd96b73b2dd50f350b8f1146625d19494f506cee Mon Sep 17 00:00:00 2001 From: Songhaoz666 Date: Thu, 11 Jun 2026 16:28:37 +0800 Subject: [PATCH] =?UTF-8?q?Swarm=20I/O=EF=BC=9A=E6=8E=A5=E6=94=B6=E7=94=A8?= =?UTF-8?q?=E6=88=B7=20prompt=EF=BC=88=E8=BF=BD=E5=8A=A0=E8=BE=93=E5=85=A5?= =?UTF-8?q?=EF=BC=89+=20=E8=BF=94=E5=9B=9E=E7=BB=93=E6=9E=9C=EF=BC=88/resu?= =?UTF-8?q?lt=20+=20swarm.completed=20=E5=B8=A6=E7=AD=94=E6=A1=88=EF=BC=89?= =?UTF-8?q?=EF=BC=88Refs=20#40=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 补齐「客户端如何把 prompt 给我们 + 如何拿到结果」的端到端路径。 输入： - POST /api/swarms/{id}/input（+别名）：接收用户后续 prompt，注入 source=user_append 任务进共享池 （终态 run 自动 reopen 为 running；stopped 拒绝 RUN_STOPPED）。指令原文作任务描述下发给 agent， **不回显进事件流**——仅产一条 task.created(user_append) 类别 message。（初始 prompt 仍走 create 的 requirement.objective） 输出： - GET /api/swarms/{id}/result（+别名）：返回 {summary, deliverable, artifacts[], termination_reason, status}（build_run_result）。产物内容在 artifact.uri（git/runtime），result 给摘要+定位。 - swarm.completed 事件 payload 增带 summary + deliverable，客户端看一条终态事件即得答案。 文档：runtime-contract §3 增 input/result 行；event-schema swarm.completed 标注带 summary/deliverable； CLIENT_GUIDE §3.5/§3.6（input/result）+ §3.2 表 + 修正 §5「Agent 平台拉起」为「Swarm 拉起 + 从 secret_ref 解析 key」（对齐 team 决议）+ 新增 §9 完整工作流地图（client→HM→Swarm→output→client）。 测试 scripts/test-swarm-io.py（TestClient：input 注入 + 原文不入事件流 + 终态 reopen + stopped 拒绝 + result 形状）接入 CI。e2e/contract 回归通过。 影响范围：仅 agent_swarm（新增 2 只读/写端点 + 终态事件增字段 + 文档 + 测试 + CI）。 追加输入原文不入事件流/回调/日志；不改鉴权/计费账本/审批链。SSE 实时流仍归 HM Phase2（#46）。 Refs #40 Co-Authored-By: Claude Opus 4.8 (1M context) --- .github/workflows/ci.yml | 4 + CLIENT_GUIDE.md | 67 +++++++++++++++- docs/integration/event-schema.md | 2 +- docs/integration/runtime-contract.md | 2 + orchestrator/main.py | 115 +++++++++++++++++++++++++++ scripts/test-swarm-io.py | 98 +++++++++++++++++++++++ 6 files changed, 285 insertions(+), 3 deletions(-) create mode 100644 scripts/test-swarm-io.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0e2f034..b56fb38 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -63,6 +63,10 @@ jobs: env: { REDIS_FAKE: "1" } run: python scripts/test-contract-events.py + - name: Swarm I/O — receive prompt + return result (#40) + env: { REDIS_FAKE: "1" } + run: python scripts/test-swarm-io.py + - name: Audit / lineage trace (replayable) (#17) env: { REDIS_FAKE: "1" } run: python scripts/test-audit-trace.py diff --git a/CLIENT_GUIDE.md b/CLIENT_GUIDE.md index 203785e..deb157f 100644 --- a/CLIENT_GUIDE.md +++ b/CLIENT_GUIDE.md @@ -24,7 +24,7 @@ WebSocket `/ws/{agent_id}` 仅供 **Agent 执行单元** 接入，**不是客户 --- -## 3. 生命周期：创建 → 观察 → （审批）→ 停止 +## 3. 生命周期：创建 → 观察 →（追加输入）→（审批）→ 取结果 → 停止 ### 3.1 创建部署 `POST /api/swarms`（别名 `/api/agent/swarm/deployments`、`/api/agnet/deployments`） @@ -55,6 +55,8 @@ curl -s -X POST http://:8000/api/swarms \ | `GET /api/swarms/{deployment_id}/logs`（`/events`） | 事件流，分页 `?limit=&cursor=`（断线后用 cursor 回补） | | `GET /api/swarms/{deployment_id}/metrics` | 任务/Agent/预算指标 | | `GET /api/swarms/{deployment_id}/diagnostics` | 失败、回调尝试、审批诊断 | +| `GET /api/swarms/{deployment_id}/result` | **结果**：`{summary, deliverable, artifacts[], termination_reason, status}`（见 §3.6） | +| `GET /api/swarms/{deployment_id}/audit` | 可回放审计/链路（谁/何模型/何工具/何审批，脱敏） | 状态机：`waiting_approval → running → (blocked ⇄ running) → completed | failed | stopped`。 @@ -73,6 +75,22 @@ curl -s -X POST http://:8000/api/swarms//stop \ -d '{ "reason": "client requested stop" }' ``` +### 3.5 追加输入（用户后续 prompt，运行中/已完成均可续） +用户在驾驶舱继续补充要求时调用。指令被注入为 `source=user_append` 的新任务进入共享池，agent 自选执行；若 run 已 `completed`/`failed` 会**自动 reopen 为 running**（「继续」语义），`stopped` 的 run 会被拒绝（请另起新 run）。 +```bash +curl -s -X POST http://:8000/api/swarms//input \ + -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" \ + -d '{ "instruction": "再补一个 subtract(a,b) 并加测试" }' +``` +响应 `data`：`task_id`、`status`、`reopened`。**红线**：指令原文仅作任务描述下发给执行 agent，**绝不回显进事件流**——事件流只产一条 `task.created`（`source=user_append`，类别 message，无原文）。 + +### 3.6 取结果 +run 进入终态（推送 `swarm.completed`，其 payload 已带 `summary`+`deliverable`）后，或任意时刻拉取聚合结果： +```bash +curl -s http://:8000/api/swarms//result -H "Authorization: Bearer $TOKEN" +``` +响应 `data`：`summary`（综述）、`deliverable`（`{has_deliverable, commit_sha, files_modified, artifact_ids, …}`）、`artifacts[]`（`{uri, checksum, …}`）、`termination_reason`、`status`。**产物内容**（代码等）在各 artifact 的 `uri`（如 `git://repo#branch`）——`result` 给的是摘要 + 定位，内容按 uri 取。 + --- ## 4. 健康检查 @@ -88,7 +106,8 @@ curl -s -X POST http://:8000/api/swarms//stop \ - `MAX_REVIEW_CYCLES`（默认 2）：交叉评审/重做的轮次预算（保留）。 - `MAX_AGENTS_PER_USER`（默认 10）：每用户并发连接的 Agent 上限（见 security-boundary §6）。 - 仅剩的能力开关：`ENABLE_QUALITY_EVAL` + `HEICODE_SANDBOX_ISOLATED`（代码沙箱评分，见 security-boundary §8.1）、`ENABLE_SUBTASK_HANDOFF`（agent 侧子任务移交）。 -- 模型走 HM 模型网关（OpenAI 兼容 `/v1`）：`OPENAI_API_KEY` = **HM 为该用户现签的 per-user `sk-`**（NewAPI token），`OPENAI_API_BASE` = **HM `/v1`**，`OPENAI_MODEL` 为所选模型。由 Agent 平台在拉起 agent 时注入其进程 env（server→server，**不入 create 请求 / 回调 / 日志**）。模型调用**扣发起用户 `user.Quota`**、按 token 名归集计量；详见 [usage-billing-schema.md §2](docs/integration/usage-billing-schema.md)。 +- **Agent 拉起 + 限额由 Swarm 运行时负责**（team 决议，runtime-contract §3.3）：编排器在 create 播种后由 `agent_launcher` 拉起能力多样的 agent 池（拉起数 `= min(AGENT_LAUNCH_POOL_SIZE, MAX_AGENTS_PER_USER − 已连)`；后端 `AGENT_LAUNCH_BACKEND` = none/subprocess/command）。AM 不再拉蜂群 agent。 +- 模型走 HM 模型网关（OpenAI 兼容 `/v1`）：`OPENAI_API_KEY` = 该用户 per-user `sk-`（NewAPI token），`OPENAI_API_BASE` = HM `/v1`。**由 Swarm 从 `billing_context.secret_ref`(`azkv://`) 服务端解析**后注入被拉起 agent 的进程 env（**不入 create 请求 / 回调 / 日志 / argv**）。模型调用**扣发起用户 `user.Quota`**、按 token 名归集；HM 侧只需 mint `sk-` → 写 KV → 传 `secret_ref` → 吊销。详见 [usage-billing-schema.md §2](docs/integration/usage-billing-schema.md)。 --- @@ -123,3 +142,47 @@ python scripts/stub_agent.py - **桌面/前端 UI**：当前桌面客户端为单 Agent A2A 直连模型；**蜂群图 / 协作时间线**的前端事件 API 尚未冻结（见 [frontend-event-api.md](docs/integration/frontend-event-api.md)）。 - **不得在请求/回调/日志中出现明文密钥**；凭据一律 `secret_ref`（`azkv://`）。见 [security-boundary.md](docs/integration/security-boundary.md)。 - **计费主线**：Agent 模型调用走 HM `/v1`，用 **HM 为该用户现签的 per-user `sk-`** 作 `OPENAI_API_KEY`，**扣发起用户 `user.Quota`**；NewAPI 是账本、按请求计量，**Swarm 不是账本、仅上报用量**（随模型请求带 `X-Agent-*` 归因头）。归因主轴 = `user.id`/`channelId`，不引入 tenant。见 [usage-billing-schema.md §1/§2](docs/integration/usage-billing-schema.md)。 + +--- + +## 9. 完整工作流地图（client → swarm → output → client） + +端到端调用模型（生产链路：终端用户 → HM 控制面/查询面 → Swarm 执行面；Swarm 不在对话回路）： + +``` +[终端用户] 在桌面客户端提交需求 + │ 设备配对签名鉴权 + ▼ +[HM Heicode Manager] 控制面 + 查询面 + │ ① mint per-user sk- → 写 Azure Key Vault + │ ② POST /api/agent/swarm/deployments (service token) + │ body: requirement.objective(=用户 prompt) + callback.url + metadata.manager_deployment_id + │ + billing_context.secret_ref=azkv://.../ (key 仅引用，不传明文) + ▼ +[Swarm 编排器 orchestrator] 执行面（本仓） + │ ③ 播种单一种子任务（objective） —— 接收「初始 prompt」 + │ ④ agent_launcher 拉起 per-user agent 池（≤ MAX_AGENTS_PER_USER）； + │ 从 secret_ref 解析 sk- 注入 agent env（OPENAI_API_KEY/BASE/ORCHESTRATOR_URL/...） + ▼ +[专家 Agent ×N] 主动出站连 WS /ws/{agent_id}，register(user_id) → 受每用户限额 + │ ⑤ 自选种子 → 自主分解(#7) → 竞争/接管(#8) → 执行(调 HM /v1 模型，扣发起用户 Quota) + │ → 同伴交叉评审(#11) → 收敛(#12) → master_agent.synthesize 汇总 + │ 每步经编排器 emit_event → 带签名回调 POST 到 HM callback.url（task.*/handoff.created/ + │ approval.*/artifact.created/review.*/rework.*/swarm.*，每条带 per-swarm 严格递增 sequence） + ▼ +[HM 查询面/回调摄入] 持久化事件 + 透传 sequence + ▼ +[桌面客户端] ⑥ 经 HM 拉取/SSE：GET /{id}（状态）、/events?after=（事件流）、 + /tasks（DAG）、/workflow、/metrics、/audit、**/result**（综述+deliverable+artifacts） + 终态事件 swarm.completed 已直接带 summary+deliverable + + 追加输入（运行中或已完成续）：客户端 → HM → POST /{id}/input {instruction} + → 注入 user_append 任务（终态自动 reopen）→ 回 ⑤ 继续；原文不进事件流 + 停止：客户端 → HM → POST /{id}/stop → 取消任务 + 发 swarm.stopped + 拆除已拉起 agent +``` + +**要点**： +- **客户端不直连 Swarm / 不直连 agent**：读写都经 HM（鉴权/计费/审计在 HM；Swarm 运行时私网、service-token；agent 无公网子域，仅出站连编排器）。 +- **输入**：初始 prompt = create 的 `requirement.objective`；后续 prompt = `POST /{id}/input`（#40）。 +- **输出**：`summary`（synthesize 综述）+ `deliverable`（git 分支/commit/文件 + artifact_ids）+ `artifacts[].uri`（内容定位）；经 `swarm.completed` 事件与 `GET /{id}/result` 两条路拿到。 +- **密钥**：HM mint sk-→写 KV→传 `secret_ref`；Swarm 服务端解析注入 agent；全程不入 create 体/回调/日志/argv。 diff --git a/docs/integration/event-schema.md b/docs/integration/event-schema.md index 0e0403f..39d677c 100644 --- a/docs/integration/event-schema.md +++ b/docs/integration/event-schema.md @@ -78,7 +78,7 @@ | ⭐ `approval.approved` | `approval_id` | approval | ✅（审批决定落地时） | | ⭐ `approval.rejected` | `approval_id` | approval | ✅（审批决定落地时） | | ⭐ `artifact.created` | `artifact_id`, `uri`, `checksum` | artifact | ✅（带 §3 扁平字段） | -| ⭐ `swarm.completed` | `status` | swarm_terminal | ✅（终态，与 `deployment.status_changed` 并发） | +| ⭐ `swarm.completed` | `status`（+ `summary`/`deliverable` 可选） | swarm_terminal | ✅（终态；**携带用户面结果** = 综述 + deliverable 摘要，客户端可直接取；完整见 `GET …/{id}/result`，#40） | | ⭐ `swarm.failed` | `status` | swarm_terminal | ✅（终态） | | ⭐ `swarm.stopped` | `status` | swarm_terminal | ✅（`stop_run`） | | `deployment.status_changed` | `status` | deployment | ✅（HM 用于 AgentDeployment.Status；客户端用 swarm.* 终态） | diff --git a/docs/integration/runtime-contract.md b/docs/integration/runtime-contract.md index 11e8fc4..609458d 100644 --- a/docs/integration/runtime-contract.md +++ b/docs/integration/runtime-contract.md @@ -36,6 +36,8 @@ Swarm 暴露以下接口（三组别名等价，便于 HM 路径覆盖）： | task graph | `GET …/{deployment_id}/tasks` | ✅ 已实现 | 任务 DAG | | logs / events | `GET …/{deployment_id}/logs`、`/events` | ✅ 已实现 | 事件流（分页 `cursor`/`limit`） | | metrics / workflow / diagnostics | `GET …/{deployment_id}/metrics`、`/workflow`、`/diagnostics` | ✅ 已实现 | 指标与编排视图 | +| **input（追加输入）** | `POST …/{deployment_id}/input` | ✅ 已实现 | 接收用户后续 prompt，注入为 `source=user_append` 任务（终态 run 自动 reopen 为 running；`stopped` 拒绝）。指令文本仅作任务描述下发给 agent，**不回显进事件流**（task.created 仅类别 message）。见 #40 | +| **result（结果）** | `GET …/{deployment_id}/result` | ✅ 已实现 | 返回用户面结果：`{summary, deliverable, artifacts[], termination_reason, status}`；产物内容在各 artifact 的 `uri`（git/runtime），非内联 | | **cancel** | `POST …/{deployment_id}/stop` | ✅ 已实现 | 停止并取消非终态任务，向 Agent 下发 `cancel_task` | | **approve** | `POST …/{deployment_id}/approvals/{approval_id}` | ✅ 已实现 | 接收 Manager 审批决定（approved/rejected） | | **resume** | — | 🟡 待对齐 | 当前仅「审批通过」隐式恢复（approvals）；无独立 resume 端点 | diff --git a/orchestrator/main.py b/orchestrator/main.py index d2f6b06..f5b83e9 100644 --- a/orchestrator/main.py +++ b/orchestrator/main.py @@ -646,6 +646,14 @@ async def refresh_swarm_run_status(run): } if termination_reason: swarm_terminal_payload["termination_reason"] = termination_reason + # Carry the user-facing answer in the terminal event so the client gets the result from the + # one event it already watches (no reconstruction): the synthesized summary + deliverable + # facts (git branch/commit/files + artifact_ids). Full result also at GET …/{id}/result. + if next_status == "completed": + if final_summary: + swarm_terminal_payload["summary"] = final_summary + if deliverable: + swarm_terminal_payload["deliverable"] = deliverable await swarm_runtime.emit_event(run, f"swarm.{next_status}", payload=swarm_terminal_payload) timeline_payload = { "summary": final_summary or f"Swarm run {next_status}", @@ -2118,6 +2126,113 @@ async def get_swarm_audit(deployment_id: str, request: Request): return {"success": True, "data": await build_audit_trace_for_run(run)} +async def build_run_result(run) -> Dict[str, Any]: + """The swarm's user-facing answer for a run: synthesized summary + deliverable + artifacts. + + A first-class result object so the client doesn't reconstruct it from /workflow + /artifacts. + Carries only non-content descriptors (summary text + deliverable facts: git branch/commit/ + files + artifact uri/checksum) — the actual artifact content lives at its uri (git/runtime). + """ + tasks = await load_runtime_tasks(run) + termination_reason = (run.metadata.get("convergence") or {}).get("termination_reason") + return { + "deployment_id": run.deployment_id, + "swarm_id": run.swarm_id, + "status": run.status, + "runtime_execution_status": run.status, + "summary": run.metadata.get("final_summary"), + "termination_reason": termination_reason, + "deliverable": build_run_deliverable(run, tasks), + "artifacts": collect_run_artifacts(run, tasks), + } + + +@app.get("/api/agent/swarm/deployments/{deployment_id}/result") +@app.get("/api/swarms/{deployment_id}/result") +@app.get("/api/agnet/deployments/{deployment_id}/result") +async def get_swarm_result(deployment_id: str, request: Request): + """Return the swarm's result (summary + deliverable + artifacts) for a deployment.""" + run, auth_error = await get_runtime_run_or_404(request, deployment_id) + if auth_error: + return auth_error + return {"success": True, "data": await build_run_result(run)} + + +@app.post("/api/agent/swarm/deployments/{deployment_id}/input") +@app.post("/api/swarms/{deployment_id}/input") +@app.post("/api/agnet/deployments/{deployment_id}/input") +async def append_swarm_input(deployment_id: str, request: Request): + """Receive a user follow-up prompt and inject it into the running swarm (agent_swarm#40). + + The instruction becomes a new `source="user_append"` task in the run's shared pool (agents + self-select it). If the run already finished (completed/failed), it is reopened to `running` + ("continue"); a `stopped` run is rejected (start a new run). The instruction text is the task + description (delivered to the claiming agent over WS) but is **redacted from the event stream** + — the `task.created` event carries only a category message, never the raw prompt. + """ + auth_error = await require_runtime_auth(request) + if auth_error: + return auth_error + try: + body = await request.json() + except json.JSONDecodeError: + body = {} + instruction = (body.get("instruction") or body.get("input") or "").strip() + correlation_id = request.headers.get("x-correlation-id") + if not instruction: + return error_response(422, "INVALID_REQUEST", "instruction is required", correlation_id) + run = await swarm_runtime.get_run_by_identifier(deployment_id) + if not run: + raise HTTPException(status_code=404, detail="Swarm run not found") + if run.status == "stopped": + return error_response(409, "RUN_STOPPED", + "run was stopped; start a new run for new input", correlation_id) + + reopened = run.status in ("completed", "failed") + n = len([t for t in run.task_ids if "-input-" in t]) + 1 + task = await task_queue.create_task( + task_id=f"{run.swarm_id}-input-{n}", + title="用户追加输入", + description=instruction, # delivered to the agent; NOT echoed to events + agent_role="general", + required_capabilities=[], + root_task_id="seed", + source="user_append", + context={ + "swarm_id": run.swarm_id, + "runtime_deployment_id": run.deployment_id, + "manager_deployment_id": run.manager_deployment_id, + "correlation_id": run.correlation_id, + "source": "user_append", + "agent_role": "general", + "workflow_mode": "swarm", + "allow_handoff": True, + }, + enqueue=True, + ) + await swarm_runtime.attach_task(run, task.task_id) + if reopened: + run.status = "running" + await swarm_runtime.save_run(run) + await swarm_runtime.emit_event(run, "deployment.status_changed", + payload=swarm_runtime.status_payload(run, phase="Execute", reason="user appended input")) + # Redacted task.created — the raw instruction is NOT in the event (security-boundary). + await swarm_runtime.emit_event(run, "task.created", task_id=task.task_id, payload={ + "task_id": task.task_id, + "title": "用户追加输入", + "source": "user_append", + "agent_role": "general", + "message": "用户追加输入已注入运行中的蜂群", + }) + return {"success": True, "data": { + "deployment_id": run.deployment_id, + "swarm_id": run.swarm_id, + "task_id": task.task_id, + "status": run.status, + "reopened": reopened, + }} + + @app.post("/api/agent/swarm/deployments/{deployment_id}/stop") @app.post("/api/swarms/{deployment_id}/stop") @app.post("/api/agnet/deployments/{deployment_id}/stop") diff --git a/scripts/test-swarm-io.py b/scripts/test-swarm-io.py new file mode 100644 index 0000000..07677b8 --- /dev/null +++ b/scripts/test-swarm-io.py @@ -0,0 +1,98 @@ +"""Swarm I/O tests (agent_swarm#40 + result delivery): receive user prompts + return result. + +Exercises the real HTTP endpoints via FastAPI TestClient (hermetic: REDIS_FAKE, planner offline, +no connected agents, launch backend = none): + * POST /api/swarms/{id}/input — injects a `source="user_append"` task; the raw instruction is + delivered to the agent (task description) but **redacted from the event stream** (task.created + carries only a category message); rejects a stopped run. + * GET /api/swarms/{id}/result — returns {summary, deliverable, artifacts, termination_reason}. + +Run from agent_swarm_v6 (install deps first): + pip install -r orchestrator/requirements.txt + REDIS_FAKE=1 python scripts/test-swarm-io.py +""" +import json +import logging +import os +import sys +from pathlib import Path + +os.environ["REDIS_FAKE"] = "1" +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from fastapi.testclient import TestClient +from orchestrator import main as orch + +orch.planner.client = None # force planner offline (no model calls) +logging.getLogger("orchestrator.swarm_runtime").setLevel(logging.ERROR) + +failures = [] +SECRET_PROMPT = "SUPER-SECRET-USER-FOLLOWUP-PROMPT-9173-do-the-thing" + + +def check(name, cond): + print(("PASS" if cond else "FAIL"), "-", name) + if not cond: + failures.append(name) + + +def create_run(client): + body = {"mode": "swarm", "requirement": {"objective": "build add(a,b)"}, + "callback": {"url": "http://127.0.0.1:9/cb", "subscribed_events": []}, + "metadata": {"manager_deployment_id": "io-1"}} + r = client.post("/api/swarms", json=body) + return r.json()["data"]["deployment_id"] + + +def main(): + with TestClient(orch.app) as client: + dep = create_run(client) + check("run created", bool(dep)) + + # --- receive user prompt (append) --- + ri = client.post(f"/api/swarms/{dep}/input", json={"instruction": SECRET_PROMPT}) + check("input accepted (200/success)", ri.status_code == 200 and ri.json().get("success")) + idata = ri.json().get("data", {}) + check("input created a user_append task", bool(idata.get("task_id"))) + + # the appended task carries the instruction (agent-facing) ... + tasks = client.get(f"/api/swarms/{dep}/tasks").json()["data"]["tasks"] + appended = [t for t in tasks if t.get("source") == "user_append"] + check("appended task present with source=user_append", len(appended) == 1) + check("appended task description = the instruction (agent-facing)", + appended and appended[0].get("description") == SECRET_PROMPT) + + # ... but the instruction is NOT echoed into the event stream (redaction). + events = client.get(f"/api/swarms/{dep}/logs").json()["data"]["events"] + blob = json.dumps(events, ensure_ascii=False) + check("instruction NOT in any event payload (redacted)", SECRET_PROMPT not in blob) + tc = [e for e in events if e["event_type"] == "task.created" + and (e.get("payload") or {}).get("source") == "user_append"] + check("a redacted task.created(user_append) event was emitted", len(tc) == 1) + check("that event carries a category message, not the prompt", + tc and "message" in tc[0]["payload"] and SECRET_PROMPT not in json.dumps(tc[0]["payload"], ensure_ascii=False)) + + # --- result endpoint --- + res = client.get(f"/api/swarms/{dep}/result") + check("result endpoint 200/success", res.status_code == 200 and res.json().get("success")) + rdata = res.json().get("data", {}) + check("result has summary/deliverable/artifacts keys", + {"summary", "deliverable", "artifacts", "status"}.issubset(rdata.keys())) + check("result deliverable is a dict", isinstance(rdata.get("deliverable"), dict)) + + # --- stopped run rejects append --- + client.post(f"/api/swarms/{dep}/stop", json={"reason": "test stop"}) + rstop = client.post(f"/api/swarms/{dep}/input", json={"instruction": "more"}) + body = rstop.json() + check("append to stopped run rejected (RUN_STOPPED)", + body.get("success") is False and (body.get("error") or {}).get("code") == "RUN_STOPPED") + + print() + if failures: + print(f"{len(failures)} swarm-io check(s) FAILED: {failures}") + sys.exit(1) + print("all swarm-io checks passed") + + +if __name__ == "__main__": + main()