fix(agent): recover interrupted runs to failed on gateway startup (#1272)

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

@@ -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,

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{

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{

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{

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

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

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

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(