From 505b7a9d99a524b10d57cd5d95d6dd99ae50a9e6 Mon Sep 17 00:00:00 2001 From: Zezae Oh Date: Wed, 24 Jun 2026 16:03:00 +0900 Subject: [PATCH] fix(agent): recover interrupted runs to failed on gateway startup (#1272) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A run records its terminal run.status (completed/failed/cancelled) only when the agent loop returns. If the gateway process is killed mid-run (restart, crash, or SIGKILL after a slow graceful shutdown), the terminal event is never emitted, so the run's last run.status item stays "started" forever — it shows as perpetually running in the timeline and is never counted as failed. Mirror the cron scheduler's startup reset (recomputeStaleJobs flips stale 'running' jobs to 'interrupted'): add RunTimelineStore.RecoverInterruptedRuns, invoked once on gateway startup. It finds runs that have a started run.status but no terminal sibling and appends a terminal failed run.status item, marked interrupted in metadata so it stays distinguishable from a genuine agent failure. Running only at startup means nothing from before is still executing, so there is no false-positive risk — unlike a periodic sweep (see the deliberately-disabled tracing stale-trace loop). - store: RecoverInterruptedRuns on the interface + PG and SQLite impls - cmd/gateway_managed: invoke once after the timeline recorder is built - tests: SQLite store test — orphan recovered, completed run untouched, idempotent on re-run --- cmd/gateway_managed.go | 11 +++ internal/agent/run_timeline_recorder_test.go | 4 + internal/gateway/methods/run_timeline_test.go | 4 + internal/http/run_timeline_test.go | 4 + internal/store/pg/run_timeline.go | 82 +++++++++++++++++++ internal/store/run_timeline_store.go | 8 ++ internal/store/sqlitestore/run_timeline.go | 82 +++++++++++++++++++ .../store/sqlitestore/run_timeline_test.go | 72 ++++++++++++++++ 8 files changed, 267 insertions(+) diff --git a/cmd/gateway_managed.go b/cmd/gateway_managed.go index dabdfd9d..5e775b37 100644 --- a/cmd/gateway_managed.go +++ b/cmd/gateway_managed.go @@ -206,6 +206,17 @@ func wireExtras( slog.Info("agent hooks dispatcher wired", "handlers", "command,http,prompt") } timelineRecorder := agent.NewRunTimelineRecorder(stores.RunTimeline) + // Reconcile runs left mid-execution by a previous gateway stop. A run whose + // process was killed never emits its terminal run.status, so it would show as + // perpetually "running" and never be recorded as failed. Mark such runs failed + // on startup, mirroring the cron scheduler's stale-'running' reset. + if stores.RunTimeline != nil { + if n, err := stores.RunTimeline.RecoverInterruptedRuns(context.Background()); err != nil { + slog.Warn("run timeline: failed to recover interrupted runs on startup", "error", err) + } else if n > 0 { + slog.Info("run timeline: marked interrupted runs as failed on startup", "count", n) + } + } resolver := agent.NewManagedResolver(agent.ResolverDeps{ AgentStore: stores.Agents, diff --git a/internal/agent/run_timeline_recorder_test.go b/internal/agent/run_timeline_recorder_test.go index 2d2080b9..7b5e7f7b 100644 --- a/internal/agent/run_timeline_recorder_test.go +++ b/internal/agent/run_timeline_recorder_test.go @@ -28,6 +28,10 @@ func (s *recordingTimelineStore) ListRunTimelineItems(context.Context, store.Run return nil, nil } +func (s *recordingTimelineStore) RecoverInterruptedRuns(context.Context) (int64, error) { + return 0, nil +} + func TestRunTimelineItemFromEventScrubsToolArguments(t *testing.T) { tenantID := uuid.Must(uuid.NewV7()) item, ok := runTimelineItemFromEvent(AgentEvent{ diff --git a/internal/gateway/methods/run_timeline_test.go b/internal/gateway/methods/run_timeline_test.go index ba4d4d07..67760447 100644 --- a/internal/gateway/methods/run_timeline_test.go +++ b/internal/gateway/methods/run_timeline_test.go @@ -34,6 +34,10 @@ func (s *stubRunTimelineStore) ListRunTimelineItems(_ context.Context, opts stor return s.items, nil } +func (s *stubRunTimelineStore) RecoverInterruptedRuns(context.Context) (int64, error) { + return 0, nil +} + func TestRunTimelineGetScopesViewerByUser(t *testing.T) { timeline := &stubRunTimelineStore{ items: []store.RunTimelineItem{ diff --git a/internal/http/run_timeline_test.go b/internal/http/run_timeline_test.go index e9138704..0e773442 100644 --- a/internal/http/run_timeline_test.go +++ b/internal/http/run_timeline_test.go @@ -27,6 +27,10 @@ func (s *stubRunTimelineStore) ListRunTimelineItems(_ context.Context, opts stor return s.items, nil } +func (s *stubRunTimelineStore) RecoverInterruptedRuns(context.Context) (int64, error) { + return 0, nil +} + func TestRunTimelineHTTPScopesViewerByUser(t *testing.T) { token := setupTraceReadToken(t, "caller") timeline := &stubRunTimelineStore{ diff --git a/internal/store/pg/run_timeline.go b/internal/store/pg/run_timeline.go index cfd3e138..193550aa 100644 --- a/internal/store/pg/run_timeline.go +++ b/internal/store/pg/run_timeline.go @@ -162,3 +162,85 @@ func (r runTimelineRow) toStore() store.RunTimelineItem { TraceID: r.TraceID, SpanID: r.SpanID, Metadata: r.Metadata, CreatedAt: r.CreatedAt, } } + +// interruptedRunMetadata marks a backfilled terminal status so it is +// distinguishable from a genuine agent failure in the timeline. +var interruptedRunMetadata = json.RawMessage(`{"event_type":"run.failed","interrupted":true,"reason":"server_restart"}`) + +const interruptedRunPreview = "interrupted: gateway stopped while this run was in progress" + +// RecoverInterruptedRuns appends a terminal failed run.status item to every run +// that has a "started" run.status but no terminal sibling — i.e. runs killed +// mid-execution by a previous gateway stop, which would otherwise stay +// "running" forever. Cross-tenant (startup reconciliation); see the interface doc. +func (s *PGRunTimelineStore) RecoverInterruptedRuns(ctx context.Context) (int64, error) { + rows, err := s.db.QueryContext(ctx, ` + SELECT st.tenant_id, st.run_id, st.session_key, st.agent_id, st.user_id, st.channel, st.chat_id, agg.max_seq + FROM ( + SELECT run_id, MAX(seq) AS max_seq, + bool_or(item_type = 'run.status' AND status = 'started') AS has_start, + bool_or(item_type = 'run.status' AND status IN ('completed', 'failed', 'cancelled')) AS has_term + FROM run_timeline_items + GROUP BY run_id + ) agg + JOIN run_timeline_items st + ON st.run_id = agg.run_id AND st.item_type = 'run.status' AND st.status = 'started' + WHERE agg.has_start AND NOT agg.has_term`) + if err != nil { + return 0, fmt.Errorf("list interrupted runs: %w", err) + } + orphans, err := scanInterruptedRuns(rows) + if err != nil { + return 0, err + } + + var recovered int64 + for i := range orphans { + item := &orphans[i] + if err := s.AppendRunTimelineItem(store.WithTenantID(ctx, item.TenantID), item); err != nil { + return recovered, fmt.Errorf("append interrupted terminal for run %s: %w", item.RunID, err) + } + recovered++ + } + return recovered, nil +} + +// scanInterruptedRuns reads orphaned-run rows and pre-builds the terminal failed +// item to append for each. Rows are fully drained and closed before returning so +// the caller can issue inserts on the same pool without cursor contention. +func scanInterruptedRuns(rows *sql.Rows) ([]store.RunTimelineItem, error) { + defer rows.Close() + var items []store.RunTimelineItem + for rows.Next() { + var ( + tenantID uuid.UUID + runID, sessionKey string + agentID uuid.NullUUID + userID, channel, chatID sql.NullString + maxSeq int + ) + if err := rows.Scan(&tenantID, &runID, &sessionKey, &agentID, &userID, &channel, &chatID, &maxSeq); err != nil { + return nil, fmt.Errorf("scan interrupted run: %w", err) + } + item := store.RunTimelineItem{ + TenantID: tenantID, + RunID: runID, + SessionKey: sessionKey, + UserID: userID.String, + Channel: channel.String, + ChatID: chatID.String, + Seq: maxSeq + 1, + ItemType: store.RunTimelineItemTypeRunStatus, + Status: store.RunTimelineStatusFailed, + Title: "Run failed", + Preview: interruptedRunPreview, + Metadata: interruptedRunMetadata, + } + if agentID.Valid { + id := agentID.UUID + item.AgentID = &id + } + items = append(items, item) + } + return items, rows.Err() +} diff --git a/internal/store/run_timeline_store.go b/internal/store/run_timeline_store.go index 1dda34e9..513b7a8d 100644 --- a/internal/store/run_timeline_store.go +++ b/internal/store/run_timeline_store.go @@ -61,4 +61,12 @@ type RunTimelineListOpts struct { type RunTimelineStore interface { AppendRunTimelineItem(ctx context.Context, item *RunTimelineItem) error ListRunTimelineItems(ctx context.Context, opts RunTimelineListOpts) ([]RunTimelineItem, error) + // RecoverInterruptedRuns reconciles runs left mid-execution when the gateway + // stopped: a run with a "started" run.status item but no terminal + // (completed/failed/cancelled) run.status sibling. For each, it appends a + // terminal failed run.status item so the run is no longer reported as + // perpetually running. Intended to run once on startup (cross-tenant), + // mirroring the cron scheduler's stale-'running' reset. Returns the number + // of runs recovered. + RecoverInterruptedRuns(ctx context.Context) (int64, error) } diff --git a/internal/store/sqlitestore/run_timeline.go b/internal/store/sqlitestore/run_timeline.go index c3c4723d..c651e80d 100644 --- a/internal/store/sqlitestore/run_timeline.go +++ b/internal/store/sqlitestore/run_timeline.go @@ -153,3 +153,85 @@ func scanRunTimelineRows(rows *sql.Rows) ([]store.RunTimelineItem, error) { } return items, rows.Err() } + +const interruptedRunPreview = "interrupted: gateway stopped while this run was in progress" + +// interruptedRunMetadata marks a backfilled terminal status so it is +// distinguishable from a genuine agent failure in the timeline. +var interruptedRunMetadata = []byte(`{"event_type":"run.failed","interrupted":true,"reason":"server_restart"}`) + +// RecoverInterruptedRuns appends a terminal failed run.status item to every run +// that has a "started" run.status but no terminal sibling — i.e. runs killed +// mid-execution by a previous gateway stop, which would otherwise stay +// "running" forever. Cross-tenant (startup reconciliation); see the interface doc. +func (s *SQLiteRunTimelineStore) RecoverInterruptedRuns(ctx context.Context) (int64, error) { + rows, err := s.db.QueryContext(ctx, ` + SELECT st.tenant_id, st.run_id, st.session_key, st.agent_id, st.user_id, st.channel, st.chat_id, agg.max_seq + FROM ( + SELECT run_id, MAX(seq) AS max_seq, + MAX(item_type = 'run.status' AND status = 'started') AS has_start, + MAX(item_type = 'run.status' AND status IN ('completed', 'failed', 'cancelled')) AS has_term + FROM run_timeline_items + GROUP BY run_id + ) agg + JOIN run_timeline_items st + ON st.run_id = agg.run_id AND st.item_type = 'run.status' AND st.status = 'started' + WHERE agg.has_start = 1 AND IFNULL(agg.has_term, 0) = 0`) + if err != nil { + return 0, fmt.Errorf("list interrupted runs: %w", err) + } + orphans, err := scanInterruptedRuns(rows) + if err != nil { + return 0, err + } + + var recovered int64 + for i := range orphans { + item := &orphans[i] + if err := s.AppendRunTimelineItem(store.WithTenantID(ctx, item.TenantID), item); err != nil { + return recovered, fmt.Errorf("append interrupted terminal for run %s: %w", item.RunID, err) + } + recovered++ + } + return recovered, nil +} + +// scanInterruptedRuns reads orphaned-run rows and pre-builds the terminal failed +// item to append for each. Rows are fully drained and closed before returning so +// the caller can issue inserts on the same connection without cursor contention. +func scanInterruptedRuns(rows *sql.Rows) ([]store.RunTimelineItem, error) { + defer rows.Close() + var items []store.RunTimelineItem + for rows.Next() { + var ( + tenantID uuid.UUID + runID, sessionKey string + agentID uuid.NullUUID + userID, channel, chatID sql.NullString + maxSeq int + ) + if err := rows.Scan(&tenantID, &runID, &sessionKey, &agentID, &userID, &channel, &chatID, &maxSeq); err != nil { + return nil, fmt.Errorf("scan interrupted run: %w", err) + } + item := store.RunTimelineItem{ + TenantID: tenantID, + RunID: runID, + SessionKey: sessionKey, + UserID: userID.String, + Channel: channel.String, + ChatID: chatID.String, + Seq: maxSeq + 1, + ItemType: store.RunTimelineItemTypeRunStatus, + Status: store.RunTimelineStatusFailed, + Title: "Run failed", + Preview: interruptedRunPreview, + Metadata: interruptedRunMetadata, + } + if agentID.Valid { + id := agentID.UUID + item.AgentID = &id + } + items = append(items, item) + } + return items, rows.Err() +} diff --git a/internal/store/sqlitestore/run_timeline_test.go b/internal/store/sqlitestore/run_timeline_test.go index 139be542..471f560b 100644 --- a/internal/store/sqlitestore/run_timeline_test.go +++ b/internal/store/sqlitestore/run_timeline_test.go @@ -132,6 +132,78 @@ func TestSQLiteRunTimelineStoreTenantScope(t *testing.T) { } } +func TestSQLiteRunTimelineStoreRecoverInterruptedRuns(t *testing.T) { + db := openTestDB(t) + if err := EnsureSchema(db); err != nil { + t.Fatalf("EnsureSchema: %v", err) + } + + timeline := NewSQLiteRunTimelineStore(db) + ctx := store.WithTenantID(context.Background(), store.MasterTenantID) + seedSQLiteRunTimelineTenant(t, db, store.MasterTenantID) + + add := func(item store.RunTimelineItem) { + t.Helper() + item.SessionKey = "agent:default:direct:user-1" + if err := timeline.AppendRunTimelineItem(ctx, &item); err != nil { + t.Fatalf("AppendRunTimelineItem(%s/%d): %v", item.RunID, item.Seq, err) + } + } + + // run-interrupted: started + an in-flight tool.call, no terminal run.status. + add(store.RunTimelineItem{RunID: "run-interrupted", Seq: 1, ItemType: store.RunTimelineItemTypeRunStatus, Status: store.RunTimelineStatusStarted, Title: "Run started"}) + add(store.RunTimelineItem{RunID: "run-interrupted", Seq: 2, ItemType: store.RunTimelineItemTypeToolCall, Status: store.RunTimelineStatusRunning, Title: "exec"}) + // run-done: started + completed terminal — must be left untouched. + add(store.RunTimelineItem{RunID: "run-done", Seq: 1, ItemType: store.RunTimelineItemTypeRunStatus, Status: store.RunTimelineStatusStarted, Title: "Run started"}) + add(store.RunTimelineItem{RunID: "run-done", Seq: 2, ItemType: store.RunTimelineItemTypeRunStatus, Status: store.RunTimelineStatusCompleted, Title: "Run completed"}) + + n, err := timeline.RecoverInterruptedRuns(context.Background()) + if err != nil { + t.Fatalf("RecoverInterruptedRuns: %v", err) + } + if n != 1 { + t.Fatalf("recovered = %d, want 1 (only run-interrupted)", n) + } + + // run-interrupted gained a terminal failed run.status at seq 3. + got, err := timeline.ListRunTimelineItems(ctx, store.RunTimelineListOpts{RunID: "run-interrupted", Limit: 10}) + if err != nil { + t.Fatalf("List run-interrupted: %v", err) + } + var terminal *store.RunTimelineItem + for i := range got { + if got[i].ItemType == store.RunTimelineItemTypeRunStatus && got[i].Status == store.RunTimelineStatusFailed { + terminal = &got[i] + } + } + if terminal == nil { + t.Fatalf("no terminal failed run.status appended to run-interrupted") + } + if terminal.Seq != 3 { + t.Fatalf("terminal seq = %d, want 3 (max+1)", terminal.Seq) + } + + // run-done must not have gained a failed status. + doneItems, err := timeline.ListRunTimelineItems(ctx, store.RunTimelineListOpts{RunID: "run-done", Limit: 10}) + if err != nil { + t.Fatalf("List run-done: %v", err) + } + for _, it := range doneItems { + if it.Status == store.RunTimelineStatusFailed { + t.Fatalf("run-done wrongly marked failed at seq %d", it.Seq) + } + } + + // Idempotent: a second pass finds nothing new (run-interrupted now terminal). + n2, err := timeline.RecoverInterruptedRuns(context.Background()) + if err != nil { + t.Fatalf("RecoverInterruptedRuns (2nd): %v", err) + } + if n2 != 0 { + t.Fatalf("second recover = %d, want 0 (idempotent)", n2) + } +} + func seedSQLiteRunTimelineTenant(t *testing.T, db execer, tenantID uuid.UUID) { t.Helper() if _, err := db.Exec(