diff --git a/.doc/ops/short-term/2026-09-23-pi-runner-response-size.md b/.doc/ops/short-term/2026-09-23-pi-runner-response-size.md new file mode 100644 index 00000000..382100dc --- /dev/null +++ b/.doc/ops/short-term/2026-09-23-pi-runner-response-size.md @@ -0,0 +1,63 @@ +--- +title: "修复 PI runner 响应体积导致 typed result rejected" +task_id: "pi-runner-response-size-2026-09-23" +status: done +created: 2026-09-23 +updated: 2026-09-23 +owner: Cursor +task_type: implementation +task_tier: small +policy_revision: "agent-governance-2026-09-21-r1 @ f1f1846925f0253c2910245c9b5c2ca821127cee" +branch: feature/pi-runner-response-size-2026-09-23 +worktree: .worktrees/pi-runner-response-size-2026-09-23 +base_dev_sha: "f1f1846925f0253c2910245c9b5c2ca821127cee" +write_scope: ["deploy/pi-agent", "internal/agent", "internal/stream", ".doc"] +allowed_environments: [] +head_sha: "" +mutex: "pi-agent-runner+agent-remote-result" +review_state: pending +integration_state: pending +deployment: not_required +cleanup: pending +cleanup_owner: "独立维护者" +cleanup_trigger: "合入 dev 祖先且无进程占用后 remove" +remote_sync: not_requested +--- + +# 修复 PI runner 响应体积导致 typed result rejected + +## 用户目标与授权 + +- 用户目标：test 环境编写文档任务出现 `typed result rejected` / `TASK_INTERRUPTED`；按已确认计划修复。 +- 交付终点：实现验证 + 本地提交；不晋升；不强推；无环境写授权故不重建镜像。 +- 禁止项：不操作未授权环境；不改 production_ready。 + +## 范围与非目标 + +- 已改：`deploy/pi-agent` 响应体积治理；`internal/agent` ErrorReason；`internal/stream` 分类；任务/报告文档。 +- 非目标：刷新 UI 重放重写；扩大 AgentTimeout；改变 typed fail-closed。 + +## 步骤与验收 + +1. [x] runner.mjs：execution.log 截断 + patches 上限 +2. [x] result-v1.mjs：members/patches 一致 + node 测试 +3. [x] piagent.go：区分超限/校验失败 + Go 负例 +4. [x] classify.go：契约永久失败 → fatal/restart +5. [x] 本地验证 + 报告；真实 PI 重建记缺项 + +## 验证与证据 + +| 范围 / 被测 SHA | 命令与环境 | 预期 | 实际 / 判定 | 证据 / 脱敏说明 | +|---|---|---|---|---| +| Node + Go + arch @ feature tip | 见测试报告 | PASS | PASS | [.doc/test-reports/2026-09-23-pi-runner-response-size-test-report.md](../../test-reports/2026-09-23-pi-runner-response-size-test-report.md) | +| test 真实 PI 重建 | 需 env:test | 不再 typed reject | SKIP 缺项 | health 只读探针，无密钥 | + +## 收口与交接 + +- 剩余：合入后获授权重建 pi-agent 镜像并在 test 重跑 intake。 +- cleanup：owner 退出后由独立维护者按触发条件复查 remove。 +- 建议下一步：integrate 候选合入 `dev`，再授权 env 重建验收。 + +## 相关链接 + +- Test report: `.doc/test-reports/2026-09-23-pi-runner-response-size-test-report.md` diff --git a/.doc/test-reports/2026-09-23-pi-runner-response-size-test-report.md b/.doc/test-reports/2026-09-23-pi-runner-response-size-test-report.md new file mode 100644 index 00000000..0d0a356a --- /dev/null +++ b/.doc/test-reports/2026-09-23-pi-runner-response-size-test-report.md @@ -0,0 +1,81 @@ +# PI Runner 响应体积治理（typed result rejected）测试报告 + +## 一、 测试背景 (Background) + +test 环境执行「编写用户初始需求收集表」时，刷新后对话退化为「任务执行中」，约两分钟后出现 `TASK_INTERRUPTED`，upstream 为 `step step_01_organize failed: typed result rejected`。根因是 runner 将含全量 stdout 的 `execution.log` 塞进 `/v1/run` 响应体，易超过 Go 端 8MB fail-closed 上限。 + +## 二、 测试目标 (Test Objectives) + +- runner 对 `execution.log` patch 仅回传截断尾部；patches 受 per-member / 总量上限约束。 +- `withResultV1` 保证 members 与 patches body 一致。 +- Go 区分 `runner result body too large` 与 `typed result rejected: <原因>`，超限不落盘。 +- 契约类永久失败分类为 `RUNNER_RESULT_REJECTED` fatal，不再误报「可继续」。 + +## 三、 测试方案与环境 (Test Plan & Environment) + +| 项 | 值 | +|----|-----| +| 分支 | `feature/pi-runner-response-size-2026-09-23` | +| base_dev_sha | `f1f1846925f0253c2910245c9b5c2ca821127cee` | +| OS | macOS (darwin) / 证据采集时本机 | +| Node | `node --test deploy/pi-agent/*.test.mjs` | +| Go | `go test -count=1 -short ./internal/agent ./internal/stream` | +| 架构门 | `./scripts/arch-check.sh --mode block` | +| test 环境只读探针 | `:8081` / `:18866` health（未重建镜像） | + +## 四、 预期结果 (Expected Results) + +- Node 超限/截断/一致性用例全 PASS。 +- Go body-too-large / typed-tamper / classify 契约失败用例 PASS。 +- arch-check production gate passed。 +- 真实 PI 重建验收：本任务未登记 `env:test`/`env:dev` 写能力 → 记缺项。 + +## 五、 实际测试结果与判定 (Actual Results & Evaluation) + +### 用例 1：Node result-v1 / settle / narrative + +- **输入**：`node --test deploy/pi-agent/*.test.mjs` +- **实际**：9 PASS / 0 FAIL（含截断、成员剔除、wire 预算） +- **判定**：PASS +- **证据**：[node-test.txt](./assets/2026-09-23-pi-runner-response-size/node-test.txt) + +### 用例 2：Go agent + stream + +- **输入**：`go test -count=1 -short ./internal/agent ./internal/stream` +- **实际**：两包 ok；含 `BodyTooLargeDoesNotWrite`、`TypedTamperDoesNotWrite`、`TypedResultRejectedIsFatalNotResumable`、`BodyTooLargeIsFatalNotResumable` +- **判定**：PASS +- **证据**：[go-test.txt](./assets/2026-09-23-pi-runner-response-size/go-test.txt) + +### 用例 3：vet + arch-check + +- **输入**：`go vet ./internal/agent/... ./internal/stream/...`；`./scripts/arch-check.sh --mode block` +- **实际**：vet 空输出；`arch-check block: production gate passed` +- **判定**：PASS +- **证据**：[go-vet.txt](./assets/2026-09-23-pi-runner-response-size/go-vet.txt)、[arch-check.txt](./assets/2026-09-23-pi-runner-response-size/arch-check.txt) + +### 用例 4：test 环境真实 PI 重建验收 + +- **输入**：拟 `env-service.sh test` 重建 pi-agent 后重跑 intake +- **实际**：任务未授权环境写能力；当前 `shuimu-pi-agent-test` 仍为旧镜像（uptime ~13h），health 仅证明环境在线、`model_id=MiniMax-M3`、`schema_version=64` +- **判定**：SKIP / 缺项（非 PASS） +- **证据**：[test-app-health.json](./assets/2026-09-23-pi-runner-response-size/test-app-health.json)、[test-runner-health.json](./assets/2026-09-23-pi-runner-response-size/test-runner-health.json) + +## 六、 自动化测试套件汇总 (Test Suite Execution) + +| 套件 | 结果 | +|------|------| +| `node --test deploy/pi-agent/*.test.mjs` | PASS | +| `go test -short ./internal/agent ./internal/stream` | PASS | +| `go vet`（受影响包） | PASS | +| `arch-check --mode block` | PASS | +| 真实 PI / 镜像重建 | SKIP（无 env 写授权） | + +## 七、 测试结论与后续优化建议 (Conclusion & Next Steps) + +本地契约与单元层达到合入标准。刷新后「任务执行中」属设计内重放瞬时态，非本轮缺陷。 + +**前滚条件**：合入 dev 后，由获 `env:test`（或 dev）授权的维护者重建 `pi-agent` 镜像并重跑「编写用户初始需求收集表」，确认不再出现 `typed result rejected`。 + +**回滚**：恢复旧 `runner.mjs` / `result-v1.mjs`；Go fail-closed 与 classify 变更可独立回退。 + +不晋升 test/main；不改 `production_ready`；未操作共享服务写动作。 diff --git a/.doc/test-reports/assets/2026-09-23-pi-runner-response-size/arch-check.txt b/.doc/test-reports/assets/2026-09-23-pi-runner-response-size/arch-check.txt new file mode 100644 index 00000000..d1840881 --- /dev/null +++ b/.doc/test-reports/assets/2026-09-23-pi-runner-response-size/arch-check.txt @@ -0,0 +1,5 @@ +arch-check mode=block +boundaries=/private/tmp/Agent-System/.worktrees/pi-runner-response-size-2026-09-23/.doc/system-design/module-boundaries.yaml +ok chat-agent-system/internal/archtest 0.540s +ok chat-agent-system/internal/archtest 0.536s +arch-check block: production gate passed diff --git a/.doc/test-reports/assets/2026-09-23-pi-runner-response-size/go-test.txt b/.doc/test-reports/assets/2026-09-23-pi-runner-response-size/go-test.txt new file mode 100644 index 00000000..2fe70d72 --- /dev/null +++ b/.doc/test-reports/assets/2026-09-23-pi-runner-response-size/go-test.txt @@ -0,0 +1,2 @@ +ok chat-agent-system/internal/agent 3.037s +ok chat-agent-system/internal/stream 1.133s diff --git a/.doc/test-reports/assets/2026-09-23-pi-runner-response-size/go-vet.txt b/.doc/test-reports/assets/2026-09-23-pi-runner-response-size/go-vet.txt new file mode 100644 index 00000000..e69de29b diff --git a/.doc/test-reports/assets/2026-09-23-pi-runner-response-size/node-test.txt b/.doc/test-reports/assets/2026-09-23-pi-runner-response-size/node-test.txt new file mode 100644 index 00000000..f74a6f4e --- /dev/null +++ b/.doc/test-reports/assets/2026-09-23-pi-runner-response-size/node-test.txt @@ -0,0 +1,19 @@ +change-report-narrative-v1.test.mjs: ok +✔ deploy/pi-agent/change-report-narrative-v1.test.mjs (61.937041ms) +✔ withResultV1 is additive and strips stderr (1.191542ms) +✔ typedErrorCode is allowlisted and ignores raw stderr (0.104958ms) +✔ path escape members are omitted (0.778875ms) +✔ truncateExecutionLogContent keeps only the tail (0.178333ms) +✔ oversized patch content is dropped from both members and patches (0.854459ms) +✔ capPatches truncates execution.log and enforces total wire budget (0.32525ms) +✔ withResultV1 wire estimate stays under soft budget for truncated logs (3.463917ms) +settle-policy.test.mjs PASS +✔ deploy/pi-agent/settle-policy.test.mjs (56.850792ms) +ℹ tests 9 +ℹ suites 0 +ℹ pass 9 +ℹ fail 0 +ℹ cancelled 0 +ℹ skipped 0 +ℹ todo 0 +ℹ duration_ms 89.694625 diff --git a/.doc/test-reports/assets/2026-09-23-pi-runner-response-size/test-app-health.json b/.doc/test-reports/assets/2026-09-23-pi-runner-response-size/test-app-health.json new file mode 100644 index 00000000..c8bd5100 --- /dev/null +++ b/.doc/test-reports/assets/2026-09-23-pi-runner-response-size/test-app-health.json @@ -0,0 +1 @@ +{"status":"ok","env":"test","agent_backend":"pi-agent","version":"v1.0","shared_kb_port":5432,"kb_ready":true,"database_ready":true,"schema_version":64,"schema_ready":true,"task_store_ready":true} diff --git a/.doc/test-reports/assets/2026-09-23-pi-runner-response-size/test-runner-health.json b/.doc/test-reports/assets/2026-09-23-pi-runner-response-size/test-runner-health.json new file mode 100644 index 00000000..822b70aa --- /dev/null +++ b/.doc/test-reports/assets/2026-09-23-pi-runner-response-size/test-runner-health.json @@ -0,0 +1,14 @@ +{ + "ok": true, + "service": "shuimu-pi-agent-runner", + "endpoint_url": "https://api.minimaxi.com/v1", + "model_id": "MiniMax-M3", + "api": "openai-compatible", + "has_token": true, + "tavily": { + "active": true, + "keys_count": 2 + }, + "data_dir": "/app/data/workspaces", + "uptime": 47409.864285404 +} diff --git a/deploy/pi-agent/result-v1.mjs b/deploy/pi-agent/result-v1.mjs index 20ea4d80..ac4f9fda 100644 --- a/deploy/pi-agent/result-v1.mjs +++ b/deploy/pi-agent/result-v1.mjs @@ -3,6 +3,10 @@ import { createHash } from "node:crypto"; export const RESULT_V1_SCHEMA_VERSION = 1; export const MAX_RESULT_MEMBERS = 32; export const MAX_RESULT_MEMBER_BYTES = 2 * 1024 * 1024; +/** Soft cap for HTTP wire payloads; Go fail-closes at 8MB. */ +export const MAX_RESULT_WIRE_BYTES = 6 * 1024 * 1024; +/** Tail window for execution.log patches; full log remains on shared volume. */ +export const EXECUTION_LOG_PATCH_BYTES = 128 * 1024; const RETRYABLE_CODES = new Set(["TIMEOUT", "CANCELLED", "TRANSIENT"]); const ALLOWED_CODES = new Set([ @@ -46,17 +50,56 @@ export function membersDigest(members) { return createHash("sha256").update(JSON.stringify(rows)).digest("hex"); } +/** Keep only the trailing bytes of execution.log for the HTTP patch wire. */ +export function truncateExecutionLogContent(content, maxBytes = EXECUTION_LOG_PATCH_BYTES) { + const buf = Buffer.from(String(content || ""), "utf8"); + if (buf.length <= maxBytes) { + return buf.toString("utf8"); + } + const tail = buf.subarray(buf.length - maxBytes); + return `[truncated ${buf.length - maxBytes} bytes]\n` + tail.toString("utf8"); +} + +/** + * Cap patch bodies so members and patches stay aligned under wire limits. + * Oversized / overflow members are dropped from both lists (fail closed on wire size). + */ +export function capPatches(patches, opts = {}) { + const maxMember = opts.maxMemberBytes ?? MAX_RESULT_MEMBER_BYTES; + const maxTotal = opts.maxTotalBytes ?? MAX_RESULT_WIRE_BYTES; + const maxMembers = opts.maxMembers ?? MAX_RESULT_MEMBERS; + const out = []; + let total = 0; + for (const p of Array.isArray(patches) ? patches : []) { + const rel = String(p?.rel_path || ""); + if (!rel || rel.includes("..") || rel.startsWith("/")) continue; + let content = String(p.content || ""); + if (rel === "execution.log" || rel.endsWith("/execution.log")) { + content = truncateExecutionLogContent(content); + } + const buf = Buffer.from(content, "utf8"); + if (buf.length > maxMember) continue; + if (out.length >= maxMembers) break; + if (total + buf.length > maxTotal) continue; + total += buf.length; + out.push({ rel_path: rel, content }); + } + return out; +} + +export function estimateWireBytes(payload) { + return Buffer.byteLength(JSON.stringify(payload), "utf8"); +} + export function withResultV1(payload) { const out = { ...payload }; delete out.stderr; - const patches = Array.isArray(out.patches) ? out.patches : []; + const capped = capPatches(Array.isArray(out.patches) ? out.patches : []); + out.patches = capped; const members = []; - for (const p of patches) { + for (const p of capped) { const rel = String(p?.rel_path || ""); - if (!rel || rel.includes("..") || rel.startsWith("/")) continue; const buf = Buffer.from(String(p.content || ""), "utf8"); - if (buf.length > MAX_RESULT_MEMBER_BYTES) continue; - if (members.length >= MAX_RESULT_MEMBERS) break; members.push({ kind: "file", path: rel, diff --git a/deploy/pi-agent/result-v1.test.mjs b/deploy/pi-agent/result-v1.test.mjs index f5334f3b..8f89cdd2 100644 --- a/deploy/pi-agent/result-v1.test.mjs +++ b/deploy/pi-agent/result-v1.test.mjs @@ -1,6 +1,16 @@ import { test } from "node:test"; import assert from "node:assert/strict"; -import { withResultV1, membersDigest, typedErrorCode } from "./result-v1.mjs"; +import { + withResultV1, + membersDigest, + typedErrorCode, + capPatches, + truncateExecutionLogContent, + estimateWireBytes, + MAX_RESULT_MEMBER_BYTES, + MAX_RESULT_WIRE_BYTES, + EXECUTION_LOG_PATCH_BYTES, +} from "./result-v1.mjs"; test("withResultV1 is additive and strips stderr", () => { const out = withResultV1({ @@ -49,4 +59,70 @@ test("path escape members are omitted", () => { ], }); assert.deepEqual(out.result_v1.members.map((m) => m.path), ["output.md"]); + assert.deepEqual(out.patches.map((p) => p.rel_path), ["output.md"]); +}); + +test("truncateExecutionLogContent keeps only the tail", () => { + const big = "A".repeat(EXECUTION_LOG_PATCH_BYTES + 4096); + const out = truncateExecutionLogContent(big); + assert.ok(out.startsWith("[truncated ")); + assert.ok(out.length < big.length); + assert.ok(Buffer.byteLength(out, "utf8") <= EXECUTION_LOG_PATCH_BYTES + 64); +}); + +test("oversized patch content is dropped from both members and patches", () => { + const huge = "x".repeat(MAX_RESULT_MEMBER_BYTES + 1); + const out = withResultV1({ + ok: true, + run_id: "run-4", + patches: [ + { rel_path: "output.md", content: "# ok\n" }, + { rel_path: "blob.bin", content: huge }, + ], + }); + assert.deepEqual(out.patches.map((p) => p.rel_path), ["output.md"]); + assert.deepEqual(out.result_v1.members.map((m) => m.path), ["output.md"]); +}); + +test("capPatches truncates execution.log and enforces total wire budget", () => { + const log = "L".repeat(EXECUTION_LOG_PATCH_BYTES * 4); + const patches = capPatches([ + { rel_path: "output.md", content: "# deliverable\n" }, + { rel_path: "execution.log", content: log }, + ]); + assert.equal(patches.length, 2); + assert.ok(patches.find((p) => p.rel_path === "execution.log").content.startsWith("[truncated ")); + const total = patches.reduce((n, p) => n + Buffer.byteLength(p.content, "utf8"), 0); + assert.ok(total <= MAX_RESULT_WIRE_BYTES); + + // Force total budget with a custom soft cap under the member limit. + const budget = 1024; + const filler = "F".repeat(budget - 2); + const tight = capPatches( + [ + { rel_path: "a.md", content: filler }, + { rel_path: "b.md", content: "too-much" }, + ], + { maxTotalBytes: budget, maxMemberBytes: budget }, + ); + assert.deepEqual(tight.map((p) => p.rel_path), ["a.md"]); +}); + +test("withResultV1 wire estimate stays under soft budget for truncated logs", () => { + const log = "E".repeat(8 * 1024 * 1024); + const out = withResultV1({ + ok: true, + run_id: "run-big", + patches: [ + { rel_path: "output.md", content: "# intake\n".repeat(100) }, + { rel_path: "execution.log", content: log }, + ], + }); + assert.ok(estimateWireBytes(out) < MAX_RESULT_WIRE_BYTES); + assert.equal(out.result_v1.members.length, out.patches.length); + for (const m of out.result_v1.members) { + const body = out.patches.find((p) => p.rel_path === m.path); + assert.ok(body); + assert.equal(Buffer.byteLength(body.content, "utf8"), m.size); + } }); diff --git a/deploy/pi-agent/runner.mjs b/deploy/pi-agent/runner.mjs index 8dc68034..9b032937 100755 --- a/deploy/pi-agent/runner.mjs +++ b/deploy/pi-agent/runner.mjs @@ -11,7 +11,13 @@ import { readFileSync, writeFileSync, existsSync, mkdirSync } from "node:fs"; import { resolve, relative, isAbsolute } from "node:path"; import { TavilyKeyPool, searchTavily, formatTavilyResults } from "./extensions/tavily-search.mjs"; import { applyAgentEvent, classifyRunOutcome, newSettleState } from "./settle-policy.mjs"; -import { withResultV1 } from "./result-v1.mjs"; +import { + withResultV1, + capPatches, + truncateExecutionLogContent, + estimateWireBytes, + MAX_RESULT_WIRE_BYTES, +} from "./result-v1.mjs"; import { buildPiLaunchArgs } from "./pi-args.mjs"; const PORT = parseInt(process.env.PORT || process.env.PI_AGENT_RUNNER_PORT || "18766", 10); @@ -76,8 +82,15 @@ function parseJsonBody(req) { } function sendJson(res, statusCode, data) { + // Compact JSON: pretty-print + oversized patches previously ballooned past Go's 8MB fail-closed. + const body = JSON.stringify(data) + "\n"; + if (Buffer.byteLength(body, "utf8") > MAX_RESULT_WIRE_BYTES + 512 * 1024) { + console.warn( + `[runner] response body ${Buffer.byteLength(body, "utf8")} bytes exceeds soft wire budget`, + ); + } res.writeHead(statusCode, { "Content-Type": "application/json; charset=utf-8" }); - res.end(JSON.stringify(data, null, 2) + "\n"); + res.end(body); } function authorizeRunner(req, res) { @@ -287,15 +300,20 @@ async function handleRun(body) { patches.push({ rel_path: rel, content }); } - // Also include execution.log as patch if present so host definitely receives it + // execution.log full text stays on the shared volume; only a capped tail goes on the wire. const execLogPath = resolve(workspace, "execution.log"); if (existsSync(execLogPath)) { try { const logContent = readFileSync(execLogPath, "utf8"); - patches.push({ rel_path: "execution.log", content: logContent }); + patches.push({ + rel_path: "execution.log", + content: truncateExecutionLogContent(logContent), + }); } catch {} } + const cappedPatches = capPatches(patches); + if (missing.length > 0) { const classified = classifyRunOutcome({ ...settle, @@ -308,7 +326,7 @@ async function handleRun(body) { run_id: runId, retryable: classified.retryable, fallback_reason: classified.fallback_reason, - patches, + patches: cappedPatches, }; } @@ -316,7 +334,7 @@ async function handleRun(body) { ok: true, agent_id: agentId, run_id: runId, - patches, + patches: cappedPatches, fallback_reason: "", }; } @@ -397,7 +415,18 @@ const server = http.createServer(async (req, res) => { try { const body = await parseJsonBody(req); const result = await handleRun(body); - sendJson(res, 200, withResultV1(result)); + const typed = withResultV1(result); + const wireBytes = estimateWireBytes(typed); + if (wireBytes > MAX_RESULT_WIRE_BYTES) { + console.warn(`[runner] typed result wire ${wireBytes} bytes; dropping non-target patches`); + typed.patches = capPatches( + (typed.patches || []).filter((p) => p.rel_path !== "execution.log"), + ); + const shrunk = withResultV1({ ...typed, patches: typed.patches }); + sendJson(res, 200, shrunk); + return; + } + sendJson(res, 200, typed); } catch (err) { sendJson(res, 500, withResultV1({ ok: false, error: "runner_failed", patches: [], run_id: "unknown" })); } diff --git a/internal/agent/backend_test.go b/internal/agent/backend_test.go index 5b65e0e7..1df81878 100644 --- a/internal/agent/backend_test.go +++ b/internal/agent/backend_test.go @@ -299,7 +299,7 @@ func TestPiAgentRemoteRunner_TypedTamperDoesNotWrite(t *testing.T) { if err != nil { t.Fatal(err) } - if res.Success || res.ErrorReason != "typed result rejected" { + if res.Success || !strings.HasPrefix(res.ErrorReason, "typed result rejected") { t.Fatalf("tamper must fail closed, got success=%v reason=%q", res.Success, res.ErrorReason) } if _, serr := os.Stat(filepath.Join(dir, "output.md")); serr == nil { @@ -332,3 +332,39 @@ func TestPiAgentRemoteRunner_ClientTimeout(t *testing.T) { t.Fatalf("timeout reason leaked stderr: %q", res.ErrorReason) } } + +func TestPiAgentRemoteRunner_BodyTooLargeDoesNotWrite(t *testing.T) { + dir := t.TempDir() + // JSON body > MaxResultBodyBytes (8MiB). Content-Length alone is not enough; + // LimitReader stops after MaxResultBodyBytes+1 of payload bytes. + payload := make([]byte, agent.MaxResultBodyBytes+1024) + for i := range payload { + payload[i] = 'A' + } + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write(payload) + })) + defer ts.Close() + res, err := agent.NewBackend("pi-agent", &config.Config{ + AgentBackend: "pi-agent", + PiAgentRunnerURL: ts.URL, + PiAgentModel: "MiniMax-M3", + PiAgentProvider: "minimax", + AgentTimeout: 5 * time.Second, + }).Execute(context.Background(), agent.Request{ + WorkDir: dir, Prompt: "draft", Targets: []string{"output.md"}, + }) + if err != nil { + t.Fatal(err) + } + if res.Success || res.ErrorReason != "runner result body too large" { + t.Fatalf("oversized body must fail closed, got success=%v reason=%q", res.Success, res.ErrorReason) + } + if res.Retryable { + t.Fatal("body-too-large must not be retryable") + } + if _, serr := os.Stat(filepath.Join(dir, "output.md")); serr == nil { + t.Fatal("must not materialize after body-too-large reject") + } +} diff --git a/internal/agent/piagent.go b/internal/agent/piagent.go index 1c61abba..615e6c06 100644 --- a/internal/agent/piagent.go +++ b/internal/agent/piagent.go @@ -187,12 +187,32 @@ func (p *PiAgentBackend) executeRemote(ctx context.Context, runnerURL string, re return &Result{Success: false, ErrorReason: "failed to read remote response", Retryable: true}, nil } if len(limited) > MaxResultBodyBytes { - return &Result{Success: false, ErrorReason: "typed result rejected"}, nil + return &Result{Success: false, ErrorReason: "runner result body too large"}, nil } decoded, decErr := DecodeRemoteRun(limited, expectedRunID, true) if decErr != nil { - return &Result{Success: false, ErrorReason: "typed result rejected"}, nil + reason := "typed result rejected" + msg := decErr.Error() + switch { + case strings.Contains(msg, "schema"): + reason = "typed result rejected: schema" + case strings.Contains(msg, "run mismatch"): + reason = "typed result rejected: run mismatch" + case strings.Contains(msg, "digest"): + reason = "typed result rejected: digest" + case strings.Contains(msg, "missing result_v1"): + reason = "typed result rejected: missing result_v1" + case strings.Contains(msg, "member"): + reason = "typed result rejected: member" + case strings.Contains(msg, "outcome"), strings.Contains(msg, "error_code"): + reason = "typed result rejected: outcome" + case strings.Contains(msg, "empty body"): + reason = "typed result rejected: empty body" + case strings.Contains(msg, "failed to decode"): + reason = "typed result rejected: decode" + } + return &Result{Success: false, ErrorReason: reason}, nil } if !decoded.OK { diff --git a/internal/stream/classify.go b/internal/stream/classify.go index 25bd8692..7754671d 100644 --- a/internal/stream/classify.go +++ b/internal/stream/classify.go @@ -53,6 +53,13 @@ func ClassifyError(err error, phaseID, requestID, sessionID, workspaceID string) info.UserMessage = "网络连接异常，请稍后重试" info.Actions = []string{"retry"} + case isRunnerContractError(msg): + // Permanent runner wire/contract failures must not look "resumable". + info.Code = "RUNNER_RESULT_REJECTED" + info.Class = ErrorClassFatal + info.UserMessage = "助手执行结果校验失败，请重新开始任务" + info.Actions = []string{"restart"} + case workspaceID != "": info.Code = "TASK_INTERRUPTED" info.Class = ErrorClassResumable @@ -89,6 +96,11 @@ func isDocumentLedgerError(msg string) bool { strings.Contains(msg, "document_records_approver_id_fkey") } +func isRunnerContractError(msg string) bool { + return strings.Contains(msg, "typed result rejected") || + strings.Contains(msg, "runner result body too large") +} + func isTimeoutError(msg string) bool { return strings.Contains(msg, "timeout") || strings.Contains(msg, "deadline") || diff --git a/internal/stream/classify_test.go b/internal/stream/classify_test.go index 33556aa8..029a9194 100644 --- a/internal/stream/classify_test.go +++ b/internal/stream/classify_test.go @@ -120,3 +120,43 @@ func TestClassifyError_DocumentLedgerFKIsLedgerError(t *testing.T) { t.Fatalf("user message leaked SQL: %q", info.UserMessage) } } + +func TestClassifyError_TypedResultRejectedIsFatalNotResumable(t *testing.T) { + info := ClassifyError( + errors.New("step step_01_organize failed: typed result rejected: digest"), + "step_executing", + "run_1", + "sess_1", + "ws_bizdev_1", + ) + if info.Code != "RUNNER_RESULT_REJECTED" { + t.Fatalf("code=%s", info.Code) + } + if info.Class != ErrorClassFatal { + t.Fatalf("class=%s (must not be resumable)", info.Class) + } + if len(info.Actions) != 1 || info.Actions[0] != "restart" { + t.Fatalf("actions=%v", info.Actions) + } + for _, a := range info.Actions { + if a == "continue" { + t.Fatal("typed reject must not offer continue") + } + } +} + +func TestClassifyError_BodyTooLargeIsFatalNotResumable(t *testing.T) { + info := ClassifyError( + errors.New("step step_01_organize failed: runner result body too large"), + "step_executing", + "run_2", + "sess_1", + "ws_bizdev_2", + ) + if info.Code != "RUNNER_RESULT_REJECTED" { + t.Fatalf("code=%s", info.Code) + } + if info.Class != ErrorClassFatal { + t.Fatalf("class=%s", info.Class) + } +}