fix(orchestrator): abandon orphan turns on session resume (#837)

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

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

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

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)

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

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
 }

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

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

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

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