fix(workflow): reconcile resume token when a context reset partially succeeds, and wire the live-ACP guard so it actually runs (#2788)

Co-authored-by: ayattara <alassane.yattara@savoirfairelinux.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>

apps/backend/internal/backendapp/adapters.go

@@ -96,9 +96,17 @@ type lifecycleAdapter struct {
 	logger   *logger.Logger
 }
 
+// orchestrator.Service.currentACPSessionID selects the generation-safety seam
+// for resetAgentContext by asserting the agent manager to this unexported
+// interface; that assertion fails silently at runtime if this method is
+// missing or its signature drifts, leaving storeResumeToken's stale-event
+// guard and the reset-failure reconcile permanently inert against a defunct
+// or dead-wrong ACP session id. Pin the exact shape here so drift is a build
+// error, not a silently-dead production guard.
 var _ interface {
 	OwnsPromptGeneration(sessionID, executionID string, generation uint64) bool
 	GetPromptGenerationForSession(ctx context.Context, sessionID string) (uint64, error)
+	GetACPSessionIDForSession(sessionID string) (string, bool)
 } = (*lifecycleAdapter)(nil)
 
 // newLifecycleAdapter creates a new lifecycle adapter
@@ -449,6 +457,14 @@ func (a *lifecycleAdapter) GetPromptGenerationForSession(ctx context.Context, se
 	return a.mgr.GetPromptGenerationForSession(ctx, sessionID)
 }
 
+// GetACPSessionIDForSession forwards to the lifecycle manager so
+// orchestrator.Service.currentACPSessionID observes the live execution's
+// actual ACP session identity in production, not just in tests against
+// mockAgentManager (see the compile-time assertion above this type).
+func (a *lifecycleAdapter) GetACPSessionIDForSession(sessionID string) (string, bool) {
+	return a.mgr.GetACPSessionIDForSession(sessionID)
+}
+
 func (a *lifecycleAdapter) GetSessionAuthMethods(sessionID string) []streams.AuthMethodInfo {
 	return a.mgr.GetSessionAuthMethods(sessionID)
 }

apps/backend/internal/backendapp/adapters_acp_session_test.go

@@ -0,0 +1,84 @@
+package backendapp
+
+import (
+	"testing"
+
+	"github.com/kandev/kandev/internal/agent/runtime/lifecycle"
+	orchestratorexecutor "github.com/kandev/kandev/internal/orchestrator/executor"
+)
+
+// acpSessionIDProvider mirrors the unexported interface
+// orchestrator.Service.currentACPSessionID asserts s.agentManager against.
+// Declaring it independently here — instead of importing an orchestrator
+// helper — proves the production *lifecycleAdapter satisfies the shape
+// orchestrator actually needs, not a copy orchestrator happens to export.
+type acpSessionIDProvider interface {
+	GetACPSessionIDForSession(sessionID string) (string, bool)
+}
+
+// TestLifecycleAdapter_SatisfiesACPSessionIDSeam is the regression test for a
+// QA-caught production gap: orchestrator.Service.currentACPSessionID selects
+// its generation-safety seam with an optional type assertion
+// (agentManager.(interface{ GetACPSessionIDForSession(string) (string, bool) })).
+// mockAgentManager implements that method directly, so orchestrator package
+// tests exercised a stronger seam than production ever wired: the concrete
+// production type assigned to orchestrator.Service's agentManager field —
+// *backendapp.lifecycleAdapter, built by newLifecycleAdapter exactly as
+// setupOrchestrator does — did not forward the method, so the assertion
+// failed silently at runtime and storeResumeToken's stale-event guard plus
+// the reset-failure reconcile (event_handlers_workflow.go) were dead code in
+// production despite green unit tests. This asserts the real adapter type,
+// held through the same executor.AgentManagerClient interface production
+// wires it as, satisfies the seam.
+func TestLifecycleAdapter_SatisfiesACPSessionIDSeam(t *testing.T) {
+	mgr := lifecycle.NewManager(nil, nil, nil, nil, nil, nil, lifecycle.ExecutorFallbackDeny, t.TempDir(), newTestLogger())
+	var client orchestratorexecutor.AgentManagerClient = newLifecycleAdapter(mgr, nil, newTestLogger())
+
+	provider, ok := client.(acpSessionIDProvider)
+	if !ok {
+		t.Fatal("production lifecycleAdapter, held as executor.AgentManagerClient, does not satisfy " +
+			"GetACPSessionIDForSession(string) (string, bool) — the same assertion orchestrator.Service." +
+			"currentACPSessionID performs would fail silently and disable the generation-safety guard")
+	}
+
+	if acpSessionID, found := provider.GetACPSessionIDForSession("no-such-session"); found {
+		t.Fatalf("expected (\"\", false) for an unknown session, got (%q, %v)", acpSessionID, found)
+	}
+}
+
+// TestLifecycleAdapter_GetACPSessionIDForSession_ForwardsLiveIdentity proves
+// the forwarder itself observes the live execution's current ACP session id
+// through a real *lifecycle.Manager — not a mock's independently-configured
+// return value. Seeds an execution directly into the manager's execution
+// store (the same seam lifecycle package tests use), then reads it back
+// through the adapter exactly as orchestrator.Service.currentACPSessionID
+// would.
+func TestLifecycleAdapter_GetACPSessionIDForSession_ForwardsLiveIdentity(t *testing.T) {
+	mgr := lifecycle.NewManager(nil, nil, nil, nil, nil, nil, lifecycle.ExecutorFallbackDeny, t.TempDir(), newTestLogger())
+	adapter := newLifecycleAdapter(mgr, nil, newTestLogger())
+
+	const sessionID = "sess-live-acp"
+	const liveACPSessionID = "acp-session-currently-owned-by-the-execution"
+	if err := mgr.ExecutionStoreForTesting().Add(&lifecycle.AgentExecution{
+		ID:            "exec-live-acp",
+		TaskID:        "task-1",
+		SessionID:     sessionID,
+		ACPSessionID:  liveACPSessionID,
+		WorkspacePath: t.TempDir(),
+	}); err != nil {
+		t.Fatalf("seed execution: %v", err)
+	}
+
+	acpSessionID, ok := adapter.GetACPSessionIDForSession(sessionID)
+	if !ok {
+		t.Fatal("expected the adapter to report the live execution's ACP session id")
+	}
+	if acpSessionID != liveACPSessionID {
+		t.Fatalf("adapter did not forward the manager's live identity: got %q, want %q",
+			acpSessionID, liveACPSessionID)
+	}
+
+	if _, ok := adapter.GetACPSessionIDForSession("no-such-session"); ok {
+		t.Fatal("expected (_, false) for a session with no registered execution")
+	}
+}

apps/backend/internal/orchestrator/event_handlers_workflow.go

@@ -3254,12 +3254,14 @@ func (s *Service) resetAgentContext(ctx context.Context, taskID string, session
 		zap.String("step_name", stepName),
 		zap.String("agent_execution_id", executionID))
 
+	previousACPSessionID := s.currentACPSessionID(sessionID)
 	if err := s.agentManager.ResetAgentContext(ctx, executionID); err != nil {
 		s.logger.Error("failed to reset agent context",
 			zap.String("task_id", taskID),
 			zap.String("session_id", sessionID),
 			zap.String("step_name", stepName),
 			zap.Error(err))
+		s.reconcileFailedContextReset(ctx, taskID, session, executionID, previousACPSessionID)
 		return false
 	}
 
@@ -3285,6 +3287,38 @@ func (s *Service) resetAgentContext(ctx context.Context, taskID string, session
 	return true
 }
 
+// reconcileFailedContextReset handles a reset that moved the live ACP session
+// before a later reset step failed. The old cursor and context metadata belong
+// to the previous ACP session and must not survive that partial transition.
+func (s *Service) reconcileFailedContextReset(
+	ctx context.Context,
+	taskID string,
+	session *models.TaskSession,
+	executionID string,
+	previousACPSessionID string,
+) {
+	liveACPSessionID := s.currentACPSessionID(session.ID)
+	if liveACPSessionID == "" || liveACPSessionID == previousACPSessionID {
+		return
+	}
+
+	running, err := s.repo.GetExecutorRunningBySessionID(ctx, session.ID)
+	if err != nil {
+		s.logger.Warn("failed to inspect execution after partial context reset",
+			zap.String("task_id", taskID),
+			zap.String("session_id", session.ID),
+			zap.String("agent_execution_id", executionID),
+			zap.Error(err))
+		return
+	}
+	if running.AgentExecutionID != executionID {
+		return
+	}
+
+	s.storeResumeToken(ctx, taskID, session.ID, executionID, liveACPSessionID, "")
+	s.clearPersistedResetState(ctx, session.ID, session)
+}
+
 // clearPersistedResetState clears the durable, DB-backed session state that a
 // later lazy resume would otherwise pick back up: the stored ACP session ID
 // in session metadata and the persisted context window. The resume token is

apps/backend/internal/orchestrator/event_handlers_workflow_lazy_resume_test.go

@@ -281,3 +281,245 @@ func TestResetAgentContext_InterleavingC_StaleOldEventDoesNotOverwriteFresh(t *t
 		t.Fatalf("stale old-session event overwrote fresh token: got %q, want %q", running.ResumeToken, freshToken)
 	}
 }
