From 9ee42b53e3b1e346b335ed8ea71a25119daf4804 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Wed, 6 May 2026 20:33:23 +0100 Subject: [PATCH] fix(orchestrator): abandon orphan turns on session resume (#837) --- apps/backend/cmd/kandev/adapters.go | 4 + .../internal/integration/test_server_test.go | 4 + apps/backend/internal/orchestrator/service.go | 5 + .../internal/orchestrator/task_operations.go | 19 ++ .../orchestrator/turn_lifecycle_test.go | 70 ++++++ .../task/handlers/process_handlers_test.go | 3 + .../internal/task/repository/interface.go | 1 + .../task/repository/sqlite/session.go | 15 ++ .../internal/task/service/service_turns.go | 67 ++++++ .../task/service/service_turns_test.go | 201 ++++++++++++++++++ 10 files changed, 389 insertions(+) diff --git a/apps/backend/cmd/kandev/adapters.go b/apps/backend/cmd/kandev/adapters.go index f71f6c8af6..ae5b6ac130 100644 --- a/apps/backend/cmd/kandev/adapters.go +++ b/apps/backend/cmd/kandev/adapters.go @@ -770,6 +770,10 @@ func (a *turnServiceAdapter) GetActiveTurn(ctx context.Context, sessionID string return a.svc.GetActiveTurn(ctx, sessionID) } +func (a *turnServiceAdapter) AbandonOpenTurns(ctx context.Context, sessionID string) error { + return a.svc.AbandonOpenTurns(ctx, sessionID) +} + func newTurnServiceAdapter(svc *taskservice.Service) *turnServiceAdapter { return &turnServiceAdapter{svc: svc} } diff --git a/apps/backend/internal/integration/test_server_test.go b/apps/backend/internal/integration/test_server_test.go index e61b1fa783..4b50966167 100644 --- a/apps/backend/internal/integration/test_server_test.go +++ b/apps/backend/internal/integration/test_server_test.go @@ -217,6 +217,10 @@ func (a *testTurnServiceAdapter) GetActiveTurn(ctx context.Context, sessionID st return a.svc.GetActiveTurn(ctx, sessionID) } +func (a *testTurnServiceAdapter) AbandonOpenTurns(ctx context.Context, sessionID string) error { + return a.svc.AbandonOpenTurns(ctx, sessionID) +} + // NewOrchestratorTestServer creates a test server with full orchestrator support func NewOrchestratorTestServer(t *testing.T) *OrchestratorTestServer { t.Helper() diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index ba91807405..7c57e518ad 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -91,6 +91,11 @@ type TurnService interface { StartTurn(ctx context.Context, sessionID string) (*models.Turn, error) CompleteTurn(ctx context.Context, turnID string) error GetActiveTurn(ctx context.Context, sessionID string) (*models.Turn, error) + // AbandonOpenTurns buries any open turns for a session by setting + // completed_at = started_at (zero duration), so a subsequent prompt starts + // a fresh turn instead of adopting one that was orphaned by a crash or + // restart. Used on session resume. + AbandonOpenTurns(ctx context.Context, sessionID string) error } // TaskEventPublisher is the orchestrator's collaborator for publishing diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index 2f23877e3b..56a6c86752 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -747,6 +747,25 @@ func (s *Service) ResumeTaskSession(ctx context.Context, taskID, sessionID strin return nil, fmt.Errorf("session is completed and cannot be resumed; create a new session instead") } + // Bury any open turns from the previous run before relaunching. Without + // this, startTurnForSession adopts the orphan on the next prompt and the + // UI's running timer counts from the orphan's started_at — which can be + // hours or days ago. Zero-duration completion keeps analytics honest about + // the dead window. A failure here shouldn't block the resume; the next + // completeTurnForSession sweep will mop up. + // + // Drop the activeTurns cache entry first, mirroring completeTurnForSession. + // Otherwise a stale entry would let getActiveTurnID return the now-abandoned + // turn ID without re-reading the DB, tagging new messages to a closed turn. + if s.turnService != nil { + s.activeTurns.Delete(sessionID) + if err := s.turnService.AbandonOpenTurns(ctx, sessionID); err != nil { + s.logger.Warn("failed to abandon orphan turns on resume; continuing", + zap.String("session_id", sessionID), + zap.Error(err)) + } + } + // Use context.WithoutCancel to prevent WebSocket request timeout from canceling the resume. // Session resume can take time and shouldn't be tied to the WS request lifecycle. resumeCtx := context.WithoutCancel(ctx) diff --git a/apps/backend/internal/orchestrator/turn_lifecycle_test.go b/apps/backend/internal/orchestrator/turn_lifecycle_test.go index c0b51f47f8..8ae6fc77b3 100644 --- a/apps/backend/internal/orchestrator/turn_lifecycle_test.go +++ b/apps/backend/internal/orchestrator/turn_lifecycle_test.go @@ -49,6 +49,18 @@ func (a *repoTurnService) GetActiveTurn(ctx context.Context, sessionID string) ( return turn, err } +func (a *repoTurnService) AbandonOpenTurns(ctx context.Context, sessionID string) error { + for { + turn, err := a.GetActiveTurn(ctx, sessionID) + if err != nil || turn == nil { + return err + } + if err := a.repo.AbandonTurn(ctx, turn.ID); err != nil { + return err + } + } +} + func openTurnCount(t *testing.T, repo *sqliterepo.Repository, sessionID string) int { t.Helper() turns, err := repo.ListTurnsBySession(context.Background(), sessionID) @@ -230,3 +242,61 @@ func TestCompleteTurnIsIdempotent(t *testing.T) { t.Fatalf("expected exactly 1 turn (no phantom), got %d", len(turns)) } } + +// TestAbandonOpenTurnsZeroesDuration covers the resume-orphan path: turns +// left open by a previous crash must close with completed_at = started_at so +// the UI's running timer doesn't count from a stale start, and analytics +// doesn't accumulate hours of dead time. +func TestAbandonOpenTurnsZeroesDuration(t *testing.T) { + svc, repo := newTurnLifecycleTestService(t) + ctx := context.Background() + + stale, err := svc.turnService.StartTurn(ctx, "session1") + if err != nil { + t.Fatalf("seed turn: %v", err) + } + + if err := svc.turnService.AbandonOpenTurns(ctx, "session1"); err != nil { + t.Fatalf("AbandonOpenTurns: %v", err) + } + + if open := openTurnCount(t, repo, "session1"); open != 0 { + t.Fatalf("expected 0 open turns after abandon, got %d", open) + } + + got, err := repo.GetTurn(ctx, stale.ID) + if err != nil { + t.Fatalf("GetTurn: %v", err) + } + if got.CompletedAt == nil { + t.Fatal("expected completed_at to be set after abandon") + } + if !got.CompletedAt.Equal(got.StartedAt) { + t.Fatalf("expected completed_at == started_at (zero duration), got started=%v completed=%v", + got.StartedAt, *got.CompletedAt) + } +} + +// TestAbandonOpenTurnsHandlesMultipleZombies verifies the loop closes every +// open turn for the session, mirroring the behavior of completeTurnForSession. +func TestAbandonOpenTurnsHandlesMultipleZombies(t *testing.T) { + svc, repo := newTurnLifecycleTestService(t) + ctx := context.Background() + + for i := 0; i < 4; i++ { + if _, err := svc.turnService.StartTurn(ctx, "session1"); err != nil { + t.Fatalf("seed turn %d: %v", i, err) + } + } + if open := openTurnCount(t, repo, "session1"); open != 4 { + t.Fatalf("expected 4 open turns before abandon, got %d", open) + } + + if err := svc.turnService.AbandonOpenTurns(ctx, "session1"); err != nil { + t.Fatalf("AbandonOpenTurns: %v", err) + } + + if open := openTurnCount(t, repo, "session1"); open != 0 { + t.Fatalf("expected 0 open turns after abandon, got %d", open) + } +} diff --git a/apps/backend/internal/task/handlers/process_handlers_test.go b/apps/backend/internal/task/handlers/process_handlers_test.go index 20d694d268..35292e029c 100644 --- a/apps/backend/internal/task/handlers/process_handlers_test.go +++ b/apps/backend/internal/task/handlers/process_handlers_test.go @@ -175,6 +175,9 @@ func (m *mockRepository) UpdateTurn(ctx context.Context, turn *models.Turn) erro func (m *mockRepository) CompleteTurn(ctx context.Context, id string) error { return nil } +func (m *mockRepository) AbandonTurn(ctx context.Context, id string) error { + return nil +} func (m *mockRepository) CompletePendingToolCallsForTurn(ctx context.Context, turnID string) (int64, error) { return 0, nil } diff --git a/apps/backend/internal/task/repository/interface.go b/apps/backend/internal/task/repository/interface.go index 7d8ea057bd..31e5f63e8d 100644 --- a/apps/backend/internal/task/repository/interface.go +++ b/apps/backend/internal/task/repository/interface.go @@ -82,6 +82,7 @@ type TurnRepository interface { GetActiveTurnBySessionID(ctx context.Context, sessionID string) (*models.Turn, error) UpdateTurn(ctx context.Context, turn *models.Turn) error CompleteTurn(ctx context.Context, id string) error + AbandonTurn(ctx context.Context, id string) error CompletePendingToolCallsForTurn(ctx context.Context, turnID string) (int64, error) ListTurnsBySession(ctx context.Context, sessionID string) ([]*models.Turn, error) } diff --git a/apps/backend/internal/task/repository/sqlite/session.go b/apps/backend/internal/task/repository/sqlite/session.go index 9e5f43c48a..cc5d3f5db8 100644 --- a/apps/backend/internal/task/repository/sqlite/session.go +++ b/apps/backend/internal/task/repository/sqlite/session.go @@ -122,6 +122,21 @@ func (r *Repository) CompleteTurn(ctx context.Context, id string) error { return err } +// AbandonTurn marks a turn as completed with completed_at = started_at, giving it +// zero duration. Used when a turn was orphaned by an interruption (backend +// restart, agent crash) and the previous "running" window was not real work — +// recording `now` would inflate analytics and the UI's last-turn duration with +// hours of dead time. +func (r *Repository) AbandonTurn(ctx context.Context, id string) error { + now := time.Now().UTC() + _, err := r.db.ExecContext(ctx, r.db.Rebind(` + UPDATE task_session_turns + SET completed_at = started_at, updated_at = ? + WHERE id = ? AND completed_at IS NULL + `), now, id) + return err +} + // ListTurnsBySession returns all turns for a session ordered by start time func (r *Repository) ListTurnsBySession(ctx context.Context, sessionID string) ([]*models.Turn, error) { ctx, span := tracing.Tracer("kandev-db").Start(ctx, "db.ListTurnsBySession") diff --git a/apps/backend/internal/task/service/service_turns.go b/apps/backend/internal/task/service/service_turns.go index 07e4d595e7..579e4b41f4 100644 --- a/apps/backend/internal/task/service/service_turns.go +++ b/apps/backend/internal/task/service/service_turns.go @@ -100,6 +100,73 @@ func (s *Service) GetActiveTurn(ctx context.Context, sessionID string) (*models. return turn, err } +// AbandonOpenTurns closes any open turns for a session by setting their +// completed_at = started_at. Used on session resume to bury orphan turns left +// behind by a previous crash/restart so a fresh prompt starts a fresh turn +// (preventing the UI from showing the orphan's stale started_at as the running +// timer, and preventing the orphan from poisoning analytics with hours of dead +// time). +// +// Mirrors the iteration shape of orchestrator.completeTurnForSession: the DB is +// authoritative, so we loop GetActiveTurn → AbandonTurn until none remain, with +// a sanity cap to break runaway loops. +func (s *Service) AbandonOpenTurns(ctx context.Context, sessionID string) error { + const maxIterations = 16 + closed := 0 + for closed < maxIterations { + turn, err := s.GetActiveTurn(ctx, sessionID) + if err != nil { + return err + } + if turn == nil { + return nil + } + if err := s.turns.AbandonTurn(ctx, turn.ID); err != nil { + s.logger.Error("failed to abandon turn", + zap.String("turn_id", turn.ID), + zap.String("session_id", sessionID), + zap.Error(err)) + return err + } + // Same safety net as CompleteTurn: any tool calls left in a non-terminal + // state belong to a turn that is no longer running, so flip them to + // "complete" instead of leaving them to spin forever in the UI. + if affected, err := s.turns.CompletePendingToolCallsForTurn(ctx, turn.ID); err != nil { + s.logger.Warn("failed to complete pending tool calls for abandoned turn", + zap.String("turn_id", turn.ID), + zap.Error(err)) + } else if affected > 0 { + s.logger.Info("completed pending tool calls on abandoned turn", + zap.String("turn_id", turn.ID), + zap.Int64("affected", affected)) + } + // Re-fetch so the published event carries the persisted completed_at + // (= started_at) rather than a synthesized one. + refreshed, err := s.turns.GetTurn(ctx, turn.ID) + if err != nil { + s.logger.Debug("failed to refetch abandoned turn", + zap.String("turn_id", turn.ID), + zap.Error(err)) + } else { + s.publishTurnEvent(events.TurnCompleted, refreshed) + } + s.logger.Info("abandoned orphan turn on session resume", + zap.String("turn_id", turn.ID), + zap.String("session_id", sessionID)) + closed++ + } + // Only warn if turns are *still* accumulating after the cap. Closing + // exactly maxIterations turns and then finding the session clean is not a + // runaway — same shape as completeTurnForSession. + if turn, err := s.GetActiveTurn(ctx, sessionID); err == nil && turn != nil { + s.logger.Warn("AbandonOpenTurns hit iteration cap; some orphan turns may remain", + zap.String("session_id", sessionID), + zap.Int("closed", closed), + zap.Int("max_iterations", maxIterations)) + } + return nil +} + // getOrStartTurn returns the active turn for a session, or starts a new one if none exists. // This is used to ensure messages always have a valid turn ID. func (s *Service) getOrStartTurn(ctx context.Context, sessionID string) (*models.Turn, error) { diff --git a/apps/backend/internal/task/service/service_turns_test.go b/apps/backend/internal/task/service/service_turns_test.go index f91cdb0e70..b1e4cd85ad 100644 --- a/apps/backend/internal/task/service/service_turns_test.go +++ b/apps/backend/internal/task/service/service_turns_test.go @@ -6,6 +6,7 @@ import ( "testing" "time" + "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/task/models" ) @@ -316,3 +317,203 @@ func TestGetWorkspaceInfoForSession_MultiRepoReturnsTaskRoot(t *testing.T) { info.WorkspacePath) } } + +// AbandonOpenTurns: turns left open by a previous crash must close with +// completed_at = started_at so analytics' active_duration_ms doesn't get +// poisoned with the dead window and the UI's running timer doesn't count +// from a stale start. +func TestAbandonOpenTurns_ZeroesDuration(t *testing.T) { + svc, _, repo := createTestService(t) + ctx := context.Background() + + setupTestTask(t, repo) + sessionID := setupTestSession(t, repo) + + stale := &models.Turn{ + ID: "turn-stale", + TaskSessionID: sessionID, + TaskID: "task-123", + StartedAt: time.Now().Add(-90 * time.Hour).UTC(), + } + if err := repo.CreateTurn(ctx, stale); err != nil { + t.Fatalf("CreateTurn: %v", err) + } + + if err := svc.AbandonOpenTurns(ctx, sessionID); err != nil { + t.Fatalf("AbandonOpenTurns: %v", err) + } + + got, err := repo.GetTurn(ctx, stale.ID) + if err != nil { + t.Fatalf("GetTurn: %v", err) + } + if got.CompletedAt == nil { + t.Fatal("expected completed_at to be set") + } + if !got.CompletedAt.Equal(got.StartedAt) { + t.Fatalf("expected completed_at == started_at (zero duration), got started=%v completed=%v", + got.StartedAt, *got.CompletedAt) + } +} + +// AbandonOpenTurns sweeps the same pending tool calls that CompleteTurn does: +// otherwise the UI would show "running" tool calls forever on an abandoned turn. +func TestAbandonOpenTurns_CompletesPendingToolCalls(t *testing.T) { + svc, _, repo := createTestService(t) + ctx := context.Background() + + setupTestTask(t, repo) + sessionID := setupTestSession(t, repo) + + turn := &models.Turn{ + ID: "turn-with-tool", + TaskSessionID: sessionID, + TaskID: "task-123", + StartedAt: time.Now().UTC(), + } + if err := repo.CreateTurn(ctx, turn); err != nil { + t.Fatalf("CreateTurn: %v", err) + } + + toolMsg := &models.Message{ + ID: "msg-tool-1", + TaskSessionID: sessionID, + TaskID: "task-123", + TurnID: turn.ID, + AuthorType: models.MessageAuthorAgent, + Type: models.MessageTypeToolCall, + Content: "running tool", + Metadata: map[string]interface{}{ + "tool_call_id": "tc-1", + "status": "running", + }, + } + if err := repo.CreateMessage(ctx, toolMsg); err != nil { + t.Fatalf("CreateMessage: %v", err) + } + + if err := svc.AbandonOpenTurns(ctx, sessionID); err != nil { + t.Fatalf("AbandonOpenTurns: %v", err) + } + + got, err := repo.GetMessageByToolCallID(ctx, sessionID, "tc-1") + if err != nil { + t.Fatalf("GetMessageByToolCallID: %v", err) + } + if got == nil { + t.Fatal("expected tool call message to exist") + } + if status, _ := got.Metadata["status"].(string); status != "complete" { + t.Fatalf("expected tool call status='complete', got %q", status) + } +} + +// AbandonOpenTurns publishes turn.completed for each closed turn so the WS +// gateway can broadcast the state change and the frontend clears its +// activeBySession entry. +func TestAbandonOpenTurns_PublishesTurnCompletedEvent(t *testing.T) { + svc, eventBus, repo := createTestService(t) + ctx := context.Background() + + setupTestTask(t, repo) + sessionID := setupTestSession(t, repo) + + turn := &models.Turn{ + ID: "turn-event", + TaskSessionID: sessionID, + TaskID: "task-123", + StartedAt: time.Now().UTC(), + } + if err := repo.CreateTurn(ctx, turn); err != nil { + t.Fatalf("CreateTurn: %v", err) + } + + eventBus.ClearEvents() + if err := svc.AbandonOpenTurns(ctx, sessionID); err != nil { + t.Fatalf("AbandonOpenTurns: %v", err) + } + + var found bool + for _, ev := range eventBus.GetPublishedEvents() { + if ev.Type == events.TurnCompleted { + found = true + break + } + } + if !found { + t.Fatal("expected turn.completed event to be published") + } +} + +// AbandonOpenTurns is a no-op when no open turn exists (e.g. resume called on +// a session whose previous turn already completed cleanly). +func TestAbandonOpenTurns_NoOpenTurns(t *testing.T) { + svc, eventBus, repo := createTestService(t) + ctx := context.Background() + + setupTestTask(t, repo) + sessionID := setupTestSession(t, repo) + + eventBus.ClearEvents() + if err := svc.AbandonOpenTurns(ctx, sessionID); err != nil { + t.Fatalf("AbandonOpenTurns: %v", err) + } + + if got := len(eventBus.GetPublishedEvents()); got != 0 { + t.Fatalf("expected no events when nothing to abandon, got %d", got) + } +} + +// AbandonOpenTurns iteration cap: with more than maxIterations open turns, +// the function still returns nil (caller swallows the error and the next +// completeTurnForSession sweep mops up the remainder). +func TestAbandonOpenTurns_IterationCap(t *testing.T) { + svc, _, repo := createTestService(t) + ctx := context.Background() + + setupTestTask(t, repo) + sessionID := setupTestSession(t, repo) + + const seeded = 20 // > maxIterations (16) + for i := 0; i < seeded; i++ { + turn := &models.Turn{ + ID: fmt.Sprintf("turn-cap-%d", i), + TaskSessionID: sessionID, + TaskID: "task-123", + StartedAt: time.Now().UTC(), + } + if err := repo.CreateTurn(ctx, turn); err != nil { + t.Fatalf("seed turn %d: %v", i, err) + } + } + + if err := svc.AbandonOpenTurns(ctx, sessionID); err != nil { + t.Fatalf("AbandonOpenTurns: %v", err) + } + + // First sweep closes exactly maxIterations; remainder still open. + turns, err := repo.ListTurnsBySession(ctx, sessionID) + if err != nil { + t.Fatalf("ListTurnsBySession: %v", err) + } + open := 0 + for _, turn := range turns { + if turn.CompletedAt == nil { + open++ + } + } + if open != seeded-16 { + t.Fatalf("expected %d open turns after first sweep (cap=16), got %d", seeded-16, open) + } + + // Second sweep finishes the job. + if err := svc.AbandonOpenTurns(ctx, sessionID); err != nil { + t.Fatalf("second AbandonOpenTurns: %v", err) + } + turns, _ = repo.ListTurnsBySession(ctx, sessionID) + for _, turn := range turns { + if turn.CompletedAt == nil { + t.Fatalf("expected all turns closed after second sweep, %s still open", turn.ID) + } + } +}