diff --git a/internal/agent/backend.go b/internal/agent/backend.go index 34b6e406..d18cbbbb 100644 --- a/internal/agent/backend.go +++ b/internal/agent/backend.go @@ -20,6 +20,9 @@ type Result struct { Patches map[string][]byte `json:"patches"` Logs string `json:"logs"` ErrorReason string `json:"error_reason"` + // Retryable marks a provider-transient failure that may receive a + // fresh step invocation. Permanent missing deliverables stay false. + Retryable bool `json:"retryable,omitempty"` } type Backend interface { diff --git a/internal/agent/backend_test.go b/internal/agent/backend_test.go index cb082435..5d9a2cb3 100644 --- a/internal/agent/backend_test.go +++ b/internal/agent/backend_test.go @@ -186,3 +186,80 @@ func TestPiAgentRemoteRunner_PathAliasesAndMissing(t *testing.T) { run("dot", "# from dot rel_path\n", true) run("missing", "", false) } + +func TestPiAgentRemoteRunner_PropagatesRetryable(t *testing.T) { + dir := t.TempDir() + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _ = json.NewEncoder(w).Encode(map[string]any{ + "ok": false, + "retryable": true, + "fallback_reason": "Connection error.", + "patches": []any{}, + }) + })) + 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.Retryable { + t.Fatalf("want retryable failure, success=%v retryable=%v reason=%s", res.Success, res.Retryable, res.ErrorReason) + } + if _, serr := os.Stat(filepath.Join(dir, "output.md")); serr == nil { + t.Fatal("must not invent output.md") + } +} + +func TestPiAgentRemoteRunner_OldSettleLieIsRetryable(t *testing.T) { + dir := t.TempDir() + log := "Outcome: SUCCESS\n" + + `{"type":"message_end","message":{"stopReason":"error","errorMessage":"terminated"}}` + "\n" + + `{"type":"auto_retry_end","success":false,"attempt":3,"finalError":"Connection error."}` + "\n" + + `{"type":"agent_settled"}` + "\n" + if err := os.WriteFile(filepath.Join(dir, "execution.log"), []byte(log), 0o644); err != nil { + t.Fatal(err) + } + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _ = json.NewEncoder(w).Encode(map[string]any{ + "ok": true, + "patches": []map[string]string{{ + "rel_path": "execution.log", + "content": "Outcome: SUCCESS\n", + }}, + }) + })) + 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 { + t.Fatal("process-0 settle without output.md must not succeed") + } + if !res.Retryable { + t.Fatalf("old runner lie + provider trace must be retryable, reason=%s", res.ErrorReason) + } + if _, serr := os.Stat(filepath.Join(dir, "output.md")); serr == nil { + t.Fatal("must not reconstruct output.md from truncated write args") + } +} diff --git a/internal/agent/piagent.go b/internal/agent/piagent.go index 389a4856..e0fd8b2d 100644 --- a/internal/agent/piagent.go +++ b/internal/agent/piagent.go @@ -74,16 +74,19 @@ func (p *PiAgentBackend) Execute(ctx context.Context, req Request) (*Result, err err := cmd.Run() if err != nil { - return &Result{ + res := &Result{ Success: false, ErrorReason: fmt.Sprintf("pi execution error: %v, stderr: %s", err, stderr.String()), - }, nil + } + res.Retryable = AssessStepFailure(StepFailureSignals{Result: res, WorkDir: workDir}).Class == FailureTransient + return res, nil } type bridgeResponse struct { OK bool `json:"ok"` FallbackReason string `json:"fallback_reason"` Stderr string `json:"stderr"` + Retryable bool `json:"retryable"` } var bResp bridgeResponse lines := strings.Split(strings.TrimSpace(stdout.String()), "\n") @@ -99,21 +102,32 @@ func (p *PiAgentBackend) Execute(ctx context.Context, req Request) (*Result, err if errMsg == "" { errMsg = "pi bridge reported task failed" } - return &Result{ + res := &Result{ Success: false, ErrorReason: errMsg, Logs: stdout.String(), - }, nil + Retryable: bResp.Retryable, + } + if !res.Retryable { + res.Retryable = AssessStepFailure(StepFailureSignals{Result: res, WorkDir: workDir}).Class == FailureTransient + } + return res, nil } patches, err := MaterializeTargets(workDir, req.Targets, nil) if err != nil { - return &Result{ + res := &Result{ Success: false, ErrorReason: err.Error(), Patches: patches, Logs: stdout.String(), - }, nil + } + res.Retryable = AssessStepFailure(StepFailureSignals{ + Result: res, + WorkDir: workDir, + TargetMissing: true, + }).Class == FailureTransient + return res, nil } return &Result{ @@ -150,6 +164,7 @@ func (p *PiAgentBackend) executeRemote(ctx context.Context, runnerURL string, re return &Result{ Success: false, ErrorReason: fmt.Sprintf("pi remote runner call failed: %v", err), + Retryable: true, }, nil } defer resp.Body.Close() @@ -158,6 +173,7 @@ func (p *PiAgentBackend) executeRemote(ctx context.Context, runnerURL string, re OK bool `json:"ok"` FallbackReason string `json:"fallback_reason"` Stderr string `json:"stderr"` + Retryable bool `json:"retryable"` Patches []struct { RelPath string `json:"rel_path"` Content string `json:"content"` @@ -178,10 +194,15 @@ func (p *PiAgentBackend) executeRemote(ctx context.Context, runnerURL string, re if reason == "" { reason = "remote pi-agent reported task failure" } - return &Result{ + res := &Result{ Success: false, ErrorReason: reason, - }, nil + Retryable: remoteResp.Retryable, + } + if !res.Retryable { + res.Retryable = AssessStepFailure(StepFailureSignals{Result: res, WorkDir: workDir}).Class == FailureTransient + } + return res, nil } incoming := make(map[string][]byte, len(remoteResp.Patches)) @@ -190,11 +211,17 @@ func (p *PiAgentBackend) executeRemote(ctx context.Context, runnerURL string, re } patches, matErr := MaterializeTargets(workDir, req.Targets, incoming) if matErr != nil { - return &Result{ + res := &Result{ Success: false, ErrorReason: matErr.Error(), Patches: patches, - }, nil + } + res.Retryable = remoteResp.Retryable || AssessStepFailure(StepFailureSignals{ + Result: res, + WorkDir: workDir, + TargetMissing: true, + }).Class == FailureTransient + return res, nil } return &Result{ diff --git a/internal/agent/step_retry.go b/internal/agent/step_retry.go new file mode 100644 index 00000000..ce46e491 --- /dev/null +++ b/internal/agent/step_retry.go @@ -0,0 +1,186 @@ +package agent + +import ( + "os" + "path/filepath" + "strings" + "unicode" +) + +// MaxTransientStepInvocations is the product-layer bound: the first +// invocation plus one fresh re-invocation. SDK in-process auto_retry is +// separate and must not be confused with this. +const MaxTransientStepInvocations = 2 + +type FailureClass int + +const ( + FailureNone FailureClass = iota + FailureTransient + FailurePermanent +) + +func (c FailureClass) String() string { + switch c { + case FailureTransient: + return "transient" + case FailurePermanent: + return "permanent" + default: + return "none" + } +} + +type TraceSignals struct { + ProviderTransient bool + Settled bool + RetryExhausted bool + HasUsableDeliverable bool // always false: write() args in traces are not deliverables +} + +type StepFailureSignals struct { + Err error + Result *Result + WorkDir string + ExtraText string + TargetMissing bool +} + +type FailureAssessment struct { + Class FailureClass + Reason string +} + +func ScanTraceText(text string) TraceSignals { + var out TraceSignals + if text == "" { + return out + } + // Line scan only. Never parse write.arguments.content into a body. + for _, line := range strings.Split(text, "\n") { + if line == "" { + continue + } + lower := strings.ToLower(line) + if strings.Contains(lower, `"type":"agent_settled"`) || strings.Contains(lower, `"type": "agent_settled"`) { + out.Settled = true + } + if strings.Contains(lower, `"type":"auto_retry_end"`) && strings.Contains(lower, `"success":false`) { + out.ProviderTransient = true + out.RetryExhausted = true + } + if strings.Contains(lower, `"type":"auto_retry_start"`) { + out.ProviderTransient = true + } + if strings.Contains(lower, `"stopreason":"error"`) { + out.ProviderTransient = true + } + if strings.Contains(lower, "connection error") { + out.ProviderTransient = true + } + if strings.Contains(lower, `"errormessage":"terminated"`) || strings.Contains(lower, `"errormessage": "terminated"`) { + out.ProviderTransient = true + } + } + return out +} + +func ReadWorkDirTraces(workDir string) string { + if strings.TrimSpace(workDir) == "" { + return "" + } + p := filepath.Join(workDir, "execution.log") + b, err := os.ReadFile(p) + if err != nil { + return "" + } + const capBytes = 512 << 10 + if len(b) > capBytes { + b = b[len(b)-capBytes:] + } + return string(b) +} + +func AssessStepFailure(sig StepFailureSignals) FailureAssessment { + reason := combinedFailureText(sig) + lower := strings.ToLower(reason) + trace := ScanTraceText(sig.ExtraText + "\n" + ReadWorkDirTraces(sig.WorkDir)) + missing := sig.TargetMissing || strings.Contains(lower, "missing target deliverable") + + if isPermanentReason(lower) && !trace.ProviderTransient { + return FailureAssessment{Class: FailurePermanent, Reason: "permanent: " + compactReason(reason)} + } + if isPermanentReason(lower) && !missing { + return FailureAssessment{Class: FailurePermanent, Reason: "permanent: " + compactReason(reason)} + } + + if sig.Result != nil && sig.Result.Retryable { + return FailureAssessment{Class: FailureTransient, Reason: "runner-retryable: " + compactReason(reason)} + } + if trace.ProviderTransient { + kind := "provider-transient" + if missing && trace.Settled { + kind = "settled-failure-missing-target" + } + return FailureAssessment{Class: FailureTransient, Reason: kind + ": " + compactReason(reason)} + } + if hasTransientKeyword(lower) { + return FailureAssessment{Class: FailureTransient, Reason: "provider-transient: " + compactReason(reason)} + } + if missing { + return FailureAssessment{Class: FailurePermanent, Reason: "permanent-missing: " + compactReason(reason)} + } + return FailureAssessment{Class: FailurePermanent, Reason: "fail-closed: " + compactReason(reason)} +} + +func combinedFailureText(sig StepFailureSignals) string { + var parts []string + if sig.Err != nil { + parts = append(parts, sig.Err.Error()) + } + if sig.Result != nil { + if sig.Result.ErrorReason != "" { + parts = append(parts, sig.Result.ErrorReason) + } + if sig.Result.Logs != "" { + parts = append(parts, sig.Result.Logs) + } + } + return strings.Join(parts, " ") +} + +func isPermanentReason(lower string) bool { + return strings.Contains(lower, "prompt_empty") || + strings.Contains(lower, "unauthorized") || + strings.Contains(lower, "治理强制终止") || + strings.Contains(lower, "治理拦截") +} + +func hasTransientKeyword(lower string) bool { + return strings.Contains(lower, "connection error") || + strings.Contains(lower, "terminated") || + strings.Contains(lower, "auto_retry") || + strings.Contains(lower, "econnreset") || + strings.Contains(lower, "socket hang up") || + strings.Contains(lower, "pi remote runner call failed") +} + +func compactReason(s string) string { + s = strings.Join(strings.FieldsFunc(s, func(r rune) bool { + return r == '\n' || r == '\r' + }), " ") + s = strings.TrimSpace(s) + if s == "" { + return "unknown" + } + // Drop accidental write-argument bodies if a caller stuffed logs here. + if len(s) > 240 { + s = s[:240] + } + return strings.Map(func(r rune) rune { + if unicode.IsControl(r) && r != '\t' { + return -1 + } + return r + }, s) +} diff --git a/internal/agent/step_retry_test.go b/internal/agent/step_retry_test.go new file mode 100644 index 00000000..f1a0c1f6 --- /dev/null +++ b/internal/agent/step_retry_test.go @@ -0,0 +1,143 @@ +package agent + +import ( + "errors" + "os" + "path/filepath" + "strings" + "testing" +) + +// Synthetic UI-11 timeline — event types only. Do not include write.content. +const ui11SettledFailureTrace = ` +{"type":"message_end","message":{"stopReason":"error","errorMessage":"terminated"}} +{"type":"auto_retry_start","attempt":1,"maxAttempts":3,"errorMessage":"terminated"} +{"type":"message_end","message":{"stopReason":"error","errorMessage":"Connection error."}} +{"type":"auto_retry_start","attempt":2,"maxAttempts":3,"errorMessage":"Connection error."} +{"type":"auto_retry_start","attempt":3,"maxAttempts":3,"errorMessage":"Connection error."} +{"type":"auto_retry_end","success":false,"attempt":3,"finalError":"Connection error."} +{"type":"agent_settled"} +` + +func TestScanTraceText_UI11ProviderTransient(t *testing.T) { + got := ScanTraceText(ui11SettledFailureTrace) + if !got.ProviderTransient { + t.Fatal("terminated + Connection error + auto_retry_end success=false must be provider-transient") + } + if !got.Settled { + t.Fatal("agent_settled must be recorded") + } + if !got.RetryExhausted { + t.Fatal("auto_retry_end success=false must mark SDK retry exhausted") + } +} + +func TestScanTraceText_IgnoresWriteArguments(t *testing.T) { + // A truncated write toolCall must never become a deliverable signal. + line := `{"type":"message_end","message":{"content":[{"type":"toolCall","name":"write","arguments":{"path":"/app/data/workspaces/x/tasks/01_draft/output.md","content":"# 伪造正文不应被采用\n"}}]}}` + got := ScanTraceText(line) + if got.HasUsableDeliverable { + t.Fatal("write arguments in a trace line are not a usable deliverable") + } + if got.ProviderTransient { + t.Fatal("a write toolCall alone is not a provider transient") + } +} + +func TestAssessStepFailure_TransientVsPermanent(t *testing.T) { + cases := []struct { + name string + sig StepFailureSignals + want FailureClass + }{ + { + name: "runner retryable flag", + sig: StepFailureSignals{Result: &Result{Success: false, ErrorReason: "Connection error.", Retryable: true}}, + want: FailureTransient, + }, + { + name: "connection error text", + sig: StepFailureSignals{Result: &Result{Success: false, ErrorReason: "Connection error."}}, + want: FailureTransient, + }, + { + name: "terminated text", + sig: StepFailureSignals{Result: &Result{Success: false, ErrorReason: "terminated"}}, + want: FailureTransient, + }, + { + name: "missing after UI-11 settled failure", + sig: StepFailureSignals{ + Result: &Result{Success: false, ErrorReason: "missing target deliverable output.md"}, + ExtraText: ui11SettledFailureTrace, + TargetMissing: true, + }, + want: FailureTransient, + }, + { + name: "missing without provider signal is permanent", + sig: StepFailureSignals{ + Result: &Result{Success: false, ErrorReason: "missing target deliverable output.md"}, + ExtraText: `{"type":"agent_settled"}`, + TargetMissing: true, + }, + want: FailurePermanent, + }, + { + name: "prompt_empty is permanent", + sig: StepFailureSignals{Result: &Result{Success: false, ErrorReason: "prompt_empty"}}, + want: FailurePermanent, + }, + { + name: "unauthorized is permanent", + sig: StepFailureSignals{Result: &Result{Success: false, ErrorReason: "unauthorized"}}, + want: FailurePermanent, + }, + { + name: "governance abort is permanent", + sig: StepFailureSignals{Err: errors.New("治理强制终止: 原地打转")}, + want: FailurePermanent, + }, + { + name: "remote runner unreachable is transient", + sig: StepFailureSignals{Result: &Result{Success: false, ErrorReason: "pi remote runner call failed: connection refused"}}, + want: FailureTransient, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got := AssessStepFailure(tc.sig) + if got.Class != tc.want { + t.Fatalf("class=%s reason=%s want=%s", got.Class, got.Reason, tc.want) + } + }) + } +} + +func TestAssessStepFailure_ReadsWorkDirExecutionLog(t *testing.T) { + dir := t.TempDir() + log := "=== PI-AGENT TASK EXECUTION LOG ===\nOutcome: SUCCESS\nError: None\n" + ui11SettledFailureTrace + if err := os.WriteFile(filepath.Join(dir, "execution.log"), []byte(log), 0o644); err != nil { + t.Fatal(err) + } + got := AssessStepFailure(StepFailureSignals{ + WorkDir: dir, + TargetMissing: true, + Result: &Result{Success: true, ErrorReason: "missing target deliverable output.md"}, + }) + if got.Class != FailureTransient { + t.Fatalf("old runner Outcome:SUCCESS + settled failure must be transient, got %s %s", got.Class, got.Reason) + } + if strings.Contains(got.Reason, "伪造") || strings.Contains(got.Reason, "系统初步设计") { + t.Fatalf("assessment must not carry write-argument body: %s", got.Reason) + } +} + +func TestMaxTransientStepInvocations_IsBounded(t *testing.T) { + if MaxTransientStepInvocations < 2 { + t.Fatal("need at least one fresh re-invocation after the first") + } + if MaxTransientStepInvocations > 3 { + t.Fatalf("bound too loose: %d would retry permanent-looking gaps too many times", MaxTransientStepInvocations) + } +}