merge: feature/pi-runner-response-size-2026-09-23 (cap /v1/run response size)

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")
+	}
+}

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 {

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") ||

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)
+	}
+}