fix(pi-runner): cap /v1/run response size to stop typed reject

Truncate execution.log patches and keep members aligned with patch bodies so verbose MiniMax-M3 runs no longer exceed the Go 8MB fail-closed limit. Surface body-too-large vs typed-reject reasons and map contract failures to fatal restart instead of misleading resumable TASK_INTERRUPTED.
2026-09-23 10:22:23 +08:00
parent f1f1846925
commit 77028af546
15 changed files with 457 additions and 16 deletions
@@ -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`
@@ -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`；未操作共享服务写动作。
@@ -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
@@ -0,0 +1,2 @@
+ok  	chat-agent-system/internal/agent	3.037s
+ok  	chat-agent-system/internal/stream	1.133s
@@ -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
@@ -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}
@@ -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
+}
@@ -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,
@@ -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);
+  }
 });
@@ -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" }));
    }
@@ -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")
+	}
+}
@@ -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 {
@@ -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") ||
@@ -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)
+	}
+}
				`@@ -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}`