Files
multica/server/internal/service/retry_deferred_test.go
Bohan Jiang 30318b79bc MUL-5426: fix(daemon): retire sessions whose history the provider refuses to replay (#6083)
* fix(daemon): retire sessions whose history the provider refuses to replay

A run killed mid-reply (machine shutdown, force-quit, SIGKILL) can leave an
empty assistant message in the agent CLI's transcript. Every later resume
replays it, the provider rejects the request, and the (agent, issue) pair is
bricked with no self-healing and no user-facing recovery.

Multica already has the mechanism for this — poisoned-session classification —
but its detector paired "400" with "invalid_request_error", which is the
Anthropic wire shape. The same defect reported by any other provider carried
neither token, so it classified as agent_error.unknown: resume-safe by
omission. GetLastTaskSession kept handing back the dead session on every
follow-up, manual Rerun resolved it through the same predicate, and the
in-turn fresh-session retry never fired because ResumeRejected is false here
(nothing rejected the resume — the transcript loaded and the provider refused
to replay it).

Add taskfailure.UnresumableHistory, which recognises the defect by what the
provider says is wrong — some content is empty, and here is which message in
the history — rather than by status code or provider name. Both signals are
required, so a tool reporting "field must not be empty" does not match.

Wire it into the four places that decide whether a session survives:

- classifyPoisonedError, so the task is written as api_invalid_request
- shouldRetryWithFreshSession, so the turn recovers on all 17 backends
  instead of the subset whose adapter learned to detect it; the tools == 0
  gate is unchanged, so a run that already used a tool is never re-run
- ResumeUnsafeFailure, covering the manual-Rerun path
- both resume queries, as defense-in-depth for hosts whose daemon predates
  this (self-host daemons upgrade on their own cadence)

Fixes #6066. Also covers the daemon half of #5760.

Co-authored-by: multica-agent <github@multica.ai>

* fix(session): close the Chat and fresh-retry paths that resurrect a poisoned session

Review found the previous commit stopped short in two places, both of which
put the dead transcript back in play.

Chat never consulted the guarded query. The claim handler reads
chat_session.session_id first and only falls back to GetLastChatTaskSession
when it is empty, so a poisoned pointer there bypasses every filter that query
applies. The fail path merely declined to OVERWRITE the pointer, leaving it in
place. It now clears it in the same transaction, matched on session and
runtime so a concurrent turn's newer pointer survives. The promote guard moves
to ResumeUnsafeFailure as well — the reason-only check passed an un-upgraded
daemon's agent_error.unknown row and re-pinned what the clear had just removed.

GetLastChatTaskSession also kept the row-level filter the issue query dropped
in GH #5975: it discarded the newest poisoned row and fell back to an older
completed row carrying the same dead session. It now judges each session by
its latest terminal state, matching GetLastTaskSession.

A recovered turn could not retire anything. A terminal report carried one
session_id, and an empty one meant both "nothing to report" and "forget the
old session", so a fresh-session retry that SUCCEEDED left the id it retried
away from selectable — through an older completed row on the issue, or through
the chat pointer. agent_task_queue.retired_session_id records the abandonment
itself, reported on every terminal path including completed, and both resume
lookups exclude it. This is the contract gap the previous PR deferred; the
fresh-retry path now runs on all backends, so deferring it is not safe.

Also narrows what the cross-backend test claims: it pins the shared decision,
not that all 17 adapters surface the error into Result.Error (#5760 is the
counter-example), and says so.

Co-authored-by: multica-agent <github@multica.ai>

* test(session): require pgx.ErrNoRows in the resume-exclusion assertions

The `if err == nil && prior.SessionID.Valid` form these tests shared is
false-green: any real fault — undefined column, syntax error, dead connection
— makes err non-nil, so the condition is false and the test passes. Run
against a database missing this branch's new column, the exclusion tests
reported PASS on a SQLSTATE 42703, meaning they could not have caught a broken
query.

requireSessionExcluded demands pgx.ErrNoRows specifically and fails loudly on
anything else, so a green run now means the filter worked rather than the
query never ran.

Applied to all nine sites, not just the four this branch added: the other five
guard the same GetLastTaskSession exclusion behaviour that this branch
changes, so leaving them false-green would leave the change under-tested. All
nine pass on a correctly migrated database.

Co-authored-by: multica-agent <github@multica.ai>

---------

Co-authored-by: Bohan-J <bohan@devv.ai>
Co-authored-by: multica-agent <github@multica.ai>
2026-07-29 15:54:51 +08:00

170 lines
6.8 KiB
Go

package service
import (
"context"
"testing"
"time"
"github.com/jackc/pgx/v5/pgtype"
"github.com/multica-ai/multica/server/internal/events"
"github.com/multica-ai/multica/server/internal/util"
db "github.com/multica-ai/multica/server/pkg/db/generated"
)
// TestCreateRetryTaskFireAtControlsDeferral locks in the SQL half of the
// three-tier provider_network schedule (MUL-4910): CreateRetryTask inserts a
// 'deferred' child carrying fire_at when the fire_at param is set (the final,
// backed-off attempt) and an immediately-claimable 'queued' child when it is
// NULL (every other retry). Both continue the resume chain — force_fresh_session
// stays false for a provider_network parent.
func TestCreateRetryTaskFireAtControlsDeferral(t *testing.T) {
pool := newResolveOriginatorPool(t)
ctx := context.Background()
q := db.New(pool)
_, _, agentID, issueID := seedAttributionFixture(t, pool)
// agent_task_queue.runtime_id is NOT NULL; reuse the fixture agent's runtime.
var runtimeID string
if err := pool.QueryRow(ctx, `SELECT runtime_id::text FROM agent WHERE id = $1`, agentID).Scan(&runtimeID); err != nil {
t.Fatalf("read agent runtime: %v", err)
}
// Parent: a provider_network failure on its second attempt — the point at
// which the schedule wants the next (final) retry deferred.
var parentID pgtype.UUID
if err := pool.QueryRow(ctx, `
INSERT INTO agent_task_queue (agent_id, runtime_id, issue_id, status, priority, attempt, max_attempts, failure_reason, session_id, work_dir)
VALUES ($1, $2, $3, 'failed', 0, 2, 2, 'agent_error.provider_network', 'src-session', '/tmp/src-workdir')
RETURNING id
`, agentID, runtimeID, issueID).Scan(&parentID); err != nil {
t.Fatalf("insert parent task: %v", err)
}
t.Cleanup(func() {
pool.Exec(context.Background(), `DELETE FROM agent_task_queue WHERE parent_task_id = $1 OR id = $1`, parentID)
})
cases := []struct {
name string
fireAt pgtype.Timestamptz
maxAttempts pgtype.Int4
wantStatus string
wantFireAt bool
wantMaxAttempts int32
}{
// Final tier: deferred, and the effective budget (3) written into the row
// so it self-describes as attempt=3/max_attempts=3, not attempt=3/max=2.
{"deferred final tier persists budget", pgtype.Timestamptz{Time: time.Now().Add(5 * time.Second), Valid: true}, pgtype.Int4{Int32: 3, Valid: true}, "deferred", true, 3},
// NULL max_attempts inherits the parent's column (COALESCE fallback).
{"queued immediate tier inherits budget", pgtype.Timestamptz{}, pgtype.Int4{}, "queued", false, 2},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
child, err := q.CreateRetryTask(ctx, db.CreateRetryTaskParams{ID: parentID, FireAt: tc.fireAt, MaxAttempts: tc.maxAttempts})
if err != nil {
t.Fatalf("CreateRetryTask: %v", err)
}
t.Cleanup(func() { pool.Exec(context.Background(), `DELETE FROM agent_task_queue WHERE id = $1`, child.ID) })
if child.Status != tc.wantStatus {
t.Errorf("status = %q, want %q", child.Status, tc.wantStatus)
}
if child.FireAt.Valid != tc.wantFireAt {
t.Errorf("fire_at valid = %v, want %v", child.FireAt.Valid, tc.wantFireAt)
}
if child.Attempt != 3 {
t.Errorf("attempt = %d, want 3 (parent attempt 2 + 1)", child.Attempt)
}
if child.MaxAttempts != tc.wantMaxAttempts {
t.Errorf("max_attempts = %d, want %d", child.MaxAttempts, tc.wantMaxAttempts)
}
// provider_network is resume-safe: the retry must continue the session.
if child.ForceFreshSession {
t.Errorf("force_fresh_session = true, want false (provider_network resumes) for %s", util.UUIDToString(child.ID))
}
})
}
}
// TestFailTaskProviderNetworkBudget is the end-to-end guard for Elon's must-fix
// (MUL-4910): FailTask must (1) grant provider_network its raised budget and
// persist a self-consistent child (attempt=3, max_attempts=3), and (2) still
// honour max_attempts=1 as "auto-retry disabled" — no child at all.
func TestFailTaskProviderNetworkBudget(t *testing.T) {
pool := newResolveOriginatorPool(t)
ctx := context.Background()
q := db.New(pool)
_, _, agentID, issueID := seedAttributionFixture(t, pool)
svc := &TaskService{Queries: q, TxStarter: pool, Bus: events.New()}
var runtimeID string
if err := pool.QueryRow(ctx, `SELECT runtime_id::text FROM agent WHERE id = $1`, agentID).Scan(&runtimeID); err != nil {
t.Fatalf("read agent runtime: %v", err)
}
cases := []struct {
name string
attempt int32
maxAttempts int32
wantChild bool
wantAttempt int32
wantMax int32
wantDeferred bool
}{
// Default budget, failing on the 2nd attempt → deferred final tier that
// records attempt=3 AND max_attempts=3 (no contradictory row).
{"final tier persists raised budget", 2, 2, true, 3, 3, true},
// Default budget, failing on the 1st attempt → immediate 2nd tier.
{"first retry is immediate", 1, 2, true, 2, 3, false},
// max_attempts=1 disables auto-retry — even provider_network gets none.
{"disabled budget is never revived", 1, 1, false, 0, 0, false},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
var parentID pgtype.UUID
if err := pool.QueryRow(ctx, `
INSERT INTO agent_task_queue (agent_id, runtime_id, issue_id, status, priority, attempt, max_attempts, session_id, work_dir)
VALUES ($1, $2, $3, 'running', 0, $4, $5, 'src-session', '/tmp/src-workdir')
RETURNING id
`, agentID, runtimeID, issueID, tc.attempt, tc.maxAttempts).Scan(&parentID); err != nil {
t.Fatalf("insert parent task: %v", err)
}
t.Cleanup(func() {
pool.Exec(context.Background(), `DELETE FROM agent_task_queue WHERE parent_task_id = $1 OR id = $1`, parentID)
})
if _, err := svc.FailTask(ctx, parentID, "API Error: Connection closed mid-response.", "src-session", "/tmp/src-workdir", "agent_error.provider_network", false, ""); err != nil {
t.Fatalf("FailTask: %v", err)
}
var (
childAttempt, childMax int32
childStatus string
n int
)
row := pool.QueryRow(ctx, `SELECT count(*), coalesce(max(attempt),0), coalesce(max(max_attempts),0), coalesce(max(status),'') FROM agent_task_queue WHERE parent_task_id = $1`, parentID)
if err := row.Scan(&n, &childAttempt, &childMax, &childStatus); err != nil {
t.Fatalf("read child: %v", err)
}
if !tc.wantChild {
if n != 0 {
t.Fatalf("expected no retry child, got %d", n)
}
return
}
if n != 1 {
t.Fatalf("expected exactly one retry child, got %d", n)
}
if childAttempt != tc.wantAttempt {
t.Errorf("child attempt = %d, want %d", childAttempt, tc.wantAttempt)
}
if childMax != tc.wantMax {
t.Errorf("child max_attempts = %d, want %d (self-consistent budget)", childMax, tc.wantMax)
}
gotDeferred := childStatus == "deferred"
if gotDeferred != tc.wantDeferred {
t.Errorf("child status = %q, want deferred=%v", childStatus, tc.wantDeferred)
}
})
}
}