From 8b89876022b826e76bcc21b35b8036a72339229c Mon Sep 17 00:00:00 2001 From: kakiuwang-ui Date: Wed, 22 Jul 2026 19:01:42 +0800 Subject: [PATCH 1/7] fix(agent): surface kimi provider.api_error 400 as task failure instead of silent completion (Fixes #5760) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a kimi ACP session is resumed and the upstream LLM rejects the rebuilt history with a 400 ("the message at position N with role 'assistant' must not be empty"), kimi logs the error to stderr and reports session/prompt as stopReason=end_turn. The acpProviderErrorSniffer did not match this line format — no emoji prefix, lowercase "error:", no known keyword — so promoteACPResultOnProviderError found no provider error and left the task status as "completed" with empty output and no error visible anywhere (MUL-5152, #5760). Three regex changes in hermes.go (shared by kimi/hermes/grok/kiro): 1. acpErrorHeaderRe: add `provider.api_error` as an additional match alternative so the kimi-style line is captured into the sniffer buffer. 2. acpErrorDetailRe: add `provider.api_error: NNN` as a detail-extraction prefix so the human-readable message after the status code is surfaced (e.g. "the message at position 43 with role 'assistant' must not be empty") rather than the raw header line. 3. acpTerminalErrorRe: add `provider.api_error: 4` so any 4xx kimi API error is classified as terminal. 4xx responses from the LLM API are client errors and cannot be resolved by retrying the same request. 5xx responses are captured but not terminal, so they only promote to failed when output is also empty (existing fallback behaviour). Two new tests: - TestACPProviderErrorSnifferKimiApiError: the exact #5760 stderr line must be classified as terminal and prompt failed with the error detail. - TestACPProviderErrorSnifferKimiApiError5xx: a 5xx line is captured but not terminal. --- server/pkg/agent/hermes.go | 18 +++++++--- server/pkg/agent/hermes_test.go | 60 +++++++++++++++++++++++++++++++++ 2 files changed, 73 insertions(+), 5 deletions(-) diff --git a/server/pkg/agent/hermes.go b/server/pkg/agent/hermes.go index c6658e459b..9d5ffe9aa4 100644 --- a/server/pkg/agent/hermes.go +++ b/server/pkg/agent/hermes.go @@ -2203,12 +2203,18 @@ type acpProviderErrorSniffer struct { // acpErrorHeaderRe matches the first line of an API-error block. // ACP agents typically prefix these with ⚠️ / ❌ and include an HTTP // status code or a non-retryable-error tag. -var acpErrorHeaderRe = regexp.MustCompile(`(?:⚠️|❌|\[ERROR\]).*(?:BadRequestError|AuthenticationError|RateLimitError|HTTP [0-9]{3}|Non-retryable|API call failed)`) +// The trailing alternative covers provider.api_error lines that kimi emits +// without an emoji prefix, e.g.: +// +// error: failed to run prompt: provider.api_error: 400 the message at position … +var acpErrorHeaderRe = regexp.MustCompile( + `(?:⚠️|❌|\[ERROR\]).*(?:BadRequestError|AuthenticationError|RateLimitError|HTTP [0-9]{3}|Non-retryable|API call failed)` + + `|provider\.api_error`) // acpErrorDetailRe pulls the most useful single-line messages out of -// the subsequent lines of the error block (the one whose "Error:" or -// "Details:" tag actually spells out what happened). -var acpErrorDetailRe = regexp.MustCompile(`(?:Error:|detail:|Details:)\s*(.+)`) +// the error block (the line whose "Error:" / "Details:" tag or whose +// "provider.api_error: NNN " prefix spells out what happened). +var acpErrorDetailRe = regexp.MustCompile(`(?:Error:|detail:|Details:|provider\.api_error: [0-9]+)\s+(.+)`) // acpTerminalErrorRe matches markers that only appear when the // adapter has *given up* on the upstream call — either after @@ -2216,7 +2222,9 @@ var acpErrorDetailRe = regexp.MustCompile(`(?:Error:|detail:|Details:)\s*(.+)`) // classified as non-retryable up front (Non-retryable, BadRequest / // Authentication errors, ❌ / [ERROR] log levels). Per-attempt // warnings ("(attempt 1/3)") deliberately do NOT match this pattern. -var acpTerminalErrorRe = regexp.MustCompile(`(?:❌|\[ERROR\]|after \d+ retr|Non-retryable|BadRequestError|AuthenticationError)`) +// "provider.api_error: 4" covers kimi-style 4xx responses which are always +// client errors and cannot be resolved by retrying the same request. +var acpTerminalErrorRe = regexp.MustCompile(`(?:❌|\[ERROR\]|after \d+ retr|Non-retryable|BadRequestError|AuthenticationError|provider\.api_error: 4)`) // acpAgentOutputTerminalRe matches the synthetic agent-text turn that // hermes-style ACP adapters inject when they exhaust retries against diff --git a/server/pkg/agent/hermes_test.go b/server/pkg/agent/hermes_test.go index d8b3769953..909fa6d64e 100644 --- a/server/pkg/agent/hermes_test.go +++ b/server/pkg/agent/hermes_test.go @@ -2239,6 +2239,66 @@ func TestHermesProviderErrorSnifferTerminalNonRetryable(t *testing.T) { } } +// TestACPProviderErrorSnifferKimiApiError covers the kimi-specific error +// format that was previously invisible to the sniffer, causing tasks to +// silently complete instead of failing (GitHub multica#5760): +// +// error: failed to run prompt: provider.api_error: 400 the message at +// position 43 with role 'assistant' must not be empty +// +// The line has no emoji prefix and uses lowercase "error:", so the original +// acpErrorHeaderRe / acpErrorDetailRe did not capture it. The fix adds +// provider.api_error as an additional header match, recognises 4xx codes as +// terminal, and extracts the error message after "provider.api_error: NNN ". +func TestACPProviderErrorSnifferKimiApiError(t *testing.T) { + t.Parallel() + + stderr := "error: failed to run prompt: provider.api_error: 400 the message at position 43 with role 'assistant' must not be empty\n" + s := newACPProviderErrorSniffer("kimi") + if _, err := s.Write([]byte(stderr)); err != nil { + t.Fatalf("Write: %v", err) + } + + msg := s.terminalMessage() + if msg == "" { + t.Fatal("expected a non-empty terminal message; sniffer did not recognise provider.api_error line") + } + if !strings.Contains(msg, "must not be empty") { + t.Errorf("expected error detail about empty assistant message, got %q", msg) + } + if !strings.Contains(msg, "kimi") { + t.Errorf("expected provider prefix 'kimi' in message, got %q", msg) + } + + // promoteACPResultOnProviderError must flip completed→failed so resumed + // sessions with a permanently broken history surface as task failures. + finalStatus, finalError := promoteACPResultOnProviderError("completed", "", "", s) + if finalStatus != "failed" { + t.Errorf("status = %q, want %q", finalStatus, "failed") + } + if !strings.Contains(finalError, "must not be empty") { + t.Errorf("error = %q, want it to mention the empty assistant message", finalError) + } +} + +// TestACPProviderErrorSnifferKimiApiError5xx verifies that a 5xx +// provider.api_error (potentially transient) is captured but not classified +// as terminal, so it only promotes to failed when output is also empty. +func TestACPProviderErrorSnifferKimiApiError5xx(t *testing.T) { + t.Parallel() + + stderr := "error: provider.api_error: 503 upstream temporarily unavailable\n" + s := newACPProviderErrorSniffer("kimi") + s.Write([]byte(stderr)) + + if s.terminalMessage() != "" { + t.Error("5xx provider.api_error should not be classified as terminal") + } + if s.message() == "" { + t.Error("5xx provider.api_error should still be captured as a non-terminal error") + } +} + // TestHermesBackendPromotesProviderErrorWithNonEmptyOutput pins the // fix for GitHub multica#1952: a hermes run that hits a 429 (or any // upstream provider error) must surface as Status=failed even though From 5adcd38187405072cad1b8faad92fcfef37e1947 Mon Sep 17 00:00:00 2001 From: kakiuwang-ui Date: Thu, 23 Jul 2026 15:46:33 +0800 Subject: [PATCH 2/7] fix(agent): narrow acpTerminalErrorRe to exact 4xx codes; add ResumeRejected for poisoned history MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three fixes addressing multica-eve review on #5785: 1. acpErrorDetailRe: restore \s* (was \s+ in the PR, breaking "detail:value" no-space format in pre-existing branches) 2. acpTerminalErrorRe: replace the prefix match "provider.api_error: 4" with an exact set "(?:400|401|403)". 429 (rate-limit) and 408 (timeout) are now excluded — the kimi adapter retries those internally, and a run that ultimately succeeds must stay status=completed. 3. isPoisonedHistory + ResumeRejected: add isPoisonedHistory() to acpProviderErrorSniffer to detect the kimi "400 + role 'assistant' + must not be empty" pattern. After promoteACPResultOnProviderError, set resumeRejected=true when the history is poisoned so the daemon drops the broken session and retries fresh via the tools==0 gate. Tests added: - TestACPProviderErrorSnifferKimiRateLimitNotTerminal: 429 not terminal, completed status preserved when output is non-empty ("429 after success still completed" case requested in review) - TestACPProviderErrorSnifferPoisonedHistory: isPoisonedHistory() true for the exact pattern, false for plain 400 with different message - TestHermesBackendPoisonedHistorySetsResumeRejected: end-to-end backend test with fake kimi ACP script; asserts Status=failed + ResumeRejected=true on poisoned-history resume --- server/pkg/agent/hermes.go | 43 ++++++++++- server/pkg/agent/hermes_test.go | 125 ++++++++++++++++++++++++++++++++ 2 files changed, 164 insertions(+), 4 deletions(-) diff --git a/server/pkg/agent/hermes.go b/server/pkg/agent/hermes.go index 9d5ffe9aa4..92d96257dd 100644 --- a/server/pkg/agent/hermes.go +++ b/server/pkg/agent/hermes.go @@ -599,6 +599,13 @@ func (b *hermesBackend) Execute(ctx context.Context, prompt string, opts ExecOpt // "API call failed after N retries..." turn the adapter // injects on give-up, or there's no output to fall back on. finalStatus, finalError = promoteACPResultOnProviderError(finalStatus, finalError, finalOutput, providerErr) + // A poisoned session history (400 "assistant must not be empty") + // will fail on every resume. Signal the daemon to drop the old + // session so it can retry fresh via the tools==0 gate instead of + // re-submitting the broken transcript indefinitely. + if finalStatus == "failed" && providerErr.isPoisonedHistory() { + resumeRejected = true + } // Build usage map. c.usageMu.Lock() @@ -2214,7 +2221,9 @@ var acpErrorHeaderRe = regexp.MustCompile( // acpErrorDetailRe pulls the most useful single-line messages out of // the error block (the line whose "Error:" / "Details:" tag or whose // "provider.api_error: NNN " prefix spells out what happened). -var acpErrorDetailRe = regexp.MustCompile(`(?:Error:|detail:|Details:|provider\.api_error: [0-9]+)\s+(.+)`) +// The existing branches use \s* so that "detail:value" (no space) is +// captured as well as "Error: message" (with space). +var acpErrorDetailRe = regexp.MustCompile(`(?:Error:|detail:|Details:|provider\.api_error: [0-9]+)\s*(.+)`) // acpTerminalErrorRe matches markers that only appear when the // adapter has *given up* on the upstream call — either after @@ -2222,9 +2231,12 @@ var acpErrorDetailRe = regexp.MustCompile(`(?:Error:|detail:|Details:|provider\. // classified as non-retryable up front (Non-retryable, BadRequest / // Authentication errors, ❌ / [ERROR] log levels). Per-attempt // warnings ("(attempt 1/3)") deliberately do NOT match this pattern. -// "provider.api_error: 4" covers kimi-style 4xx responses which are always -// client errors and cannot be resolved by retrying the same request. -var acpTerminalErrorRe = regexp.MustCompile(`(?:❌|\[ERROR\]|after \d+ retr|Non-retryable|BadRequestError|AuthenticationError|provider\.api_error: 4)`) +// "provider.api_error: (?:400|401|403)" covers kimi-style client +// errors that are never resolved by retrying the same request. +// 429 (rate-limit) and 408 (timeout) are intentionally excluded: +// the kimi adapter retries those internally, and a run that +// eventually succeeds must stay status=completed. +var acpTerminalErrorRe = regexp.MustCompile(`(?:❌|\[ERROR\]|after \d+ retr|Non-retryable|BadRequestError|AuthenticationError|provider\.api_error: (?:400|401|403))`) // acpAgentOutputTerminalRe matches the synthetic agent-text turn that // hermes-style ACP adapters inject when they exhaust retries against @@ -2423,6 +2435,29 @@ func (s *acpProviderErrorSniffer) terminalMessage() string { return s.messageLocked() } +// isPoisonedHistory reports whether the terminal error is the +// kimi "assistant message must not be empty" pattern — a sign that +// the session history is permanently corrupted and resuming it will +// keep failing with the same 400. When true, the caller should set +// ResumeRejected=true on the Result so the daemon starts a fresh +// session next time rather than re-submitting the broken history. +func (s *acpProviderErrorSniffer) isPoisonedHistory() bool { + s.mu.Lock() + defer s.mu.Unlock() + + if !s.terminal { + return false + } + for _, line := range s.lines { + if strings.Contains(line, "provider.api_error: 400") && + strings.Contains(line, "role") && + strings.Contains(line, "must not be empty") { + return true + } + } + return false +} + // messageLocked is the lock-held implementation shared by message() // and terminalMessage(). Caller must hold s.mu. func (s *acpProviderErrorSniffer) messageLocked() string { diff --git a/server/pkg/agent/hermes_test.go b/server/pkg/agent/hermes_test.go index 909fa6d64e..a091b9f378 100644 --- a/server/pkg/agent/hermes_test.go +++ b/server/pkg/agent/hermes_test.go @@ -2299,6 +2299,56 @@ func TestACPProviderErrorSnifferKimiApiError5xx(t *testing.T) { } } +// TestACPProviderErrorSnifferKimiRateLimitNotTerminal verifies that a +// kimi-style 429 is captured for diagnostic purposes but is NOT marked +// as terminal. The kimi adapter retries rate-limit errors internally; +// a run that ultimately succeeds must stay status=completed regardless +// of how many 429 warnings appeared on stderr during retries. +func TestACPProviderErrorSnifferKimiRateLimitNotTerminal(t *testing.T) { + t.Parallel() + + s := newACPProviderErrorSniffer("kimi") + s.Write([]byte("error: failed to run prompt: provider.api_error: 429 rate limit exceeded\n")) + + if msg := s.terminalMessage(); msg != "" { + t.Errorf("429 should not be terminal, got %q", msg) + } + if msg := s.message(); msg == "" { + t.Error("429 should still be captured for diagnostics (message() must be non-empty)") + } + + // promoteACPResultOnProviderError must leave status=completed when the + // adapter ultimately produced a real answer after internal retries. + finalStatus, _ := promoteACPResultOnProviderError("completed", "", "here is the answer", s) + if finalStatus != "completed" { + t.Errorf("status = %q, want completed: 429 + non-empty output must not be promoted to failed", finalStatus) + } +} + +// TestACPProviderErrorSnifferPoisonedHistory verifies that the specific +// kimi "assistant message must not be empty" (400) pattern is detected +// as a poisoned session so the backend can set ResumeRejected=true and +// the daemon will drop the broken history and start a fresh session. +func TestACPProviderErrorSnifferPoisonedHistory(t *testing.T) { + t.Parallel() + + // Exact pattern from the field: 400 + role + must not be empty. + s := newACPProviderErrorSniffer("kimi") + s.Write([]byte("error: failed to run prompt: provider.api_error: 400 the message at position 43 with role 'assistant' must not be empty\n")) + + if !s.isPoisonedHistory() { + t.Fatal("expected isPoisonedHistory()=true for 400 assistant-empty error") + } + + // A plain 400 with a different error message is terminal but not a + // poisoned history — the session may still be healthy. + s2 := newACPProviderErrorSniffer("kimi") + s2.Write([]byte("error: failed to run prompt: provider.api_error: 400 invalid model specified\n")) + if s2.isPoisonedHistory() { + t.Error("plain 400 without role/must-not-be-empty should not be classified as poisoned history") + } +} + // TestHermesBackendPromotesProviderErrorWithNonEmptyOutput pins the // fix for GitHub multica#1952: a hermes run that hits a 429 (or any // upstream provider error) must surface as Status=failed even though @@ -2653,6 +2703,81 @@ func TestHermesBackendDoesNotPromoteOnTransientRetry(t *testing.T) { } } +// fakeKimiACPPoisonedHistoryScript mimics a kimi adapter that emits the +// "assistant message must not be empty" 400 error when asked to resume +// a session whose history is corrupted. +func fakeKimiACPPoisonedHistoryScript() string { + return `#!/bin/sh +while IFS= read -r line; do + id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9]*\).*/\1/p') + case "$line" in + *'"method":"initialize"'*) + printf '{"jsonrpc":"2.0","id":%s,"result":{"protocolVersion":1,"agentCapabilities":{}}}\n' "$id" + ;; + *'"method":"session/resume"'*) + printf '{"jsonrpc":"2.0","id":%s,"result":{"sessionId":"ses_poison"}}\n' "$id" + ;; + *'"method":"session/prompt"'*) + printf '%s\n' "error: failed to run prompt: provider.api_error: 400 the message at position 43 with role 'assistant' must not be empty" >&2 + printf '{"jsonrpc":"2.0","id":%s,"result":{"stopReason":"end_turn"}}\n' "$id" + exit 0 + ;; + esac +done +` +} + +// TestHermesBackendPoisonedHistorySetsResumeRejected verifies the fix +// for the kimi "assistant message must not be empty" scenario: when +// the sniffer detects a 400 with the poisoned-history signal the +// backend must set ResumeRejected=true so the daemon drops the broken +// session and tries fresh (via the tools==0 gate) instead of +// resuming the same corrupt history on every subsequent task. +func TestHermesBackendPoisonedHistorySetsResumeRejected(t *testing.T) { + t.Parallel() + + fakePath := filepath.Join(t.TempDir(), "hermes") + writeTestExecutable(t, fakePath, []byte(fakeKimiACPPoisonedHistoryScript())) + + backend, err := New("hermes", Config{ExecutablePath: fakePath, Logger: slog.Default()}) + if err != nil { + t.Fatalf("new hermes backend: %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + session, err := backend.Execute(ctx, "prompt-ignored", ExecOptions{ + Timeout: 5 * time.Second, + ResumeSessionID: "ses_poison", + }) + if err != nil { + t.Fatalf("execute: %v", err) + } + go func() { + for range session.Messages { + } + }() + + select { + case result, ok := <-session.Result: + if !ok { + t.Fatal("result channel closed without a value") + } + if result.Status != "failed" { + t.Fatalf("expected status=failed for poisoned history, got %q", result.Status) + } + if !result.ResumeRejected { + t.Error("expected ResumeRejected=true so the daemon clears the broken session") + } + if !strings.Contains(result.Error, "must not be empty") { + t.Errorf("expected error to mention the poisoned history detail, got %q", result.Error) + } + case <-time.After(10 * time.Second): + t.Fatal("timeout waiting for result") + } +} + // ── extractACPMcpCapabilities ── func TestExtractACPMcpCapabilities(t *testing.T) { From 911497df3faf4caf57c9a0402c4fad1ef40a035e Mon Sep 17 00:00:00 2001 From: kakiuwang-ui Date: Fri, 24 Jul 2026 10:00:20 +0800 Subject: [PATCH 3/7] =?UTF-8?q?fix(agent):=20mirror=20isPoisonedHistory?= =?UTF-8?q?=E2=86=92ResumeRejected=20to=20kimi=20backend?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The poisoned-history 400 error (role 'assistant' must not be empty) is emitted by the kimi adapter, so kimi.go must also set resumeRejected=true after promoteACPResultOnProviderError when isPoisonedHistory() fires. The previous commit only wired this in hermes.go. --- server/pkg/agent/kimi.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/server/pkg/agent/kimi.go b/server/pkg/agent/kimi.go index 821935fb50..c823a67f54 100644 --- a/server/pkg/agent/kimi.go +++ b/server/pkg/agent/kimi.go @@ -377,6 +377,9 @@ func (b *kimiBackend) Execute(ctx context.Context, prompt string, opts ExecOptio // per-attempt warnings followed by a successful retry stay // "completed". finalStatus, finalError = promoteACPResultOnProviderError(finalStatus, finalError, finalOutput, providerErr) + if finalStatus == "failed" && providerErr.isPoisonedHistory() { + resumeRejected = true + } c.usageMu.Lock() u := c.usage From 046c3b513dfb5904d85d4f64f04e159c1e50d28e Mon Sep 17 00:00:00 2001 From: kakiuwang-ui Date: Fri, 24 Jul 2026 14:01:03 +0800 Subject: [PATCH 4/7] fix(agent): tighten poisoned-session detection and extend classifyPoisonedError to Kimi format MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three issues addressed from review: 1. isPoisonedHistory: tighten "role" → "role 'assistant'" to avoid matching role 'user' or other role values that cannot cause this specific error. 2. ResumeRejected guard: add opts.ResumeSessionID != "" in both kimi.go and hermes.go so a fresh run (no resume) that encounters the same error fingerprint is not incorrectly flagged. 3. classifyPoisonedError: add Kimi-specific pattern. messageLocked strips the "provider.api_error: 400" prefix, leaving "kimi provider error: … role 'assistant' must not be empty". Without this classifier the task was saved with a non-api_invalid_request failure_reason, so GetLastTaskSession did not exclude it and subsequent tasks could still resume the poisoned session. Tests added: - TestKimiBackendPoisonedHistorySetsResumeRejected: true kimi backend path - TestKimiBackendPoisonedHistoryFreshRunNoResumeRejected: fresh run stays false - TestClassifyPoisonedError: Kimi stripped format and false-positive guards - TestACPProviderErrorSnifferPoisonedHistory: role 'user' must not match --- server/internal/daemon/poisoned.go | 37 ++++-- server/internal/daemon/poisoned_test.go | 25 ++++ server/pkg/agent/hermes.go | 6 +- server/pkg/agent/hermes_test.go | 10 +- server/pkg/agent/kimi.go | 4 +- server/pkg/agent/kimi_test.go | 144 ++++++++++++++++++++++++ 6 files changed, 211 insertions(+), 15 deletions(-) diff --git a/server/internal/daemon/poisoned.go b/server/internal/daemon/poisoned.go index d168bf6450..180f3920ac 100644 --- a/server/internal/daemon/poisoned.go +++ b/server/internal/daemon/poisoned.go @@ -93,27 +93,42 @@ func classifyPoisonedOutput(output string) (string, bool) { // next task on the issue starts a fresh session instead of permanently // inheriting the bad state. // -// Match shape: the Claude Code SDK and similar backends surface upstream -// API failures verbatim, e.g. +// Two recognised shapes: // -// API Error: 400 {"type":"error","error":{"type":"invalid_request_error","message":"Could not process image"},"request_id":"..."} +// 1. Anthropic/Claude style — Claude Code surfaces upstream API failures +// verbatim with the numeric status code and error type, e.g. // -// Matching on both "400" and "invalid_request_error" keeps the classifier -// narrow: 429 rate-limits, 5xx overloads, and tool-shaped errors are -// transient and SHOULD resume on retry. +// API Error: 400 {"type":"error","error":{"type":"invalid_request_error","message":"Could not process image"},...} +// +// Matching on both "400" and "invalid_request_error" keeps it narrow: +// 429 rate-limits, 5xx overloads, and tool-shaped errors are transient. +// +// 2. Kimi style — the Kimi ACP adapter strips the numeric status code before +// the error surfaces (acpProviderErrorSniffer.messageLocked), leaving: +// +// kimi provider error: the message at position N with role 'assistant' must not be empty +// +// This "assistant message must not be empty" pattern indicates the same +// permanently poisoned history: resuming replays the same bad message in +// the same position. Matching on both "role 'assistant'" and "must not be +// empty" keeps it specific to this API contract. func classifyPoisonedError(errMsg string) (string, bool) { if errMsg == "" { return "", false } lowered := strings.ToLower(errMsg) - // Both markers must be present: "400" alone is too generic (a tool - // could surface a 400 from anywhere) and "invalid_request_error" - // alone could in theory appear in non-poisoning contexts. The - // combination is the canonical Anthropic error shape and indicates - // the request body — i.e. the conversation history — is the problem. + // Anthropic/Claude: "400" + "invalid_request_error" is the canonical + // poisoned-conversation fingerprint. Both must appear; "400" alone is too + // generic (a tool could surface it), and "invalid_request_error" alone + // could theoretically appear in benign contexts. if strings.Contains(lowered, "invalid_request_error") && strings.Contains(lowered, "400") { return FailureReasonAPIInvalidRequest, true } + // Kimi: the status code is stripped by the sniffer's messageLocked path, + // so match on the precise error text instead. + if strings.Contains(lowered, "role 'assistant'") && strings.Contains(lowered, "must not be empty") { + return FailureReasonAPIInvalidRequest, true + } return "", false } diff --git a/server/internal/daemon/poisoned_test.go b/server/internal/daemon/poisoned_test.go index 6ce5b7752d..7ca3fdf077 100644 --- a/server/internal/daemon/poisoned_test.go +++ b/server/internal/daemon/poisoned_test.go @@ -157,6 +157,31 @@ func TestClassifyPoisonedError(t *testing.T) { errMsg: "claude execution timeout after 10m", wantOK: false, }, + { + // Kimi poisoned history: acpProviderErrorSniffer.messageLocked strips + // the "provider.api_error: 400" prefix, so the numeric status code is + // absent in the stored error. The "role 'assistant' … must not be + // empty" fingerprint is specific enough to serve as the classifier. + name: "kimi assistant-empty poisoned history (stripped format)", + errMsg: "kimi provider error: the message at position 43 with role 'assistant' must not be empty", + wantOK: true, + wantReason: FailureReasonAPIInvalidRequest, + }, + { + // Variant with a different position number — ensure the classifier + // doesn't hard-code the position. + name: "kimi assistant-empty poisoned history (different position)", + errMsg: "kimi provider error: the message at position 7 with role 'assistant' must not be empty", + wantOK: true, + wantReason: FailureReasonAPIInvalidRequest, + }, + { + // A generic "must not be empty" without "role 'assistant'" must NOT + // be classified — too likely to appear in benign tool errors. + name: "must-not-be-empty without role assistant is not poisoning", + errMsg: "validation error: field must not be empty", + wantOK: false, + }, } for _, tc := range cases { diff --git a/server/pkg/agent/hermes.go b/server/pkg/agent/hermes.go index 92d96257dd..2a39329fdc 100644 --- a/server/pkg/agent/hermes.go +++ b/server/pkg/agent/hermes.go @@ -603,7 +603,9 @@ func (b *hermesBackend) Execute(ctx context.Context, prompt string, opts ExecOpt // will fail on every resume. Signal the daemon to drop the old // session so it can retry fresh via the tools==0 gate instead of // re-submitting the broken transcript indefinitely. - if finalStatus == "failed" && providerErr.isPoisonedHistory() { + // Guard on ResumeSessionID so a fresh run that somehow sees the + // same error fingerprint is not incorrectly flagged. + if finalStatus == "failed" && opts.ResumeSessionID != "" && providerErr.isPoisonedHistory() { resumeRejected = true } @@ -2450,7 +2452,7 @@ func (s *acpProviderErrorSniffer) isPoisonedHistory() bool { } for _, line := range s.lines { if strings.Contains(line, "provider.api_error: 400") && - strings.Contains(line, "role") && + strings.Contains(line, "role 'assistant'") && strings.Contains(line, "must not be empty") { return true } diff --git a/server/pkg/agent/hermes_test.go b/server/pkg/agent/hermes_test.go index a091b9f378..fa0c508a7a 100644 --- a/server/pkg/agent/hermes_test.go +++ b/server/pkg/agent/hermes_test.go @@ -2332,7 +2332,7 @@ func TestACPProviderErrorSnifferKimiRateLimitNotTerminal(t *testing.T) { func TestACPProviderErrorSnifferPoisonedHistory(t *testing.T) { t.Parallel() - // Exact pattern from the field: 400 + role + must not be empty. + // Exact pattern from the field: 400 + role 'assistant' + must not be empty. s := newACPProviderErrorSniffer("kimi") s.Write([]byte("error: failed to run prompt: provider.api_error: 400 the message at position 43 with role 'assistant' must not be empty\n")) @@ -2347,6 +2347,14 @@ func TestACPProviderErrorSnifferPoisonedHistory(t *testing.T) { if s2.isPoisonedHistory() { t.Error("plain 400 without role/must-not-be-empty should not be classified as poisoned history") } + + // A 400 that mentions "role" but not specifically "role 'assistant'" must + // not match — the check was tightened to avoid false positives. + s3 := newACPProviderErrorSniffer("kimi") + s3.Write([]byte("error: failed to run prompt: provider.api_error: 400 the message at position 7 with role 'user' must not be empty\n")) + if s3.isPoisonedHistory() { + t.Error("400 with role 'user' (not 'assistant') should not be classified as poisoned history") + } } // TestHermesBackendPromotesProviderErrorWithNonEmptyOutput pins the diff --git a/server/pkg/agent/kimi.go b/server/pkg/agent/kimi.go index c823a67f54..16f298b9f0 100644 --- a/server/pkg/agent/kimi.go +++ b/server/pkg/agent/kimi.go @@ -377,7 +377,9 @@ func (b *kimiBackend) Execute(ctx context.Context, prompt string, opts ExecOptio // per-attempt warnings followed by a successful retry stay // "completed". finalStatus, finalError = promoteACPResultOnProviderError(finalStatus, finalError, finalOutput, providerErr) - if finalStatus == "failed" && providerErr.isPoisonedHistory() { + // Guard on ResumeSessionID: only a resume can inherit a poisoned + // history; a fresh run cannot reproduce this failure deterministically. + if finalStatus == "failed" && opts.ResumeSessionID != "" && providerErr.isPoisonedHistory() { resumeRejected = true } diff --git a/server/pkg/agent/kimi_test.go b/server/pkg/agent/kimi_test.go index 5734908130..e9c2944d68 100644 --- a/server/pkg/agent/kimi_test.go +++ b/server/pkg/agent/kimi_test.go @@ -335,3 +335,147 @@ func TestKimiResumeIncludesMcpServers(t *testing.T) { t.Fatalf("session/resume.mcpServers: got %v, want one entry named fetch", servers) } } + +// fakeKimiPoisonedHistoryScript mimics the kimi CLI when asked to resume a +// session whose conversation history contains an empty assistant message. The +// adapter emits the "provider.api_error: 400 … role 'assistant' must not be +// empty" line to stderr and returns a completed (end_turn) JSON-RPC response — +// matching the real kimi adapter behaviour that prompted MUL-5154. +func fakeKimiPoisonedHistoryScript() string { + return `#!/bin/sh +while IFS= read -r line; do + id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9]*\).*/\1/p') + case "$line" in + *'"method":"initialize"'*) + printf '{"jsonrpc":"2.0","id":%s,"result":{"protocolVersion":1,"agentCapabilities":{}}}\n' "$id" + ;; + *'"method":"session/resume"'*) + printf '{"jsonrpc":"2.0","id":%s,"result":{"sessionId":"ses_poison"}}\n' "$id" + ;; + *'"method":"session/prompt"'*) + printf '%s\n' "error: failed to run prompt: provider.api_error: 400 the message at position 43 with role 'assistant' must not be empty" >&2 + printf '{"jsonrpc":"2.0","id":%s,"result":{"stopReason":"end_turn"}}\n' "$id" + exit 0 + ;; + esac +done +` +} + +// TestKimiBackendPoisonedHistorySetsResumeRejected verifies that when the kimi +// backend receives the "assistant must not be empty" 400 error while resuming +// a session, it sets ResumeRejected=true so the daemon clears the broken +// session pointer and starts fresh on the next task. +func TestKimiBackendPoisonedHistorySetsResumeRejected(t *testing.T) { + t.Parallel() + + fakePath := filepath.Join(t.TempDir(), "kimi") + writeTestExecutable(t, fakePath, []byte(fakeKimiPoisonedHistoryScript())) + + backend, err := New("kimi", Config{ExecutablePath: fakePath, Logger: slog.Default()}) + if err != nil { + t.Fatalf("new kimi backend: %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + session, err := backend.Execute(ctx, "prompt-ignored", ExecOptions{ + Timeout: 5 * time.Second, + ResumeSessionID: "ses_poison", + }) + if err != nil { + t.Fatalf("execute: %v", err) + } + go func() { + for range session.Messages { + } + }() + + select { + case result, ok := <-session.Result: + if !ok { + t.Fatal("result channel closed without a value") + } + if result.Status != "failed" { + t.Fatalf("expected status=failed for poisoned history, got %q", result.Status) + } + if !result.ResumeRejected { + t.Error("expected ResumeRejected=true so the daemon clears the broken session") + } + if !strings.Contains(result.Error, "must not be empty") { + t.Errorf("expected error to mention the poisoned-history detail, got %q", result.Error) + } + case <-time.After(10 * time.Second): + t.Fatal("timeout waiting for result") + } +} + +// TestKimiBackendPoisonedHistoryFreshRunNoResumeRejected verifies that when +// the same poisoned-history error occurs on a FRESH run (ResumeSessionID==""), +// ResumeRejected must stay false. A fresh run cannot inherit a broken history, +// so flagging it would incorrectly prevent the next task from resuming a good +// session. +func TestKimiBackendPoisonedHistoryFreshRunNoResumeRejected(t *testing.T) { + t.Parallel() + + // Re-use the same fake script; the difference is no ResumeSessionID in opts. + // The script only handles session/resume, so without it the session/new path + // runs but the fake exits early — that's fine, we only care about the flag. + fakePath := filepath.Join(t.TempDir(), "kimi") + // Use a script that emits the poisoned error without needing a resume request. + script := `#!/bin/sh +while IFS= read -r line; do + id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9]*\).*/\1/p') + case "$line" in + *'"method":"initialize"'*) + printf '{"jsonrpc":"2.0","id":%s,"result":{"protocolVersion":1,"agentCapabilities":{}}}\n' "$id" + ;; + *'"method":"session/new"'*) + printf '{"jsonrpc":"2.0","id":%s,"result":{"sessionId":"ses_fresh"}}\n' "$id" + ;; + *'"method":"session/prompt"'*) + printf '%s\n' "error: failed to run prompt: provider.api_error: 400 the message at position 1 with role 'assistant' must not be empty" >&2 + printf '{"jsonrpc":"2.0","id":%s,"result":{"stopReason":"end_turn"}}\n' "$id" + exit 0 + ;; + esac +done +` + writeTestExecutable(t, fakePath, []byte(script)) + + backend, err := New("kimi", Config{ExecutablePath: fakePath, Logger: slog.Default()}) + if err != nil { + t.Fatalf("new kimi backend: %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + // No ResumeSessionID — fresh run. + session, err := backend.Execute(ctx, "prompt-ignored", ExecOptions{ + Timeout: 5 * time.Second, + }) + if err != nil { + t.Fatalf("execute: %v", err) + } + go func() { + for range session.Messages { + } + }() + + select { + case result, ok := <-session.Result: + if !ok { + t.Fatal("result channel closed without a value") + } + if result.Status != "failed" { + t.Fatalf("expected status=failed, got %q", result.Status) + } + if result.ResumeRejected { + t.Error("ResumeRejected must be false on a fresh run (no ResumeSessionID)") + } + case <-time.After(10 * time.Second): + t.Fatal("timeout waiting for result") + } +} From 34eec5455a8d4119c3dfab808d47f0cddef3ecc4 Mon Sep 17 00:00:00 2001 From: kakiuwang-ui Date: Sat, 25 Jul 2026 01:02:17 +0800 Subject: [PATCH 5/7] fix(daemon): drop prior session when agent is renamed to prevent identity contamination (MUL-5736) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When an agent is renamed (e.g. "Reviewer" → "Backend Builder"), the workdir from its last run still holds a CLAUDE.md whose ## Agent Identity section names the old role. Resuming that session means the conversation history has already established the old identity, causing the agent to misidentify itself and refuse tasks — the behaviour reported in MUL-5736. Fix: before overwriting CLAUDE.md with the new identity, read the name already recorded in the managed marker block. If it differs from the current agent name, drop PriorSessionID and start a fresh conversation. The check runs after gateCodexResumeToRolloutPresence while the old file is still on disk. The drop does not set PriorSessionResumeUnavailable because a rename is intentional: the user changed the agent's identity on purpose, so a fresh session is expected behaviour rather than an unexpected loss to surface. ReadPriorAgentName is added to execenv so the parsing logic is testable in isolation; gateResumeToSameAgentName wraps the daemon-level gate. --- server/internal/daemon/daemon.go | 30 +++++++ server/internal/daemon/daemon_test.go | 88 +++++++++++++++++++ .../internal/daemon/execenv/runtime_config.go | 33 +++++++ .../daemon/execenv/runtime_config_test.go | 82 +++++++++++++++++ 4 files changed, 233 insertions(+) diff --git a/server/internal/daemon/daemon.go b/server/internal/daemon/daemon.go index 4eaa8fc331..3e5eb1c0ff 100644 --- a/server/internal/daemon/daemon.go +++ b/server/internal/daemon/daemon.go @@ -3740,6 +3740,33 @@ func gateResumeToReusedWorkdir(task *Task, taskCtx *execenv.TaskContextForEnv, e return reused } +// gateResumeToSameAgentName drops the prior session when the reused workdir's +// existing runtime config records a different agent name than the current task. +// This prevents a renamed agent from resuming a session whose conversation +// history was established under the old identity — which can cause the agent to +// misidentify itself even though CLAUDE.md is updated before launch (MUL-5736). +// No-op when there is nothing to resume, no current agent name, or when the +// prior name cannot be read (e.g. fresh workdir, legacy config without the +// identity section). Unlike gateResumeToReusedWorkdir, this does not set +// PriorSessionResumeUnavailable: the rename is intentional, so starting fresh +// is expected behaviour rather than a loss the agent must disclose. +func gateResumeToSameAgentName(task *Task, taskCtx *execenv.TaskContextForEnv, provider, workDir string, taskLog *slog.Logger) { + if task.PriorSessionID == "" || taskCtx.AgentName == "" { + return + } + priorName := execenv.ReadPriorAgentName(workDir, provider) + if priorName == "" || priorName == taskCtx.AgentName { + return + } + taskLog.Warn("dropping prior session: agent identity changed since last run", + "session_id", task.PriorSessionID, + "prior_name", priorName, + "current_name", taskCtx.AgentName, + ) + task.PriorSessionID = "" + taskCtx.PriorSessionResumed = false +} + // shouldReusePriorWorkdir keeps the local_directory lock invariant without // forcing every squad-leader follow-up onto a fresh provider session. Worker // tasks already expose their current local-directory assignment, so their @@ -4398,6 +4425,9 @@ func (d *Daemon) runTask(ctx context.Context, task Task, provider string, slot i // conversation Codex will silently restart from scratch. if reused { gateCodexResumeToRolloutPresence(&task, &taskCtx, provider, env.CodexHome, taskLog) + // Check for agent rename BEFORE overwriting CLAUDE.md — we need to + // read the old identity while it is still on disk (MUL-5736). + gateResumeToSameAgentName(&task, &taskCtx, provider, env.WorkDir, taskLog) } // Inject runtime-specific config (meta skill) so the agent discovers .agent_context/. diff --git a/server/internal/daemon/daemon_test.go b/server/internal/daemon/daemon_test.go index 4677dd1939..f6502507e9 100644 --- a/server/internal/daemon/daemon_test.go +++ b/server/internal/daemon/daemon_test.go @@ -1342,6 +1342,94 @@ func TestGateResumeToReusedWorkdir(t *testing.T) { } } +func TestGateResumeToSameAgentName(t *testing.T) { + t.Parallel() + + // writeConfig writes a CLAUDE.md with the given agent name so we can + // exercise name-mismatch detection against a realistic on-disk file. + writeConfig := func(t *testing.T, dir, agentName string) { + t.Helper() + ctx := execenv.TaskContextForEnv{ + IssueID: "00000000-0000-0000-0000-000000000001", + AgentName: agentName, + } + if _, err := execenv.InjectRuntimeConfig(dir, "claude", ctx); err != nil { + t.Fatalf("InjectRuntimeConfig: %v", err) + } + } + + t.Run("same name keeps session", func(t *testing.T) { + t.Parallel() + dir := t.TempDir() + writeConfig(t, dir, "Backend Builder") + task := Task{PriorSessionID: "sess-bb"} + taskCtx := execenv.TaskContextForEnv{AgentName: "Backend Builder", PriorSessionResumed: true} + gateResumeToSameAgentName(&task, &taskCtx, "claude", dir, slog.Default()) + if task.PriorSessionID != "sess-bb" { + t.Fatalf("session should be kept when name matches, got %q", task.PriorSessionID) + } + if !taskCtx.PriorSessionResumed { + t.Fatal("PriorSessionResumed should remain true") + } + }) + + t.Run("renamed agent drops session", func(t *testing.T) { + t.Parallel() + dir := t.TempDir() + // Workdir was last used when agent was named "Reviewer". + writeConfig(t, dir, "Reviewer") + // Agent was renamed to "Backend Builder". + task := Task{PriorSessionID: "sess-reviewer"} + taskCtx := execenv.TaskContextForEnv{AgentName: "Backend Builder", PriorSessionResumed: true} + gateResumeToSameAgentName(&task, &taskCtx, "claude", dir, slog.Default()) + if task.PriorSessionID != "" { + t.Fatalf("session should be dropped on name change, got %q", task.PriorSessionID) + } + if taskCtx.PriorSessionResumed { + t.Fatal("PriorSessionResumed should be cleared") + } + // Name-change drop is intentional, not a lost session — no continuity notice. + if taskCtx.PriorSessionResumeUnavailable { + t.Fatal("PriorSessionResumeUnavailable should not be set on an intentional name-change drop") + } + }) + + t.Run("no prior session is a no-op", func(t *testing.T) { + t.Parallel() + dir := t.TempDir() + writeConfig(t, dir, "Reviewer") + task := Task{PriorSessionID: ""} + taskCtx := execenv.TaskContextForEnv{AgentName: "Backend Builder", PriorSessionResumed: false} + gateResumeToSameAgentName(&task, &taskCtx, "claude", dir, slog.Default()) + if task.PriorSessionID != "" || taskCtx.PriorSessionResumed { + t.Fatal("no-op expected when there is no prior session") + } + }) + + t.Run("no current agent name is a no-op", func(t *testing.T) { + t.Parallel() + dir := t.TempDir() + writeConfig(t, dir, "Reviewer") + task := Task{PriorSessionID: "sess-x"} + taskCtx := execenv.TaskContextForEnv{AgentName: "", PriorSessionResumed: true} + gateResumeToSameAgentName(&task, &taskCtx, "claude", dir, slog.Default()) + if task.PriorSessionID != "sess-x" { + t.Fatalf("session should be kept when current agent name is unknown, got %q", task.PriorSessionID) + } + }) + + t.Run("missing config file keeps session", func(t *testing.T) { + t.Parallel() + dir := t.TempDir() // no CLAUDE.md written + task := Task{PriorSessionID: "sess-x"} + taskCtx := execenv.TaskContextForEnv{AgentName: "Backend Builder", PriorSessionResumed: true} + gateResumeToSameAgentName(&task, &taskCtx, "claude", dir, slog.Default()) + if task.PriorSessionID != "sess-x" { + t.Fatalf("session should be kept when config is missing, got %q", task.PriorSessionID) + } + }) +} + // newCodexStoreGuardDaemon builds a Daemon with only the per-issue Codex session // store guard initialised — enough to exercise the reserve-vs-mark protocol. func newCodexStoreGuardDaemon() *Daemon { diff --git a/server/internal/daemon/execenv/runtime_config.go b/server/internal/daemon/execenv/runtime_config.go index 09b82f7c44..93bfd806d3 100644 --- a/server/internal/daemon/execenv/runtime_config.go +++ b/server/internal/daemon/execenv/runtime_config.go @@ -300,6 +300,39 @@ func locateMarkerBlock(content string) (start, end int, found bool) { return start, end, true } +// ReadPriorAgentName reads the agent name recorded in the Multica-managed +// block of the current runtime config file for the given workDir and provider. +// It returns the agent name from the "## Agent Identity" section, or an empty +// string when the file does not exist, has no managed block, or has no agent +// identity entry. Used by gateResumeToSameAgentName to detect agent renames +// before a session resume is attempted. +func ReadPriorAgentName(workDir, provider string) string { + path := runtimeConfigPath(workDir, provider) + if path == "" { + return "" + } + data, err := os.ReadFile(path) + if err != nil { + return "" + } + start, end, ok := locateMarkerBlock(string(data)) + if !ok { + return "" + } + block := string(data)[start:end] + const prefix = "**You are: " + idx := strings.Index(block, prefix) + if idx < 0 { + return "" + } + rest := block[idx+len(prefix):] + end2 := strings.Index(rest, "**") + if end2 < 0 { + return "" + } + return strings.TrimSpace(rest[:end2]) +} + // CleanupRuntimeConfig excises the Multica marker block from the runtime // config file for the given provider and restores the file to its exact // pre-injection state, byte for byte. The cleanup is the second half of diff --git a/server/internal/daemon/execenv/runtime_config_test.go b/server/internal/daemon/execenv/runtime_config_test.go index 6307dcf4aa..1e2b393357 100644 --- a/server/internal/daemon/execenv/runtime_config_test.go +++ b/server/internal/daemon/execenv/runtime_config_test.go @@ -1585,3 +1585,85 @@ func TestCommentTriggeredBriefSingleThreadKeepsSingleReply(t *testing.T) { t.Errorf("single/same-thread brief must keep the single --parent=trigger cookbook, got:\n%s", out) } } + +func TestReadPriorAgentName(t *testing.T) { + t.Parallel() + + writeWithAgentName := func(t *testing.T, dir, name string) { + t.Helper() + ctx := TaskContextForEnv{ + IssueID: "00000000-0000-0000-0000-000000000001", + AgentName: name, + } + if _, err := InjectRuntimeConfig(dir, "claude", ctx); err != nil { + t.Fatalf("InjectRuntimeConfig: %v", err) + } + } + + t.Run("reads name from injected CLAUDE.md", func(t *testing.T) { + t.Parallel() + dir := t.TempDir() + writeWithAgentName(t, dir, "Backend Builder") + if got := ReadPriorAgentName(dir, "claude"); got != "Backend Builder" { + t.Fatalf("got %q, want %q", got, "Backend Builder") + } + }) + + t.Run("reads name with ID suffix present", func(t *testing.T) { + t.Parallel() + dir := t.TempDir() + ctx := TaskContextForEnv{ + IssueID: "00000000-0000-0000-0000-000000000001", + AgentName: "Reviewer", + AgentID: "de53a53c-3d5e-4829-bf82-10dd35ce4858", + } + if _, err := InjectRuntimeConfig(dir, "claude", ctx); err != nil { + t.Fatalf("InjectRuntimeConfig: %v", err) + } + if got := ReadPriorAgentName(dir, "claude"); got != "Reviewer" { + t.Fatalf("got %q, want %q", got, "Reviewer") + } + }) + + t.Run("no file returns empty", func(t *testing.T) { + t.Parallel() + dir := t.TempDir() + if got := ReadPriorAgentName(dir, "claude"); got != "" { + t.Fatalf("got %q, want empty", got) + } + }) + + t.Run("file without marker block returns empty", func(t *testing.T) { + t.Parallel() + dir := t.TempDir() + if err := os.WriteFile(filepath.Join(dir, "CLAUDE.md"), []byte("# My CLAUDE.md\nSome content.\n"), 0o644); err != nil { + t.Fatal(err) + } + if got := ReadPriorAgentName(dir, "claude"); got != "" { + t.Fatalf("got %q, want empty", got) + } + }) + + t.Run("marker block without agent identity returns empty", func(t *testing.T) { + t.Parallel() + dir := t.TempDir() + ctx := TaskContextForEnv{ + IssueID: "00000000-0000-0000-0000-000000000001", + // AgentName intentionally empty — no identity section will be written + } + if _, err := InjectRuntimeConfig(dir, "claude", ctx); err != nil { + t.Fatalf("InjectRuntimeConfig: %v", err) + } + if got := ReadPriorAgentName(dir, "claude"); got != "" { + t.Fatalf("got %q, want empty (no agent name in config)", got) + } + }) + + t.Run("unknown provider returns empty", func(t *testing.T) { + t.Parallel() + dir := t.TempDir() + if got := ReadPriorAgentName(dir, "unknown-provider"); got != "" { + t.Fatalf("got %q, want empty", got) + } + }) +} From e534cc23b391d609306d6e56e2f58e5406f24b91 Mon Sep 17 00:00:00 2001 From: kakiuwang-ui Date: Sat, 25 Jul 2026 12:13:18 +0800 Subject: [PATCH 6/7] fix(agent): fix two-line kimi stderr poisoned-session detection and classifyPoisonedError Kimi's two-line stderr format emits "provider.api_error: 400" on the error header line and "detail: messages[N].content: content must not be empty" on the next line. Three bugs combined to let poisoned sessions slip through the exclusion chain: 1. messageLocked(): acpErrorDetailRe's `provider\.api_error: [0-9]+` branch was backtracking one digit into `(.+)` when the status code was the last token on the line, capturing "0" as the detail and returning "kimi provider error: provider.api_error: 400 0" before ever reaching the real detail line. Fixed by tracking apiErrLineIdx and skipping pure-digit captures on that line (backtrack artefact). 2. isPoisonedHistory(): required all three markers (provider.api_error: 400, role 'assistant', must not be empty) on the same stderr line. Fixed to scan across all captured lines with separate hasBadRequest and hasEmptyContent flags. 3. classifyPoisonedError(): only had a "role 'assistant' + must not be empty" pattern (Kimi single-line format). Added a "provider.api_error: 400 + must not be empty" pattern so the status prefix forwarded by the fixed messageLocked is recognised and the task row gets failure_reason=api_invalid_request, excluding it from GetLastTaskSession resume lookups. Adds: - TestACPProviderErrorSnifferMessageLockedPreservesAPIErrorStatus - Two-line case in TestACPProviderErrorSnifferPoisonedHistory - TestKimiBackendPoisonedHistoryTwoLineFormat (end-to-end integration) - Three new classifyPoisonedError table cases in poisoned_test.go --- server/internal/daemon/poisoned.go | 30 +++++--- server/internal/daemon/poisoned_test.go | 27 ++++++++ server/pkg/agent/hermes.go | 92 +++++++++++++++++++++---- server/pkg/agent/hermes_test.go | 71 +++++++++++++++---- server/pkg/agent/kimi_test.go | 72 +++++++++++++++++++ 5 files changed, 257 insertions(+), 35 deletions(-) diff --git a/server/internal/daemon/poisoned.go b/server/internal/daemon/poisoned.go index 180f3920ac..9450608386 100644 --- a/server/internal/daemon/poisoned.go +++ b/server/internal/daemon/poisoned.go @@ -93,7 +93,7 @@ func classifyPoisonedOutput(output string) (string, bool) { // next task on the issue starts a fresh session instead of permanently // inheriting the bad state. // -// Two recognised shapes: +// Three recognised shapes: // // 1. Anthropic/Claude style — Claude Code surfaces upstream API failures // verbatim with the numeric status code and error type, e.g. @@ -103,15 +103,22 @@ func classifyPoisonedOutput(output string) (string, bool) { // Matching on both "400" and "invalid_request_error" keeps it narrow: // 429 rate-limits, 5xx overloads, and tool-shaped errors are transient. // -// 2. Kimi style — the Kimi ACP adapter strips the numeric status code before -// the error surfaces (acpProviderErrorSniffer.messageLocked), leaving: +// 2. Kimi style (current) — acpProviderErrorSniffer.messageLocked now +// forwards the provider.api_error status code as a prefix, producing: +// +// kimi provider error: provider.api_error: 400 messages[43].content: content must not be empty +// +// Matching on "provider.api_error: 400" + "must not be empty" is +// specific to the poisoned-history case: a 400 with a different message +// (e.g. invalid model, unknown field) is NOT classified here. +// +// 3. Kimi backward-compat — before messageLocked preserved the status +// prefix, single-line errors that included role information produced: // // kimi provider error: the message at position N with role 'assistant' must not be empty // -// This "assistant message must not be empty" pattern indicates the same -// permanently poisoned history: resuming replays the same bad message in -// the same position. Matching on both "role 'assistant'" and "must not be -// empty" keeps it specific to this API contract. +// Kept so rows written by older daemon versions are still classified +// correctly when the server re-processes them. func classifyPoisonedError(errMsg string) (string, bool) { if errMsg == "" { return "", false @@ -124,8 +131,13 @@ func classifyPoisonedError(errMsg string) (string, bool) { if strings.Contains(lowered, "invalid_request_error") && strings.Contains(lowered, "400") { return FailureReasonAPIInvalidRequest, true } - // Kimi: the status code is stripped by the sniffer's messageLocked path, - // so match on the precise error text instead. + // Kimi (current): messageLocked forwards the status tag so both markers + // appear in the stored error string regardless of the number of stderr lines. + if strings.Contains(lowered, "provider.api_error: 400") && strings.Contains(lowered, "must not be empty") { + return FailureReasonAPIInvalidRequest, true + } + // Kimi (backward compat): single-line format before the status-prefix fix, + // where the 400 was stripped but the role name was preserved in the detail. if strings.Contains(lowered, "role 'assistant'") && strings.Contains(lowered, "must not be empty") { return FailureReasonAPIInvalidRequest, true } diff --git a/server/internal/daemon/poisoned_test.go b/server/internal/daemon/poisoned_test.go index 7ca3fdf077..6d9cd76b78 100644 --- a/server/internal/daemon/poisoned_test.go +++ b/server/internal/daemon/poisoned_test.go @@ -182,6 +182,33 @@ func TestClassifyPoisonedError(t *testing.T) { errMsg: "validation error: field must not be empty", wantOK: false, }, + { + // Kimi two-line format: acpProviderErrorSniffer.messageLocked now + // forwards the "provider.api_error: NNN" status prefix so that + // classifyPoisonedError can detect the 400 fingerprint even when + // the status and the detail arrived on separate stderr lines. + // The stored error looks like: + // kimi provider error: provider.api_error: 400 messages[43].content: content must not be empty + name: "kimi two-line format with status prefix forwarded", + errMsg: "kimi provider error: provider.api_error: 400 messages[43].content: content must not be empty", + wantOK: true, + wantReason: FailureReasonAPIInvalidRequest, + }, + { + // provider.api_error: 400 alone (without a "must not be empty" + // detail) must NOT trigger — a 400 with a different error body + // (invalid model, unknown field, etc.) is not a poisoned history. + name: "provider.api_error: 400 without must-not-be-empty is not poisoning", + errMsg: "kimi provider error: provider.api_error: 400 invalid model specified", + wantOK: false, + }, + { + // A wrong status code (429) combined with "must not be empty" + // must NOT match — rate-limit responses are transient, not poisoning. + name: "provider.api_error: 429 with must-not-be-empty is not poisoning", + errMsg: "kimi provider error: provider.api_error: 429 messages[5].content: content must not be empty", + wantOK: false, + }, } for _, tc := range cases { diff --git a/server/pkg/agent/hermes.go b/server/pkg/agent/hermes.go index 2a39329fdc..35ef3ad21e 100644 --- a/server/pkg/agent/hermes.go +++ b/server/pkg/agent/hermes.go @@ -2240,6 +2240,13 @@ var acpErrorDetailRe = regexp.MustCompile(`(?:Error:|detail:|Details:|provider\. // eventually succeeds must stay status=completed. var acpTerminalErrorRe = regexp.MustCompile(`(?:❌|\[ERROR\]|after \d+ retr|Non-retryable|BadRequestError|AuthenticationError|provider\.api_error: (?:400|401|403))`) +// providerApiErrorStatusRe extracts the numeric status code from a kimi +// "provider.api_error: NNN" error line so messageLocked can forward it +// into the formatted detail string, letting downstream classifiers +// (classifyPoisonedError in daemon/poisoned.go) detect the 400 fingerprint +// even when the human-readable detail text lives on a separate stderr line. +var providerApiErrorStatusRe = regexp.MustCompile(`provider\.api_error: (\d+)`) + // acpAgentOutputTerminalRe matches the synthetic agent-text turn that // hermes-style ACP adapters inject when they exhaust retries against // the upstream LLM ("API call failed after 3 retries: HTTP 429..."), @@ -2437,12 +2444,22 @@ func (s *acpProviderErrorSniffer) terminalMessage() string { return s.messageLocked() } -// isPoisonedHistory reports whether the terminal error is the -// kimi "assistant message must not be empty" pattern — a sign that -// the session history is permanently corrupted and resuming it will -// keep failing with the same 400. When true, the caller should set -// ResumeRejected=true on the Result so the daemon starts a fresh -// session next time rather than re-submitting the broken history. +// isPoisonedHistory reports whether the terminal error indicates a +// permanently poisoned Kimi session history — the provider rejected the +// conversation with a 400 because a message content field is empty. Resuming +// the same session will replay the identical body and reproduce the same 400, +// so the daemon must drop the session pointer and start fresh. +// +// Detection requires two signals that may arrive on separate stderr lines: +// +// - "provider.api_error: 400" — the HTTP 400 from Moonshot AI, signalling +// a bad request that retrying will not fix. +// - "must not be empty" — the error detail indicating which part of the +// conversation history is malformed. +// +// Checking across all captured lines (not on a single line) handles Kimi's +// two-line format where the status appears on the error header line and the +// detail text on the following "detail:" line. func (s *acpProviderErrorSniffer) isPoisonedHistory() bool { s.mu.Lock() defer s.mu.Unlock() @@ -2450,10 +2467,16 @@ func (s *acpProviderErrorSniffer) isPoisonedHistory() bool { if !s.terminal { return false } + var hasBadRequest, hasEmptyContent bool for _, line := range s.lines { - if strings.Contains(line, "provider.api_error: 400") && - strings.Contains(line, "role 'assistant'") && - strings.Contains(line, "must not be empty") { + lowered := strings.ToLower(line) + if strings.Contains(lowered, "provider.api_error: 400") { + hasBadRequest = true + } + if strings.Contains(lowered, "must not be empty") { + hasEmptyContent = true + } + if hasBadRequest && hasEmptyContent { return true } } @@ -2464,12 +2487,41 @@ func (s *acpProviderErrorSniffer) isPoisonedHistory() bool { // and terminalMessage(). Caller must hold s.mu. func (s *acpProviderErrorSniffer) messageLocked() string { prefix := s.provider + " provider error: " - for _, line := range s.lines { + + // When the terminal failure came from a provider.api_error line, extract + // and forward the status code into the detail string. Without this, + // Kimi's two-line stderr format (status on the header line, human-readable + // text on the next "detail:" line) loses the "400" token after extraction, + // so classifyPoisonedError in daemon/poisoned.go cannot detect the + // permanently-poisoned-history fingerprint. + var apiErrTag string + apiErrLineIdx := -1 + for i, line := range s.lines { + if acpTerminalErrorRe.MatchString(line) { + if m := providerApiErrorStatusRe.FindStringSubmatch(line); m != nil { + apiErrTag = "provider.api_error: " + m[1] + " " + apiErrLineIdx = i + break + } + } + } + + for i, line := range s.lines { if m := acpErrorDetailRe.FindStringSubmatch(line); m != nil { detail := strings.TrimSpace(m[1]) - if detail != "" { - return acpTruncateError(prefix + detail) + if detail == "" { + continue + } + // acpErrorDetailRe's `provider\.api_error: [0-9]+` branch can + // backtrack one digit into `(.+)` when the status code is the last + // token on the line (two-line stderr format). Skip that artefact: a + // real detail is never pure digits. In the single-line format the + // status and detail text are on the same line, so `(.+)` captures + // the full text (non-digits) and the check passes correctly. + if i == apiErrLineIdx && isOnlyASCIIDigits(detail) { + continue } + return acpTruncateError(prefix + apiErrTag + detail) } } for _, line := range s.lines { @@ -2491,6 +2543,22 @@ func acpTruncateError(msg string) string { return strings.ToValidUTF8(msg[:acpMaxErrorLineLen], "") + "…(truncated)" } +// isOnlyASCIIDigits reports whether s is non-empty and consists entirely of +// ASCII digit characters. Used to detect regex backtracking artefacts in +// messageLocked where the "detail" captured is just the trailing digit(s) of +// a status code. +func isOnlyASCIIDigits(s string) bool { + if s == "" { + return false + } + for i := 0; i < len(s); i++ { + if s[i] < '0' || s[i] > '9' { + return false + } + } + return true +} + // promoteACPResultOnProviderError flips finalStatus to "failed" if // either (a) the stderr sniffer captured a terminal-failure marker, // (b) the adapter injected a synthetic "API call failed after N diff --git a/server/pkg/agent/hermes_test.go b/server/pkg/agent/hermes_test.go index fa0c508a7a..50841922bb 100644 --- a/server/pkg/agent/hermes_test.go +++ b/server/pkg/agent/hermes_test.go @@ -2325,19 +2325,30 @@ func TestACPProviderErrorSnifferKimiRateLimitNotTerminal(t *testing.T) { } } -// TestACPProviderErrorSnifferPoisonedHistory verifies that the specific -// kimi "assistant message must not be empty" (400) pattern is detected -// as a poisoned session so the backend can set ResumeRejected=true and -// the daemon will drop the broken history and start a fresh session. +// TestACPProviderErrorSnifferPoisonedHistory verifies that the +// "provider.api_error: 400 … must not be empty" pattern is detected as a +// poisoned session so the backend can set ResumeRejected=true and the daemon +// drops the broken session pointer instead of replaying the same bad history. func TestACPProviderErrorSnifferPoisonedHistory(t *testing.T) { t.Parallel() - // Exact pattern from the field: 400 + role 'assistant' + must not be empty. + // Single-line format: 400 + role 'assistant' + must not be empty on the + // same line. This is the format emitted by some kimi-cli versions. s := newACPProviderErrorSniffer("kimi") s.Write([]byte("error: failed to run prompt: provider.api_error: 400 the message at position 43 with role 'assistant' must not be empty\n")) - if !s.isPoisonedHistory() { - t.Fatal("expected isPoisonedHistory()=true for 400 assistant-empty error") + t.Fatal("single-line: expected isPoisonedHistory()=true for 400 assistant-empty error") + } + + // Two-line format: kimi emits the status on the error header and the + // human-readable detail on a separate "detail:" line. isPoisonedHistory + // must detect this across lines, not only when both markers are on the + // same line. + s4 := newACPProviderErrorSniffer("kimi") + s4.Write([]byte("error: failed to run prompt: provider.api_error: 400\n")) + s4.Write([]byte("detail: messages[43].content: content must not be empty\n")) + if !s4.isPoisonedHistory() { + t.Fatal("two-line: expected isPoisonedHistory()=true when 400 and must-not-be-empty are on separate lines") } // A plain 400 with a different error message is terminal but not a @@ -2345,15 +2356,47 @@ func TestACPProviderErrorSnifferPoisonedHistory(t *testing.T) { s2 := newACPProviderErrorSniffer("kimi") s2.Write([]byte("error: failed to run prompt: provider.api_error: 400 invalid model specified\n")) if s2.isPoisonedHistory() { - t.Error("plain 400 without role/must-not-be-empty should not be classified as poisoned history") + t.Error("plain 400 without must-not-be-empty should not be classified as poisoned history") + } + + // "must not be empty" on a detail line paired with a 429 header must + // not match — 429 is transient and not provider.api_error: 400. + s5 := newACPProviderErrorSniffer("kimi") + s5.Write([]byte("⚠️ API call failed after 3 retries: RateLimitError [HTTP 429]\n")) + s5.Write([]byte("detail: field must not be empty\n")) + if s5.isPoisonedHistory() { + t.Error("429 RateLimitError with must-not-be-empty detail should not be classified as poisoned history") + } +} + +// TestACPProviderErrorSnifferMessageLockedPreservesAPIErrorStatus verifies +// that messageLocked forwards the "provider.api_error: NNN" prefix into the +// formatted output even when the human-readable detail arrives on a separate +// stderr line. This is the contract classifyPoisonedError relies on. +func TestACPProviderErrorSnifferMessageLockedPreservesAPIErrorStatus(t *testing.T) { + t.Parallel() + + // Two-line format: status on header, detail on next line. + s := newACPProviderErrorSniffer("kimi") + s.Write([]byte("error: failed to run prompt: provider.api_error: 400\n")) + s.Write([]byte("detail: messages[43].content: content must not be empty\n")) + msg := s.terminalMessage() + if !strings.Contains(msg, "provider.api_error: 400") { + t.Errorf("two-line: terminalMessage() should include provider.api_error: 400, got %q", msg) + } + if !strings.Contains(msg, "must not be empty") { + t.Errorf("two-line: terminalMessage() should include the detail text, got %q", msg) } - // A 400 that mentions "role" but not specifically "role 'assistant'" must - // not match — the check was tightened to avoid false positives. - s3 := newACPProviderErrorSniffer("kimi") - s3.Write([]byte("error: failed to run prompt: provider.api_error: 400 the message at position 7 with role 'user' must not be empty\n")) - if s3.isPoisonedHistory() { - t.Error("400 with role 'user' (not 'assistant') should not be classified as poisoned history") + // Single-line format: status and detail on the same line. + s2 := newACPProviderErrorSniffer("kimi") + s2.Write([]byte("error: failed to run prompt: provider.api_error: 400 the message at position 43 with role 'assistant' must not be empty\n")) + msg2 := s2.terminalMessage() + if !strings.Contains(msg2, "provider.api_error: 400") { + t.Errorf("single-line: terminalMessage() should include provider.api_error: 400, got %q", msg2) + } + if !strings.Contains(msg2, "must not be empty") { + t.Errorf("single-line: terminalMessage() should include the detail text, got %q", msg2) } } diff --git a/server/pkg/agent/kimi_test.go b/server/pkg/agent/kimi_test.go index e9c2944d68..eb1e3a4fa9 100644 --- a/server/pkg/agent/kimi_test.go +++ b/server/pkg/agent/kimi_test.go @@ -479,3 +479,75 @@ done t.Fatal("timeout waiting for result") } } + +// TestKimiBackendPoisonedHistoryTwoLineFormat verifies that the two-line kimi +// stderr format (status on one line, detail on the next) is correctly detected +// as a poisoned session. This is the P1 blocker: isPoisonedHistory and +// classifyPoisonedError must both fire even when "provider.api_error: 400" and +// "must not be empty" arrive on separate lines. +func TestKimiBackendPoisonedHistoryTwoLineFormat(t *testing.T) { + t.Parallel() + + fakePath := filepath.Join(t.TempDir(), "kimi") + script := `#!/bin/sh +while IFS= read -r line; do + id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9]*\).*/\1/p') + case "$line" in + *'"method":"initialize"'*) + printf '{"jsonrpc":"2.0","id":%s,"result":{"protocolVersion":1,"agentCapabilities":{}}}\n' "$id" + ;; + *'"method":"session/resume"'*) + printf '{"jsonrpc":"2.0","id":%s,"result":{"sessionId":"ses_poison_2line"}}\n' "$id" + ;; + *'"method":"session/prompt"'*) + printf '%s\n' "error: failed to run prompt: provider.api_error: 400" >&2 + printf '%s\n' "detail: messages[43].content: content must not be empty" >&2 + printf '{"jsonrpc":"2.0","id":%s,"result":{"stopReason":"end_turn"}}\n' "$id" + exit 0 + ;; + esac +done +` + writeTestExecutable(t, fakePath, []byte(script)) + + backend, err := New("kimi", Config{ExecutablePath: fakePath, Logger: slog.Default()}) + if err != nil { + t.Fatalf("new kimi backend: %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + session, err := backend.Execute(ctx, "prompt-ignored", ExecOptions{ + Timeout: 5 * time.Second, + ResumeSessionID: "ses_poison_2line", + }) + if err != nil { + t.Fatalf("execute: %v", err) + } + go func() { + for range session.Messages { + } + }() + + select { + case result, ok := <-session.Result: + if !ok { + t.Fatal("result channel closed without a value") + } + if result.Status != "failed" { + t.Fatalf("two-line: expected status=failed, got %q (error=%q)", result.Status, result.Error) + } + if !result.ResumeRejected { + t.Error("two-line: expected ResumeRejected=true — isPoisonedHistory must detect across stderr lines") + } + if !strings.Contains(result.Error, "provider.api_error: 400") { + t.Errorf("two-line: expected error to contain provider.api_error: 400 (messageLocked must forward the status prefix), got %q", result.Error) + } + if !strings.Contains(result.Error, "must not be empty") { + t.Errorf("two-line: expected error to contain must not be empty, got %q", result.Error) + } + case <-time.After(10 * time.Second): + t.Fatal("timeout waiting for result") + } +} From f84b0e367728e909d9586d7acb52298138119153 Mon Sep 17 00:00:00 2001 From: kakiuwang-ui Date: Sat, 25 Jul 2026 12:22:15 +0800 Subject: [PATCH 7/7] fix(agent): exclude poisoned session from resume lookup after failed fresh retry (MUL-5785) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three-part fix for the remaining P1: when a Kimi fresh retry fails without establishing a new session, the prior completed task's session is still returned by GetLastTaskSession, causing the next run to resume the same poisoned history indefinitely. 1. daemon.go: when shouldRetryWithFreshSession is true and the fresh retry fails without establishing a new session (SessionID == ""), restore the original poisoned error message and PriorSessionID on the result. This ensures classifyPoisonedError labels the task api_invalid_request and the session_id column points at the poisoned session. 2. agent.sql / chat.sql: add NOT EXISTS subquery to both GetLastTaskSession and GetLastChatTaskSession. The subquery excludes any session that a subsequent task for the same (agent, issue) / chat_session has flagged as api_invalid_request — catching the completed-Task-A / poisoned-Task-B pattern that the direct failure_reason filter cannot reach. 3. rerun_session_test.go: add two integration tests covering both query paths: TestGetLastTaskSessionExcludesPoisonedCompletedViaSubsequentTask and TestGetLastChatTaskSessionExcludesPoisonedViaSubsequentTask. --- server/cmd/server/rerun_session_test.go | 97 +++++++++++++++++++++++++ server/internal/daemon/daemon.go | 11 +++ server/pkg/db/generated/agent.sql.go | 31 ++++++-- server/pkg/db/generated/chat.sql.go | 27 +++++-- server/pkg/db/queries/agent.sql | 31 ++++++-- server/pkg/db/queries/chat.sql | 27 +++++-- 6 files changed, 192 insertions(+), 32 deletions(-) diff --git a/server/cmd/server/rerun_session_test.go b/server/cmd/server/rerun_session_test.go index 48486a43ad..51ceb03a89 100644 --- a/server/cmd/server/rerun_session_test.go +++ b/server/cmd/server/rerun_session_test.go @@ -168,6 +168,103 @@ func TestGetLastTaskSessionExcludesAPIInvalidRequest(t *testing.T) { } } +// TestGetLastTaskSessionExcludesPoisonedCompletedViaSubsequentTask covers the +// MUL-5785 fresh-retry scenario: Task A completed successfully with session "S", +// then Task B resumed "S", detected a poisoned history, did a fresh retry that +// also failed, and was stored with session_id = "S" and failure_reason = +// api_invalid_request. The NOT EXISTS subquery must exclude "S" entirely so the +// next issue task does not resume the poisoned session, even though Task A's +// own row looks clean (status='completed'). +func TestGetLastTaskSessionExcludesPoisonedCompletedViaSubsequentTask(t *testing.T) { + if testPool == nil { + t.Skip("no database connection") + } + + issueID, agentID, runtimeID := setupRerunTestFixture(t) + t.Cleanup(func() { cleanupRerunFixture(t, issueID) }) + + ctx := context.Background() + + // Task A: completed successfully with session "S" — what the resume lookup + // would naively return. + if _, err := testPool.Exec(ctx, ` + INSERT INTO agent_task_queue (agent_id, runtime_id, issue_id, status, priority, started_at, completed_at, session_id, work_dir) + VALUES ($1, $2, $3, 'completed', 0, now() - interval '2 minutes', now() - interval '2 minutes', 'KIMI-POISONED', '/tmp/kimi') + `, agentID, runtimeID, issueID); err != nil { + t.Fatalf("insert Task A: %v", err) + } + + // Task B: resumed "S", got poisoned-history error, fresh retry also failed. + // Daemon stores session_id = PriorSessionID ("S") + failure_reason = api_invalid_request. + if _, err := testPool.Exec(ctx, ` + INSERT INTO agent_task_queue (agent_id, runtime_id, issue_id, status, priority, started_at, completed_at, session_id, work_dir, failure_reason, error) + VALUES ($1, $2, $3, 'failed', 0, now() - interval '1 minute', now() - interval '1 minute', 'KIMI-POISONED', '/tmp/kimi', 'api_invalid_request', + 'kimi provider error: the message at position 1 with role ''assistant'' must not be empty') + `, agentID, runtimeID, issueID); err != nil { + t.Fatalf("insert Task B: %v", err) + } + + queries := db.New(testPool) + prior, err := queries.GetLastTaskSession(ctx, db.GetLastTaskSessionParams{ + AgentID: pgtype.UUID{Bytes: parseUUIDBytes(agentID), Valid: true}, + IssueID: pgtype.UUID{Bytes: parseUUIDBytes(issueID), Valid: true}, + }) + if err == nil && prior.SessionID.Valid { + t.Fatalf("expected poisoned session to be excluded via NOT EXISTS, but got %q", prior.SessionID.String) + } +} + +// TestGetLastChatTaskSessionExcludesPoisonedViaSubsequentTask mirrors +// TestGetLastTaskSessionExcludesPoisonedCompletedViaSubsequentTask for the +// chat path. GetLastChatTaskSession must also exclude a session that a +// subsequent task in the same chat_session has marked as api_invalid_request. +func TestGetLastChatTaskSessionExcludesPoisonedViaSubsequentTask(t *testing.T) { + if testPool == nil { + t.Skip("no database connection") + } + + _, agentID, runtimeID := setupRerunTestFixture(t) + + ctx := context.Background() + + var chatSessionID string + if err := testPool.QueryRow(ctx, ` + INSERT INTO chat_session (workspace_id, agent_id, creator_id, title, runtime_id) + VALUES ($1, $2, $3, 'kimi-poison-test', $4) + RETURNING id + `, testWorkspaceID, agentID, testUserID, runtimeID).Scan(&chatSessionID); err != nil { + t.Fatalf("insert chat_session: %v", err) + } + t.Cleanup(func() { + testPool.Exec(ctx, `DELETE FROM agent_task_queue WHERE chat_session_id = $1`, chatSessionID) + testPool.Exec(ctx, `DELETE FROM chat_session WHERE id = $1`, chatSessionID) + }) + + // Task A: completed successfully with session "S" in this chat session. + if _, err := testPool.Exec(ctx, ` + INSERT INTO agent_task_queue (agent_id, runtime_id, chat_session_id, status, priority, started_at, completed_at, session_id, work_dir) + VALUES ($1, $2, $3, 'completed', 0, now() - interval '2 minutes', now() - interval '2 minutes', 'KIMI-CHAT-POISONED', '/tmp/kimi-chat') + `, agentID, runtimeID, chatSessionID); err != nil { + t.Fatalf("insert Task A: %v", err) + } + + // Task B: stored with session_id = "S" + api_invalid_request after a + // failed fresh retry following a poisoned-history rejection. + if _, err := testPool.Exec(ctx, ` + INSERT INTO agent_task_queue (agent_id, runtime_id, chat_session_id, status, priority, started_at, completed_at, session_id, work_dir, failure_reason, error) + VALUES ($1, $2, $3, 'failed', 0, now() - interval '1 minute', now() - interval '1 minute', 'KIMI-CHAT-POISONED', '/tmp/kimi-chat', 'api_invalid_request', + 'kimi provider error: the message at position 1 with role ''assistant'' must not be empty') + `, agentID, runtimeID, chatSessionID); err != nil { + t.Fatalf("insert Task B: %v", err) + } + + queries := db.New(testPool) + prior, err := queries.GetLastChatTaskSession(ctx, pgtype.UUID{Bytes: parseUUIDBytes(chatSessionID), Valid: true}) + if err == nil && prior.SessionID.Valid { + t.Fatalf("expected poisoned chat session to be excluded via NOT EXISTS, but got %q", prior.SessionID.String) + } +} + // TestGetLastTaskSessionExcludesCodexSemanticInactivity covers Codex // semantic inactivity timeouts: the failed task did establish a session, but // resuming it can replay the same stuck Codex state. The resume lookup must diff --git a/server/internal/daemon/daemon.go b/server/internal/daemon/daemon.go index 3e5eb1c0ff..1d9078c4f2 100644 --- a/server/internal/daemon/daemon.go +++ b/server/internal/daemon/daemon.go @@ -4770,6 +4770,7 @@ func (d *Daemon) runTask(ctx context.Context, task Task, provider string, slot i if shouldRetryWithFreshSession(result, task.PriorSessionID, tools, provider) { firstUsage := result.Usage + originalError := result.Error // preserve: carries the Kimi poisoned-session fingerprint taskLog.Warn("session resume failed, retrying with fresh session", "error", result.Error) execOpts.ResumeSessionID = "" retryResult, retryTools, retryErr := d.executeAndDrain(ctx, backend, prompt, execOpts, taskLog, task.ID, &msgSeq) @@ -4779,6 +4780,16 @@ func (d *Daemon) runTask(ctx context.Context, task Task, provider string, slot i result = retryResult result.Usage = mergeUsage(firstUsage, result.Usage) tools = retryTools + // If the fresh retry also failed without establishing a new session, + // carry forward the original poisoned session ID and error so + // classifyPoisonedError labels this task api_invalid_request. + // GetLastTaskSession's NOT EXISTS check then excludes the poisoned + // session from future resume lookups even though Task A (the prior + // successful run) was stored as "completed" (MUL-5785). + if result.Status != "completed" && result.SessionID == "" { + result.SessionID = task.PriorSessionID + result.Error = originalError + } } } diff --git a/server/pkg/db/generated/agent.sql.go b/server/pkg/db/generated/agent.sql.go index 594afdaa4d..f9315e21c2 100644 --- a/server/pkg/db/generated/agent.sql.go +++ b/server/pkg/db/generated/agent.sql.go @@ -2736,18 +2736,25 @@ func (q *Queries) GetAgentTaskInWorkspace(ctx context.Context, arg GetAgentTaskI } const getLastTaskSession = `-- name: GetLastTaskSession :one -SELECT session_id, work_dir, runtime_id FROM agent_task_queue -WHERE agent_id = $1 AND issue_id = $2 +SELECT t.session_id, t.work_dir, t.runtime_id FROM agent_task_queue t +WHERE t.agent_id = $1 AND t.issue_id = $2 AND ( - status = 'completed' + t.status = 'completed' OR ( - status = 'failed' - AND COALESCE(failure_reason, '') NOT IN ('iteration_limit', 'agent_fallback_message', 'api_invalid_request', 'codex_semantic_inactivity', 'agent_error.context_overflow') - AND NOT (COALESCE(error, '') ILIKE '%400%' AND COALESCE(error, '') ILIKE '%invalid_request_error%') + t.status = 'failed' + AND COALESCE(t.failure_reason, '') NOT IN ('iteration_limit', 'agent_fallback_message', 'api_invalid_request', 'codex_semantic_inactivity', 'agent_error.context_overflow') + AND NOT (COALESCE(t.error, '') ILIKE '%400%' AND COALESCE(t.error, '') ILIKE '%invalid_request_error%') ) ) - AND session_id IS NOT NULL -ORDER BY COALESCE(completed_at, started_at, dispatched_at, created_at) DESC + AND t.session_id IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM agent_task_queue poison + WHERE poison.agent_id = $1 + AND poison.issue_id = $2 + AND poison.session_id = t.session_id + AND poison.failure_reason = 'api_invalid_request' + ) +ORDER BY COALESCE(t.completed_at, t.started_at, t.dispatched_at, t.created_at) DESC LIMIT 1 ` @@ -2801,6 +2808,14 @@ type GetLastTaskSessionRow struct { // error text. Migration 079 backfills the failure_reason column itself, // so observability stays accurate; this clause guarantees session resume // never picks up a bad session even when failure_reason hasn't caught up. +// +// The NOT EXISTS subquery handles the Kimi fresh-retry scenario (MUL-5785): +// Task A completes successfully with session "S". Task B resumes "S", gets a +// poisoned-history error, does a fresh retry that also fails. The daemon sets +// Task B's session_id = PriorSessionID ("S") and failure_reason = +// api_invalid_request. Without NOT EXISTS, Task A is still returned because +// its own row looks clean (status='completed'). NOT EXISTS finds Task B and +// excludes session "S" entirely, even though Task A appeared healthy. func (q *Queries) GetLastTaskSession(ctx context.Context, arg GetLastTaskSessionParams) (GetLastTaskSessionRow, error) { row := q.db.QueryRow(ctx, getLastTaskSession, arg.AgentID, arg.IssueID) var i GetLastTaskSessionRow diff --git a/server/pkg/db/generated/chat.sql.go b/server/pkg/db/generated/chat.sql.go index a1ff29a067..96e83ec0b8 100644 --- a/server/pkg/db/generated/chat.sql.go +++ b/server/pkg/db/generated/chat.sql.go @@ -485,18 +485,24 @@ func (q *Queries) GetChatSessionInWorkspace(ctx context.Context, arg GetChatSess } const getLastChatTaskSession = `-- name: GetLastChatTaskSession :one -SELECT session_id, work_dir, runtime_id FROM agent_task_queue -WHERE chat_session_id = $1 +SELECT t.session_id, t.work_dir, t.runtime_id FROM agent_task_queue t +WHERE t.chat_session_id = $1 AND ( - status = 'completed' + t.status = 'completed' OR ( - status = 'failed' - AND COALESCE(failure_reason, '') NOT IN ('iteration_limit', 'agent_fallback_message', 'api_invalid_request', 'codex_semantic_inactivity', 'agent_error.context_overflow') - AND NOT (COALESCE(error, '') ILIKE '%400%' AND COALESCE(error, '') ILIKE '%invalid_request_error%') + t.status = 'failed' + AND COALESCE(t.failure_reason, '') NOT IN ('iteration_limit', 'agent_fallback_message', 'api_invalid_request', 'codex_semantic_inactivity', 'agent_error.context_overflow') + AND NOT (COALESCE(t.error, '') ILIKE '%400%' AND COALESCE(t.error, '') ILIKE '%invalid_request_error%') ) ) - AND session_id IS NOT NULL -ORDER BY completed_at DESC + AND t.session_id IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM agent_task_queue poison + WHERE poison.chat_session_id = $1 + AND poison.session_id = t.session_id + AND poison.failure_reason = 'api_invalid_request' + ) +ORDER BY t.completed_at DESC LIMIT 1 ` @@ -514,6 +520,11 @@ type GetLastChatTaskSessionRow struct { // excluded because replaying those sessions deterministically reproduces the // same terminal state. Keep this list in sync with resumeUnsafeFailureReason // and GetLastTaskSession. +// +// The NOT EXISTS subquery mirrors the same MUL-5785 fix as GetLastTaskSession: +// when a Kimi fresh retry fails without establishing a new session the daemon +// sets session_id = PriorSessionID so this subquery can exclude the poisoned +// session even if the prior task in that chat session appeared "completed". func (q *Queries) GetLastChatTaskSession(ctx context.Context, chatSessionID pgtype.UUID) (GetLastChatTaskSessionRow, error) { row := q.db.QueryRow(ctx, getLastChatTaskSession, chatSessionID) var i GetLastChatTaskSessionRow diff --git a/server/pkg/db/queries/agent.sql b/server/pkg/db/queries/agent.sql index d1dfc0ed86..3852aef0b5 100644 --- a/server/pkg/db/queries/agent.sql +++ b/server/pkg/db/queries/agent.sql @@ -710,18 +710,33 @@ RETURNING *; -- error text. Migration 079 backfills the failure_reason column itself, -- so observability stays accurate; this clause guarantees session resume -- never picks up a bad session even when failure_reason hasn't caught up. -SELECT session_id, work_dir, runtime_id FROM agent_task_queue -WHERE agent_id = $1 AND issue_id = $2 +-- +-- The NOT EXISTS subquery handles the Kimi fresh-retry scenario (MUL-5785): +-- Task A completes successfully with session "S". Task B resumes "S", gets a +-- poisoned-history error, does a fresh retry that also fails. The daemon sets +-- Task B's session_id = PriorSessionID ("S") and failure_reason = +-- api_invalid_request. Without NOT EXISTS, Task A is still returned because +-- its own row looks clean (status='completed'). NOT EXISTS finds Task B and +-- excludes session "S" entirely, even though Task A appeared healthy. +SELECT t.session_id, t.work_dir, t.runtime_id FROM agent_task_queue t +WHERE t.agent_id = $1 AND t.issue_id = $2 AND ( - status = 'completed' + t.status = 'completed' OR ( - status = 'failed' - AND COALESCE(failure_reason, '') NOT IN ('iteration_limit', 'agent_fallback_message', 'api_invalid_request', 'codex_semantic_inactivity', 'agent_error.context_overflow') - AND NOT (COALESCE(error, '') ILIKE '%400%' AND COALESCE(error, '') ILIKE '%invalid_request_error%') + t.status = 'failed' + AND COALESCE(t.failure_reason, '') NOT IN ('iteration_limit', 'agent_fallback_message', 'api_invalid_request', 'codex_semantic_inactivity', 'agent_error.context_overflow') + AND NOT (COALESCE(t.error, '') ILIKE '%400%' AND COALESCE(t.error, '') ILIKE '%invalid_request_error%') ) ) - AND session_id IS NOT NULL -ORDER BY COALESCE(completed_at, started_at, dispatched_at, created_at) DESC + AND t.session_id IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM agent_task_queue poison + WHERE poison.agent_id = $1 + AND poison.issue_id = $2 + AND poison.session_id = t.session_id + AND poison.failure_reason = 'api_invalid_request' + ) +ORDER BY COALESCE(t.completed_at, t.started_at, t.dispatched_at, t.created_at) DESC LIMIT 1; -- name: GetLastTaskStartedAtForIssueAndAgent :one diff --git a/server/pkg/db/queries/chat.sql b/server/pkg/db/queries/chat.sql index 1a90d64ecd..98d40ef15f 100644 --- a/server/pkg/db/queries/chat.sql +++ b/server/pkg/db/queries/chat.sql @@ -276,18 +276,29 @@ RETURNING *; -- excluded because replaying those sessions deterministically reproduces the -- same terminal state. Keep this list in sync with resumeUnsafeFailureReason -- and GetLastTaskSession. -SELECT session_id, work_dir, runtime_id FROM agent_task_queue -WHERE chat_session_id = $1 +-- +-- The NOT EXISTS subquery mirrors the same MUL-5785 fix as GetLastTaskSession: +-- when a Kimi fresh retry fails without establishing a new session the daemon +-- sets session_id = PriorSessionID so this subquery can exclude the poisoned +-- session even if the prior task in that chat session appeared "completed". +SELECT t.session_id, t.work_dir, t.runtime_id FROM agent_task_queue t +WHERE t.chat_session_id = $1 AND ( - status = 'completed' + t.status = 'completed' OR ( - status = 'failed' - AND COALESCE(failure_reason, '') NOT IN ('iteration_limit', 'agent_fallback_message', 'api_invalid_request', 'codex_semantic_inactivity', 'agent_error.context_overflow') - AND NOT (COALESCE(error, '') ILIKE '%400%' AND COALESCE(error, '') ILIKE '%invalid_request_error%') + t.status = 'failed' + AND COALESCE(t.failure_reason, '') NOT IN ('iteration_limit', 'agent_fallback_message', 'api_invalid_request', 'codex_semantic_inactivity', 'agent_error.context_overflow') + AND NOT (COALESCE(t.error, '') ILIKE '%400%' AND COALESCE(t.error, '') ILIKE '%invalid_request_error%') ) ) - AND session_id IS NOT NULL -ORDER BY completed_at DESC + AND t.session_id IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM agent_task_queue poison + WHERE poison.chat_session_id = $1 + AND poison.session_id = t.session_id + AND poison.failure_reason = 'api_invalid_request' + ) +ORDER BY t.completed_at DESC LIMIT 1; -- name: GetPendingChatTask :one