+
+// TestResetAgentContext_FailureAfterLiveSessionMovedReconcilesState covers a
+// deeper shape of "preserve recoverability on ResetAgentContext failure" than
+// TestResetAgentContext_FailedProviderResetRetainsResumeToken: the lifecycle
+// manager's ResetAgentContext can commit the execution's live ACP session to
+// a NEW id (ResetSession itself succeeded) and only fail afterward — e.g. a
+// later re-apply-session-model step erroring — so it still returns a
+// non-nil error overall. Leaving the persisted resume_token at its pre-reset
+// value in that case does not preserve a resumable session: the agent has
+// already moved on and no longer recognizes the old ACP session, so the
+// stale token can never be resumed. resetAgentContext must reconcile the
+// persisted token to the live truth even when it reports failure.
+func TestResetAgentContext_FailureAfterLiveSessionMovedReconcilesState(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "step1")
+
+	const execID = "exec-1"
+	const staleToken = "old-acp-session"
+	const movedToken = "new-acp-session-committed-before-failure"
+	seedExecutorRunning(t, repo, "s1", "t1", execID)
+	if err := repo.UpdateResumeToken(ctx, "s1", execID, staleToken, ""); err != nil {
+		t.Fatalf("seed stale resume token: %v", err)
+	}
+	if err := repo.SetSessionMetadataKey(ctx, "s1", contextWindowMetadataKey, map[string]interface{}{
+		"size": int64(200000), "used": int64(190000),
+	}); err != nil {
+		t.Fatalf("seed context window: %v", err)
+	}
+
+	var agentManager *mockAgentManager
+	agentManager = &mockAgentManager{
+		repoForExecutionLookup: repo,
+		restartProcessErr:      errors.New("model reapplication failed after session reset"),
+		getACPSessionIDForSessionFunc: func(string) (string, bool) {
+			if len(agentManager.restartProcessCalls) == 0 {
+				return staleToken, true
+			}
+			return movedToken, true
+		},
+	}
+	svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentManager)
+	staleContextGeneration := svc.captureContextWindowGeneration("s1")
+	session, err := repo.GetTaskSession(ctx, "s1")
+	if err != nil {
+		t.Fatalf("load session: %v", err)
+	}
+
+	if svc.resetAgentContext(ctx, "t1", session, "test") {
+		t.Fatal("expected reset to report failure")
+	}
+
+	running, err := repo.GetExecutorRunningBySessionID(ctx, "s1")
+	if err != nil {
+		t.Fatalf("load executor row: %v", err)
+	}
+	if running.ResumeToken != movedToken {
+		t.Fatalf("failed reset left a dead token persisted: got %q, want live session %q",
+			running.ResumeToken, movedToken)
+	}
+	updated, err := repo.GetTaskSession(ctx, "s1")
+	if err != nil {
+		t.Fatalf("load session after failed reset: %v", err)
+	}
+	if updated.Metadata[contextWindowMetadataKey] != nil {
+		t.Fatalf("partial reset retained context from the old ACP session: %#v",
+			updated.Metadata[contextWindowMetadataKey])
+	}
+	persisted, _, err := svc.persistContextWindowUpdate(ctx, "s1", staleContextGeneration, map[string]interface{}{
+		"size": int64(200000), "used": int64(195000),
+	})
+	if err != nil {
+		t.Fatalf("persist stale context update: %v", err)
+	}
+	if persisted {
+		t.Fatal("context update captured before the partial reset was persisted")
+	}
+}
+
+// TestResetAgentContext_FailureWithUnchangedLiveSessionPreservesToken pins the
+// other half of the reconcile contract: when ResetAgentContext fails and the
+// live ACP session did NOT move, the still-valid token and recovery cursor must
+// survive. The observable guarantee from #2765 ("a failed reset keeps a usable
+// recovery token") is unchanged. Without this test, reconciling unchanged state
+// would blank a valid cursor and no test would notice.
+func TestResetAgentContext_FailureWithUnchangedLiveSessionPreservesToken(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "step1")
+
+	const execID = "exec-1"
+	const liveToken = "acp-session-that-never-moved"
+	const lastMessageUUID = "last-message-before-failed-reset"
+	seedExecutorRunning(t, repo, "s1", "t1", execID)
+	if err := repo.UpdateResumeToken(ctx, "s1", execID, liveToken, lastMessageUUID); err != nil {
+		t.Fatalf("seed resume token: %v", err)
+	}
+
+	agentManager := &mockAgentManager{
+		repoForExecutionLookup: repo,
+		restartProcessErr:      errors.New("provider reset failed before touching the session"),
+		getACPSessionIDForSessionFunc: func(string) (string, bool) {
+			return liveToken, true
+		},
+	}
+	svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentManager)
+	session, err := repo.GetTaskSession(ctx, "s1")
+	if err != nil {
+		t.Fatalf("load session: %v", err)
+	}
+
+	if svc.resetAgentContext(ctx, "t1", session, "test") {
+		t.Fatal("expected reset to report failure")
+	}
+
+	running, err := repo.GetExecutorRunningBySessionID(ctx, "s1")
+	if err != nil {
+		t.Fatalf("load executor row: %v", err)
+	}
+	if running.ResumeToken != liveToken {
+		t.Fatalf("failed reset destroyed a still-valid token: got %q, want %q",
+			running.ResumeToken, liveToken)
+	}
+	if running.LastMessageUUID != lastMessageUUID {
+		t.Fatalf("failed reset destroyed the recovery cursor: got %q, want %q",
+			running.LastMessageUUID, lastMessageUUID)
+	}
+}
+
+// TestResetAgentContext_FailureFromRotatedExecutionDoesNotOverwrite proves that
+// reconciliation stays scoped to the execution that attempted the reset. If
+// the executors_running row has already rotated, the failed reset belongs to a
+// defunct execution and must not change the live execution's state.
+func TestResetAgentContext_FailureFromRotatedExecutionDoesNotOverwrite(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "step1")
+
+	const rotatedInExecID = "exec-new"
+	const staleExecID = "exec-old"
+	const liveExecToken = "token-owned-by-new-execution"
+
+	// The row has already rotated to the newer execution and carries its token.
+	seedExecutorRunning(t, repo, "s1", "t1", rotatedInExecID)
+	if err := repo.UpdateResumeToken(ctx, "s1", rotatedInExecID, liveExecToken, ""); err != nil {
+		t.Fatalf("seed resume token: %v", err)
+	}
+
+	var agentManager *mockAgentManager
+	agentManager = &mockAgentManager{
+		// The reset is still operating on the older execution.
+		getExecutionIDForSessionFunc: func(context.Context, string) (string, error) {
+			return staleExecID, nil
+		},
+		restartProcessErr: errors.New("model reapplication failed after session reset"),
+		getACPSessionIDForSessionFunc: func(string) (string, bool) {
+			if len(agentManager.restartProcessCalls) == 0 {
+				return "acp-session-before-defunct-reset", true
+			}
+			return "acp-session-of-defunct-execution", true
+		},
+	}
+	svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentManager)
+	session, err := repo.GetTaskSession(ctx, "s1")
+	if err != nil {
+		t.Fatalf("load session: %v", err)
+	}
+
+	if svc.resetAgentContext(ctx, "t1", session, "test") {
+		t.Fatal("expected reset to report failure")
+	}
+
+	running, err := repo.GetExecutorRunningBySessionID(ctx, "s1")
+	if err != nil {
+		t.Fatalf("load executor row: %v", err)
+	}
+	if running.ResumeToken != liveExecToken {
+		t.Fatalf("defunct execution overwrote the live execution's token: got %q, want %q",
+			running.ResumeToken, liveExecToken)
+	}
+}
+
+// TestResetAgentContext_FailureReconcileDroppedWhenSessionMovesMidFlight models
+// the race the reconcile write is exposed to: the ACP session can move again
+// between resetAgentContext reading the live id and storeResumeToken
+// re-reading it for its stale-generation guard. Staged deterministically by
+// returning a different id on each lookup — the first is what the reconcile
+// tries to persist, the second is the newer generation the guard compares
+// against. The reconcile must lose: a fresher generation already owns the
+// session, and its own event will persist the correct token.
+func TestResetAgentContext_FailureReconcileDroppedWhenSessionMovesMidFlight(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "step1")
+
+	const execID = "exec-1"
+	const seededToken = "acp-session-before-reset"
+	seedExecutorRunning(t, repo, "s1", "t1", execID)
+	if err := repo.UpdateResumeToken(ctx, "s1", execID, seededToken, ""); err != nil {
+		t.Fatalf("seed resume token: %v", err)
+	}
+
+	var lookups int
+	agentManager := &mockAgentManager{
+		repoForExecutionLookup: repo,
+		restartProcessErr:      errors.New("model reapplication failed after session reset"),
+		getACPSessionIDForSessionFunc: func(string) (string, bool) {
+			lookups++
+			switch lookups {
+			case 1:
+				return seededToken, true
+			case 2:
+				// What resetAgentContext observes and tries to reconcile to.
+				return "acp-session-observed-by-reconcile", true
+			default:
+				// A newer generation took over before the guard re-read it.
+				return "acp-session-from-newer-generation", true
+			}
+		},
+	}
+	svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentManager)
+	session, err := repo.GetTaskSession(ctx, "s1")
+	if err != nil {
+		t.Fatalf("load session: %v", err)
+	}
+
+	if svc.resetAgentContext(ctx, "t1", session, "test") {
+		t.Fatal("expected reset to report failure")
+	}
+	if lookups < 3 {
+		t.Fatalf("expected the reconcile write to be re-checked against the live id, got %d lookups", lookups)
+	}
+
+	running, err := repo.GetExecutorRunningBySessionID(ctx, "s1")
+	if err != nil {
+		t.Fatalf("load executor row: %v", err)
+	}
+	if running.ResumeToken != seededToken {
+		t.Fatalf("stale reconcile beat a newer ACP session generation: got %q, want %q",
+			running.ResumeToken, seededToken)
+	}
+}