From ce6bd5986dcdb4ece30b5967b2d4fa974c35f027 Mon Sep 17 00:00:00 2001 From: Alassane <79718216+yattdev@users.noreply.github.com> Date: Tue, 18 Aug 2026 18:47:30 -0400 Subject: [PATCH] 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 Co-authored-by: Carlos Florencio --- apps/backend/internal/backendapp/adapters.go | 16 ++ .../backendapp/adapters_acp_session_test.go | 84 ++++++ .../orchestrator/event_handlers_workflow.go | 34 +++ ...vent_handlers_workflow_lazy_resume_test.go | 242 ++++++++++++++++++ 4 files changed, 376 insertions(+) create mode 100644 apps/backend/internal/backendapp/adapters_acp_session_test.go diff --git a/apps/backend/internal/backendapp/adapters.go b/apps/backend/internal/backendapp/adapters.go index 3f1a7ea7e3..bc107632aa 100644 --- a/apps/backend/internal/backendapp/adapters.go +++ b/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) } diff --git a/apps/backend/internal/backendapp/adapters_acp_session_test.go b/apps/backend/internal/backendapp/adapters_acp_session_test.go new file mode 100644 index 0000000000..2e5f986206 --- /dev/null +++ b/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") + } +} diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index 47e96c5a4a..54d7c08904 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/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 diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow_lazy_resume_test.go b/apps/backend/internal/orchestrator/event_handlers_workflow_lazy_resume_test.go index 8643819bcb..0fdd15d4d5 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow_lazy_resume_test.go +++ b/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) + } +}