diff --git a/apps/backend/internal/agent/runtime/agentctl/agent_session_test.go b/apps/backend/internal/agent/runtime/agentctl/agent_session_test.go index 28cf8d1941..a92c3fd23c 100644 --- a/apps/backend/internal/agent/runtime/agentctl/agent_session_test.go +++ b/apps/backend/internal/agent/runtime/agentctl/agent_session_test.go @@ -122,6 +122,62 @@ func TestResetSession_FailureModes(t *testing.T) { }) } +func TestResetSession_ContextCancellationRemovesPendingRequestBeforeLateResponse(t *testing.T) { + received := make(chan ws.Message) + releaseResponse := make(chan struct{}) + c, ts := newTestClientWithStream(t, func(msg ws.Message) *ws.Message { + received <- msg + <-releaseResponse + resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]any{ + "success": true, "session_id": "late-session", + }) + return resp + }) + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(releaseResponse) }) } + t.Cleanup(func() { + release() + c.Close() + ts.Close() + }) + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + result := make(chan error, 1) + go func() { + _, err := c.ResetSession(ctx, "/workspace", nil) + result <- err + }() + + select { + case msg := <-received: + if msg.Action != "agent.session.reset" { + t.Fatalf("request action = %q, want agent.session.reset", msg.Action) + } + case <-time.After(time.Second): + t.Fatal("timed out waiting for reset request") + } + + select { + case err := <-result: + if err == nil || !strings.Contains(err.Error(), "context deadline exceeded") { + t.Fatalf("ResetSession error = %v, want context deadline exceeded", err) + } + case <-time.After(time.Second): + t.Fatal("ResetSession did not stop waiting after context cancellation") + } + + c.pendingMu.Lock() + pendingRequests := len(c.pendingRequests) + pendingConnections := len(c.pendingRequestConns) + c.pendingMu.Unlock() + if pendingRequests != 0 || pendingConnections != 0 { + t.Fatalf("pending reset request state = %d requests, %d connections, want both zero", pendingRequests, pendingConnections) + } + + release() +} + // SetMode / SetModel / SetConfigOption / Authenticate share one shape: send a // typed payload, treat an error frame as a failure, ignore the response body. func TestAgentSessionSetters_SendExpectedActionAndPayload(t *testing.T) { diff --git a/apps/backend/internal/agent/runtime/errors.go b/apps/backend/internal/agent/runtime/errors.go index 3280fc5f94..b8b459a625 100644 --- a/apps/backend/internal/agent/runtime/errors.go +++ b/apps/backend/internal/agent/runtime/errors.go @@ -9,3 +9,7 @@ type BootstrapFailure = lifecycle.BootstrapFailure // RepositoryPreparationError identifies the repository whose preparation // prevented a multi-repository launch. type RepositoryPreparationError = lifecycle.RepositoryPreparationError + +// ErrCancelEscalated reports that cancellation released local admission after +// the provider failed to acknowledge the cancellation within its bound. +var ErrCancelEscalated = lifecycle.ErrCancelEscalated diff --git a/apps/backend/internal/agent/runtime/lifecycle/execution_context_reset.go b/apps/backend/internal/agent/runtime/lifecycle/execution_context_reset.go new file mode 100644 index 0000000000..36fe83f022 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/execution_context_reset.go @@ -0,0 +1,203 @@ +package lifecycle + +import ( + "context" + "errors" + "time" + + agentctl "github.com/kandev/kandev/internal/agent/runtime/agentctl" + "github.com/kandev/kandev/internal/agentctl/types/streams" +) + +// ErrContextResetFenced means a reset attempt may still have provider-side +// effects, so prompt and session-configuration admission is closed until an +// explicit process restart succeeds. +var ErrContextResetFenced = errors.New("agent context reset requires explicit recovery") + +var errContextResetInProgress = errors.New("agent context reset is in progress") + +const maxContextResetEvents = 64 + +const contextResetLeasePoll = 10 * time.Millisecond + +func waitForContextResetLease(ctx context.Context, try func() bool) error { + if ctx == nil { + ctx = context.Background() + } + for { + if err := ctx.Err(); err != nil { + return err + } + if try() { + return nil + } + timer := time.NewTimer(contextResetLeasePoll) + select { + case <-ctx.Done(): + if !timer.Stop() { + <-timer.C + } + return ctx.Err() + case <-timer.C: + } + } +} + +func (e *AgentExecution) acquireContextResetOperation(ctx context.Context) (func(), error) { + if err := waitForContextResetLease(ctx, e.contextResetAdmissionMu.TryRLock); err != nil { + return nil, err + } + if err := e.contextResetAdmissionError(); err != nil { + e.contextResetAdmissionMu.RUnlock() + return nil, err + } + return e.contextResetAdmissionMu.RUnlock, nil +} + +func (e *AgentExecution) acquireContextResetExclusive(ctx context.Context) (func(), error) { + if err := waitForContextResetLease(ctx, e.contextResetAdmissionMu.TryLock); err != nil { + return nil, err + } + return e.contextResetAdmissionMu.Unlock, nil +} + +func (e *AgentExecution) beginContextReset() error { + e.contextResetMu.Lock() + defer e.contextResetMu.Unlock() + if e.contextResetFenced { + return ErrContextResetFenced + } + if e.contextResetInFlight { + return errContextResetInProgress + } + e.contextResetInFlight = true + e.contextResetEvents = nil + return nil +} + +// drainContextResetEvents returns only setup +// events for the session that the caller just committed. Events from the old +// session can be queued while session/new is waiting and must never be replayed +// into the replacement conversation. +func (e *AgentExecution) drainContextResetEvents(newSessionID string) []agentctl.AgentEvent { + e.contextResetMu.Lock() + defer e.contextResetMu.Unlock() + buffered := e.contextResetEvents + e.contextResetEvents = nil + if len(buffered) == 0 { + return nil + } + accepted := make([]agentctl.AgentEvent, 0, len(buffered)) + for _, event := range buffered { + if event.SessionID == "" || event.SessionID == newSessionID { + accepted = append(accepted, event) + } + } + return accepted +} + +// finishContextReset closes the reset boundary and returns setup events that +// arrived after the last explicit drain. The transition and snapshot are one +// critical section: an event handler that observed the in-flight boundary is +// either included in the returned batch or, after the transition, proceeds +// through ordinary lifecycle handling. No buffered event is discarded at the +// boundary. +func (e *AgentExecution) finishContextReset(newSessionID string) []agentctl.AgentEvent { + e.contextResetMu.Lock() + buffered := e.contextResetEvents + e.contextResetEvents = nil + e.contextResetInFlight = false + e.contextResetMu.Unlock() + + accepted := make([]agentctl.AgentEvent, 0, len(buffered)) + for _, event := range buffered { + if event.SessionID == "" || event.SessionID == newSessionID { + accepted = append(accepted, event) + } + } + return accepted +} + +func (e *AgentExecution) failContextReset(fence bool, reason string) { + e.contextResetMu.Lock() + defer e.contextResetMu.Unlock() + e.contextResetInFlight = false + e.contextResetRecoveryInFlight = false + e.contextResetEvents = nil + if fence { + e.contextResetFenced = true + e.contextResetFenceReason = reason + } +} + +func (e *AgentExecution) beginContextResetRecovery() { + e.contextResetMu.Lock() + e.contextResetRecoveryInFlight = true + e.contextResetMu.Unlock() +} + +func (e *AgentExecution) finishContextResetRecovery(success bool, reason string) { + e.contextResetMu.Lock() + defer e.contextResetMu.Unlock() + e.contextResetRecoveryInFlight = false + if success { + e.contextResetFenced = false + e.contextResetFenceReason = "" + return + } + if e.contextResetFenced { + e.contextResetFenceReason = reason + } +} + +func (e *AgentExecution) contextResetAdmissionError() error { + e.contextResetMu.Lock() + defer e.contextResetMu.Unlock() + if e.contextResetFenced { + return ErrContextResetFenced + } + if e.contextResetInFlight { + return errContextResetInProgress + } + return nil +} + +func (e *AgentExecution) contextResetFencedState() bool { + e.contextResetMu.Lock() + defer e.contextResetMu.Unlock() + return e.contextResetFenced +} + +// bufferOrDropContextResetEvent returns true when the event was consumed by +// the reset boundary and must not enter ordinary lifecycle handling. +func (e *AgentExecution) bufferOrDropContextResetEvent(event agentctl.AgentEvent) bool { + e.contextResetMu.Lock() + defer e.contextResetMu.Unlock() + if e.contextResetFenced { + if e.contextResetRecoveryInFlight && isContextResetSetupEvent(event) { + return false + } + return true + } + if !e.contextResetInFlight { + return false + } + if isContextResetSetupEvent(event) && len(e.contextResetEvents) < maxContextResetEvents { + e.contextResetEvents = append(e.contextResetEvents, event) + } + return true +} + +func isContextResetSetupEvent(event agentctl.AgentEvent) bool { + switch event.Type { + case streams.EventTypeSessionStatus, + streams.EventTypeSessionMode, + streams.EventTypeSessionModels, + streams.EventTypeAgentCapabilities, + streams.EventTypeAvailableCommands, + streams.EventTypeContextWindow: + return true + default: + return false + } +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/execution_context_test.go b/apps/backend/internal/agent/runtime/lifecycle/execution_context_test.go index e5b7026935..9928d86d20 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/execution_context_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/execution_context_test.go @@ -6,6 +6,10 @@ import ( "testing/synctest" "time" + "github.com/stretchr/testify/require" + + agentctl "github.com/kandev/kandev/internal/agent/runtime/agentctl" + "github.com/kandev/kandev/internal/agentctl/types/streams" "github.com/kandev/kandev/internal/common/constants" ) @@ -56,3 +60,26 @@ func TestPreparationContextRetainsItsIndependentBudget(t *testing.T) { } }) } + +func TestFinishContextResetReturnsEventsBufferedDuringFinalization(t *testing.T) { + execution := &AgentExecution{} + require.NoError(t, execution.beginContextReset()) + + buffered := agentctl.AgentEvent{ + Type: streams.EventTypeSessionModels, + SessionID: "new-session", + } + stale := agentctl.AgentEvent{ + Type: streams.EventTypeSessionStatus, + SessionID: "old-session", + } + require.True(t, execution.bufferOrDropContextResetEvent(buffered)) + require.True(t, execution.bufferOrDropContextResetEvent(stale)) + + final := execution.finishContextReset("new-session") + require.Equal(t, []agentctl.AgentEvent{buffered}, final) + require.False(t, execution.bufferOrDropContextResetEvent(agentctl.AgentEvent{ + Type: streams.EventTypeSessionStatus, + SessionID: "new-session", + })) +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_events.go b/apps/backend/internal/agent/runtime/lifecycle/manager_events.go index 7096205329..47b928e895 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_events.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_events.go @@ -857,6 +857,27 @@ func (m *Manager) handleAgentEventWithAttempt( execution *AgentExecution, event agentctl.AgentEvent, attemptID string, +) { + m.handleAgentEventAtContextResetBoundary(execution, event, true, attemptID) +} + +// handleAgentEventAfterContextReset replays setup events after the lifecycle +// manager has committed the replacement session while the exclusive reset +// lease is still held. The event has already crossed the reset boundary and +// must not be buffered again. +func (m *Manager) handleAgentEventAfterContextReset(execution *AgentExecution, event agentctl.AgentEvent) { + startupGeneration := execution.startupAttemptSnapshot() + execution.withStartupAttempt(startupGeneration, func(attemptID string) { + m.handleAgentEventAtContextResetBoundary(execution, event, false, attemptID) + }) +} + +//nolint:cyclop,funlen // ACP event types share lifecycle bookkeeping before publication. +func (m *Manager) handleAgentEventAtContextResetBoundary( + execution *AgentExecution, + event agentctl.AgentEvent, + enforceResetBoundary bool, + attemptID string, ) { event.AttemptID = attemptID // A terminal event that was already applied from a retained turn outcome @@ -868,6 +889,13 @@ func (m *Manager) handleAgentEventWithAttempt( zap.Int64("control_turn_id", event.ControlTurnID)) return } + if enforceResetBoundary && execution.bufferOrDropContextResetEvent(event) { + m.logger.Debug("ignoring agent event at context reset boundary", + zap.String("execution_id", execution.ID), + zap.String("event_type", event.Type), + zap.String("session_id", event.SessionID)) + return + } if m.handleMCPAttachmentEvent(execution, &event, attemptID) { return } diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go b/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go index 4aee691234..e0ccde0744 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go @@ -25,8 +25,9 @@ import ( ) const ( - freshSessionModelStateWait = 2 * time.Second - freshSessionModelStatePoll = 10 * time.Millisecond + freshSessionModelStateWait = 2 * time.Second + freshSessionModelStatePoll = 10 * time.Millisecond + resetAgentContextRequestTimeout = 10 * time.Second ) // WasSessionInitialized reports whether the execution completed ACP session setup. @@ -112,6 +113,13 @@ func (m *Manager) PromptAgentWithDispatchCallback(ctx context.Context, execution key := executionActivityKey(executionID) m.trackActivity(key, lease) m.setRuntimeInterest(execution.SessionID, true) + operationRelease, err := execution.acquireContextResetOperation(ctx) + if err != nil { + m.releaseActivity(key) + m.setRuntimeInterest(execution.SessionID, false) + return nil, err + } + defer operationRelease() result, err := m.sessionManager.SendPromptWithDispatchCallback(ctx, execution, prompt, true, attachments, dispatchOnly, onDispatched) if err != nil || !dispatchOnly { m.releaseActivity(key) @@ -138,6 +146,13 @@ func (m *Manager) SteerAgentWithDispatchCallback(ctx context.Context, executionI key := executionActivityKey(executionID) m.trackActivity(key, lease) m.setRuntimeInterest(execution.SessionID, true) + operationRelease, err := execution.acquireContextResetOperation(ctx) + if err != nil { + m.releaseActivity(key) + m.setRuntimeInterest(execution.SessionID, false) + return nil, err + } + defer operationRelease() result, err := m.sessionManager.SendPromptSteerWithDispatchCallback(ctx, execution, prompt, true, attachments, dispatchOnly, onDispatched) if err != nil || !dispatchOnly { m.releaseActivity(key) @@ -361,6 +376,11 @@ func (m *Manager) SetSessionMode(ctx context.Context, executionID, _ string, mod if !exists { return fmt.Errorf("execution %q not found", executionID) } + operationRelease, err := execution.acquireContextResetOperation(ctx) + if err != nil { + return err + } + defer operationRelease() client, releaseClient := execution.AcquireAgentCtlClient() defer releaseClient() if client == nil { @@ -391,15 +411,22 @@ func (m *Manager) SetSessionModel(ctx context.Context, executionID, modelID stri if !exists { return fmt.Errorf("execution %q not found", executionID) } + operationRelease, err := execution.acquireContextResetOperation(ctx) + if err != nil { + return err + } if execution.PassthroughProcessID != "" { if err := m.executionStore.WithLock(executionID, func(exec *AgentExecution) { exec.setMetadataValue(MetadataKeyModelOverride, modelID) }); err != nil { + operationRelease() return fmt.Errorf("failed to persist model override for execution %q: %w", executionID, err) } + operationRelease() return m.RestartAgentProcess(ctx, executionID) } + defer operationRelease() client, releaseClient := execution.AcquireAgentCtlClient() defer releaseClient() @@ -424,6 +451,11 @@ func (m *Manager) SetSessionConfigOption(ctx context.Context, executionID, confi if !exists { return fmt.Errorf("execution %q not found", executionID) } + operationRelease, err := execution.acquireContextResetOperation(ctx) + if err != nil { + return err + } + defer operationRelease() client, releaseClient := execution.AcquireAgentCtlClient() defer releaseClient() if client == nil { @@ -797,14 +829,27 @@ func (m *Manager) ResetAgentContext(ctx context.Context, executionID string) err if execution.PassthroughProcessID != "" { return m.RestartAgentProcess(ctx, executionID) } + if err := execution.contextResetAdmissionError(); err != nil { + return err + } execution.remoteInstanceLifecycleMu.Lock() defer execution.remoteInstanceLifecycleMu.Unlock() if current, currentExists := m.executionStore.Get(executionID); !currentExists || current != execution { return fmt.Errorf("execution %q not found: %w", executionID, ErrExecutionNotFound) } + if err := execution.beginContextReset(); err != nil { + return err + } + operationRelease, err := execution.acquireContextResetExclusive(ctx) + if err != nil { + execution.failContextReset(false, err.Error()) + return err + } + defer operationRelease() client, releaseClient := execution.AcquireAgentCtlClient() if client == nil { + execution.failContextReset(false, "agentctl client is unavailable") return fmt.Errorf("execution %q has no agentctl client", executionID) } @@ -824,30 +869,68 @@ func (m *Manager) ResetAgentContext(ctx context.Context, executionID string) err agentConfig, err := m.getAgentConfigForExecution(execution) if err != nil { releaseClient() + execution.failContextReset(false, err.Error()) m.logger.Info("cannot resolve agent config for session reset, falling back to process restart", zap.String("execution_id", executionID), zap.Error(err)) - return m.restartAgentProcess(ctx, executionID, &runtimeConfig) + if restartErr := m.restartAgentProcess(ctx, executionID, &runtimeConfig); restartErr != nil { + execution.failContextReset(true, restartErr.Error()) + return restartErr + } + return nil } mcpServers, err := m.resolveMcpServers(ctx, execution, agentConfig) if err != nil { releaseClient() + execution.failContextReset(false, err.Error()) m.logger.Warn("cannot resolve MCP servers for session reset, falling back to process restart", zap.String("execution_id", executionID), zap.Error(err)) - return m.restartAgentProcess(ctx, executionID, &runtimeConfig) + if restartErr := m.restartAgentProcess(ctx, executionID, &runtimeConfig); restartErr != nil { + execution.failContextReset(true, restartErr.Error()) + return restartErr + } + return nil } - // Try session-level reset (only ACP adapters support this) - newSessionID, err := client.ResetSession(ctx, execution.WorkspacePath, mcpServers) + // Try session-level reset (only ACP adapters support this). The provider + // request has its own bound so an unanswered response cannot hold the + // execution lifecycle lock indefinitely. + resetCtx, cancelReset := context.WithTimeout(ctx, resetAgentContextRequestTimeout) + newSessionID, err := client.ResetSession(resetCtx, execution.WorkspacePath, mcpServers) + resetErr := resetCtx.Err() + cancelReset() releaseClient() + // A successful session/new response is authoritative even if the request + // context becomes done while the adapter is returning. In particular, the + // adapter may still be doing detached superseded-session cleanup after it + // has committed the replacement. Only an unsuccessful RPC can turn the + // request deadline into an uncertain reset that needs fencing. if err != nil { + if resetErr != nil { + resetFailure := fmt.Errorf("agent session reset request ended with %w", resetErr) + if err != nil { + resetFailure = fmt.Errorf("%w: %v", resetFailure, err) + } + execution.failContextReset(true, resetFailure.Error()) + return resetFailure + } + execution.failContextReset(false, err.Error()) m.logger.Info("session reset not supported, falling back to process restart", zap.String("execution_id", executionID), zap.Error(err)) - return m.restartAgentProcess(ctx, executionID, &runtimeConfig) + if restartErr := m.restartAgentProcess(ctx, executionID, &runtimeConfig); restartErr != nil { + execution.failContextReset(true, restartErr.Error()) + return restartErr + } + return nil + } + if newSessionID == "" { + err := errors.New("agent session reset returned an empty session ID") + execution.failContextReset(true, err.Error()) + return err } // Success — update execution state without restarting process - _ = m.executionStore.WithLock(executionID, func(exec *AgentExecution) { + if err := m.executionStore.WithLock(executionID, func(exec *AgentExecution) { exec.ACPSessionID = newSessionID exec.needsResumeContext = false exec.resumeContextInjected = false @@ -860,25 +943,41 @@ func (m *Manager) ResetAgentContext(ctx context.Context, executionID string) err default: } exec.dispatchedPromptPending.Store(false) - }) + }); err != nil { + execution.failContextReset(true, err.Error()) + return fmt.Errorf("commit reset session state: %w", err) + } + for _, event := range execution.drainContextResetEvents(newSessionID) { + m.handleAgentEventAfterContextReset(execution, event) + } + reconcileCtx, cancelReconcile := context.WithTimeout( + context.WithoutCancel(ctx), resetAgentContextRequestTimeout, + ) + defer cancelReconcile() // Restore the complete captured configuration. A strict-mode model that is // rejected by the fresh session fails the reset explicitly instead of // silently dropping to the provider default. if !cacheFreshSessionModelState(execution) && (runtimeConfig.Model != "" || len(runtimeConfig.ConfigOptions) > 0) { - waitForFreshSessionModelState(ctx, m.logger, execution) + waitForFreshSessionModelState(reconcileCtx, m.logger, execution) } - if err := m.restoreSessionRuntimeConfig(ctx, execution, newSessionID, runtimeConfig); err != nil { + if err := m.restoreSessionRuntimeConfig(reconcileCtx, execution, newSessionID, runtimeConfig); err != nil { restoreErr := fmt.Errorf("failed to restore session runtime configuration after reset: %w", err) + execution.failContextReset(true, restoreErr.Error()) m.updateExecutionError(executionID, restoreErr.Error()) - m.persistExecutorRunning(context.WithoutCancel(ctx), execution) + m.persistExecutorRunning(context.WithoutCancel(reconcileCtx), execution) return restoreErr } - if err := m.updateStatusAndPersist(ctx, executionID, v1.AgentStatusReady); err != nil { + for _, event := range execution.drainContextResetEvents(newSessionID) { + m.handleAgentEventAfterContextReset(execution, event) + } + if err := m.updateStatusAndPersist(reconcileCtx, executionID, v1.AgentStatusReady); err != nil { + resetErr := fmt.Errorf("failed to mark reset agent ready: %w", err) + execution.failContextReset(true, resetErr.Error()) m.updateExecutionError(executionID, "failed to mark reset agent ready: "+err.Error()) - m.persistExecutorRunning(context.WithoutCancel(ctx), execution) - return fmt.Errorf("failed to mark reset agent ready: %w", err) + m.persistExecutorRunning(context.WithoutCancel(reconcileCtx), execution) + return resetErr } m.logger.Info("agent context reset via session (no process restart)", @@ -886,13 +985,16 @@ func (m *Manager) ResetAgentContext(ctx context.Context, executionID string) err zap.String("session_id", execution.SessionID), zap.String("new_acp_session_id", newSessionID)) - m.eventPublisher.PublishAgentEvent(ctx, events.AgentContextReset, execution) + m.eventPublisher.PublishAgentEvent(reconcileCtx, events.AgentContextReset, execution) // Boot-equivalent signal: a fresh ACP session is alive but no turn has run yet. // Use AgentBootReady so the orchestrator routes this to handleAgentBootReady // (idle/WAITING transition) rather than handleAgentReady (turn-end transition, // which would fire on_turn_complete against the current step — the original // boot-vs-turn ambiguity bug). - m.eventPublisher.PublishAgentEvent(ctx, events.AgentBootReady, execution) + m.eventPublisher.PublishAgentEvent(reconcileCtx, events.AgentBootReady, execution) + for _, event := range execution.finishContextReset(newSessionID) { + m.handleAgentEventAfterContextReset(execution, event) + } return nil } @@ -1142,7 +1244,18 @@ func (m *Manager) RestartAgentProcess(ctx context.Context, executionID string) e if current, currentExists := m.executionStore.Get(executionID); !currentExists || current != execution { return fmt.Errorf("execution %q not found: %w", executionID, ErrExecutionNotFound) } - return m.restartAgentProcess(ctx, executionID, nil) + operationRelease, err := execution.acquireContextResetExclusive(ctx) + if err != nil { + return err + } + defer operationRelease() + err = m.restartAgentProcess(ctx, executionID, nil) + if err != nil { + execution.finishContextResetRecovery(false, err.Error()) + return err + } + execution.finishContextResetRecovery(true, "") + return nil } // restartAgentProcess restarts an ACP process. When runtimeConfigOverride is @@ -1319,6 +1432,13 @@ func cloneSessionRuntimeConfig(config models.SessionRuntimeConfig) models.Sessio } func (m *Manager) resetAgentRestartState(executionID string, commands agentCommands) { + if execution, exists := m.executionStore.Get(executionID); exists && execution.contextResetFencedState() { + // The old reset outcome is fenced until this explicit restart owns the + // replacement. Streams are closed before this state reset, so setup + // events from the new process may be admitted while the fence remains in + // place; finishContextResetRecovery clears it only after restart success. + execution.beginContextResetRecovery() + } _ = m.executionStore.WithLock(executionID, func(exec *AgentExecution) { exec.ACPSessionID = "" exec.Status = v1.AgentStatusStarting diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_interaction_reset_timeout_test.go b/apps/backend/internal/agent/runtime/lifecycle/manager_interaction_reset_timeout_test.go new file mode 100644 index 0000000000..7976fd9c77 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_interaction_reset_timeout_test.go @@ -0,0 +1,224 @@ +package lifecycle + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + agentctl "github.com/kandev/kandev/internal/agent/runtime/agentctl" + "github.com/kandev/kandev/internal/agentctl/types/streams" + "github.com/kandev/kandev/internal/events" + v1 "github.com/kandev/kandev/pkg/api/v1" + "github.com/stretchr/testify/require" +) + +func TestManager_ResetAgentContext_ResetRequestTimeout(t *testing.T) { + mgr := newTestManager(t) + mock := newRestartMockAgentctlServer(t, false, false) + mock.suppressSessionResetResponse = true + resetReceived := make(chan struct{}) + mock.onReset = func() { close(resetReceived) } + + client := createTestClient(t, mock.server.URL) + t.Cleanup(client.Close) + streamCtx, streamCancel := context.WithCancel(context.Background()) + t.Cleanup(streamCancel) + require.NoError(t, client.StreamUpdates(streamCtx, func(agentctl.AgentEvent) {}, nil, nil)) + + exec := &AgentExecution{ + ID: "exec-reset-timeout", + TaskID: "task-reset-timeout", + SessionID: "session-reset-timeout", + AgentProfileID: "profile-reset-timeout", + ACPSessionID: "old-session", + AgentCommand: "auggie --model test", + Status: v1.AgentStatusRunning, + WorkspacePath: "/workspace", + sessionInitialized: true, + agentctl: client, + } + require.NoError(t, mgr.executionStore.Add(exec)) + + requestCtx, requestCancel := context.WithCancel(context.Background()) + t.Cleanup(requestCancel) + result := make(chan error, 1) + go func() { result <- mgr.ResetAgentContext(requestCtx, exec.ID) }() + + select { + case <-resetReceived: + case <-time.After(time.Second): + t.Fatal("timed out waiting for the provider to receive the reset request") + } + + // A terminal stop is the lifecycle operation used by session deletion. Start + // it while the reset request is still outstanding so the assertion proves + // the reset lock is released when the provider deadline expires. + stopResult := make(chan error, 1) + go func() { + stopResult <- mgr.StopAgentWithReason(context.Background(), exec.ID, StopReasonTaskDeleted, true) + }() + select { + case err := <-stopResult: + t.Fatalf("terminal stop completed before reset cleanup returned: %v", err) + case <-time.After(100 * time.Millisecond): + } + + select { + case err := <-result: + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("ResetAgentContext error = %v, want caller deadline", err) + } + case <-time.After(11 * time.Second): + requestCancel() + err := <-result + t.Fatalf("ResetAgentContext did not return from its bounded request wait; outer cancellation returned %v", err) + } + + require.Equal(t, []string{"agent.session.reset"}, mock.getWSActions()) + require.Empty(t, mock.getHTTPActions(), "a cancelled reset request must not enter restart fallback") + + select { + case err := <-stopResult: + require.NoError(t, err, "terminal stop must acquire the released reset lock") + case <-time.After(time.Second): + t.Fatal("terminal stop remained blocked after the reset request deadline") + } + if _, exists := mgr.executionStore.Get(exec.ID); exists { + t.Fatal("terminal stop did not remove the execution after reset cleanup") + } + assertResetDidNotPublishSuccessEvents(t, mgr) +} + +func TestManager_ResetAgentContext_CallerDeadlinePrecedesRequestTimeout(t *testing.T) { + mgr := newTestManager(t) + mock := newRestartMockAgentctlServer(t, false, false) + mock.suppressSessionResetResponse = true + resetReceived := make(chan struct{}) + mock.onReset = func() { close(resetReceived) } + + client := createTestClient(t, mock.server.URL) + t.Cleanup(client.Close) + streamCtx, streamCancel := context.WithCancel(context.Background()) + t.Cleanup(streamCancel) + require.NoError(t, client.StreamUpdates(streamCtx, func(agentctl.AgentEvent) {}, nil, nil)) + + exec := &AgentExecution{ + ID: "exec-reset-caller-deadline", + TaskID: "task-reset-caller-deadline", + SessionID: "session-reset-caller-deadline", + AgentProfileID: "profile-reset-caller-deadline", + ACPSessionID: "old-session", + AgentCommand: "auggie --model test", + Status: v1.AgentStatusRunning, + WorkspacePath: "/workspace", + sessionInitialized: true, + agentctl: client, + } + require.NoError(t, mgr.executionStore.Add(exec)) + + requestCtx, requestCancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + t.Cleanup(requestCancel) + result := make(chan error, 1) + started := time.Now() + go func() { result <- mgr.ResetAgentContext(requestCtx, exec.ID) }() + + select { + case <-resetReceived: + case <-time.After(time.Second): + t.Fatal("timed out waiting for the provider to receive the reset request") + } + + select { + case err := <-result: + require.ErrorIs(t, err, context.DeadlineExceeded) + require.Less(t, time.Since(started), time.Second, "caller deadline should beat the internal request timeout") + case <-time.After(time.Second): + t.Fatal("ResetAgentContext did not honor the caller deadline") + } + + require.Equal(t, []string{"agent.session.reset"}, mock.getWSActions()) + require.Empty(t, mock.getHTTPActions(), "a cancelled reset request must not enter restart fallback") + assertResetDidNotPublishSuccessEvents(t, mgr) +} + +func TestManager_ResetAgentContext_DelayedProviderSessionIsFenced(t *testing.T) { + mgr := newTestManager(t) + mock := newRestartMockAgentctlServer(t, false, false) + mock.resetResponseDelay = 250 * time.Millisecond + mock.resetLateEventSent = make(chan struct{}) + mock.resetLateEvent = &agentctl.AgentEvent{ + Type: streams.EventTypeSessionModels, + SessionID: "late-session", + CurrentModelID: "late-model", + SessionModels: []streams.SessionModelInfo{{ModelID: "late-model"}}, + } + + client := createTestClient(t, mock.server.URL) + t.Cleanup(client.Close) + streamCtx, streamCancel := context.WithCancel(context.Background()) + t.Cleanup(streamCancel) + lateEventReceived := make(chan struct{}) + receivedOnce := sync.Once{} + exec := &AgentExecution{ + ID: "exec-reset-delayed", + TaskID: "task-reset-delayed", + SessionID: "session-reset-delayed", + AgentProfileID: "profile-reset-delayed", + ACPSessionID: "old-session", + AgentCommand: "auggie --model test", + Status: v1.AgentStatusRunning, + WorkspacePath: "/workspace", + sessionInitialized: true, + agentctl: client, + } + exec.SetModelState(&CachedModelState{ + CurrentModelID: "old-model", + Models: []streams.SessionModelInfo{{ModelID: "old-model"}}, + }) + require.NoError(t, mgr.executionStore.Add(exec)) + require.NoError(t, client.StreamUpdates(streamCtx, func(event agentctl.AgentEvent) { + receivedOnce.Do(func() { close(lateEventReceived) }) + mgr.handleAgentEvent(exec, event) + }, nil, nil)) + + requestCtx, requestCancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer requestCancel() + result := make(chan error, 1) + go func() { result <- mgr.ResetAgentContext(requestCtx, exec.ID) }() + + select { + case err := <-result: + require.ErrorIs(t, err, context.DeadlineExceeded) + case <-time.After(time.Second): + t.Fatal("ResetAgentContext did not honor the caller deadline") + } + require.ErrorIs(t, exec.contextResetAdmissionError(), ErrContextResetFenced) + + select { + case <-mock.resetLateEventSent: + case <-time.After(time.Second): + t.Fatal("mock provider did not emit the delayed replacement-session event") + } + select { + case <-lateEventReceived: + case <-time.After(time.Second): + t.Fatal("lifecycle stream did not receive the delayed replacement-session event") + } + + state := exec.GetModelState() + require.Nil(t, state, "late setup events must not repopulate the discarded reset catalog") + require.Equal(t, "old-session", exec.ACPSessionID) + _, err := mgr.PromptAgent(context.Background(), exec.ID, "late prompt", nil, true) + require.ErrorIs(t, err, ErrContextResetFenced) +} + +func assertResetDidNotPublishSuccessEvents(t *testing.T, mgr *Manager) { + t.Helper() + managerEvents := mgr.eventBus.(*MockEventBus).PublishedEvents + for _, event := range managerEvents { + require.NotEqual(t, events.AgentContextReset, event.Type) + require.NotEqual(t, events.AgentBootReady, event.Type) + } +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_interaction_test.go b/apps/backend/internal/agent/runtime/lifecycle/manager_interaction_test.go index 23b8912b35..dd81a07fa0 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_interaction_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_interaction_test.go @@ -47,19 +47,23 @@ type restartMockAgentctlServer struct { setModeIDs []string setOptions []restartConfigOption - failStop bool - failSessionNew bool - failSessionReset bool - failCacheRepair bool - failMode bool - failModel bool - failConfigOptionID string - stderrLines []string - modelState *streams.SessionModelState - newModelState *streams.SessionModelState - onReset func() - onSessionNew func() - onCacheRepair func() + failStop bool + failSessionNew bool + failSessionReset bool + failCacheRepair bool + failMode bool + failModel bool + failConfigOptionID string + stderrLines []string + modelState *streams.SessionModelState + newModelState *streams.SessionModelState + suppressSessionResetResponse bool + resetResponseDelay time.Duration + resetLateEvent *agentctl.AgentEvent + resetLateEventSent chan struct{} + onReset func() + onSessionNew func() + onCacheRepair func() } type restartConfigOption struct { @@ -260,6 +264,21 @@ func newRestartMockAgentctlServer(t *testing.T, failStop, failSessionNew bool) * if m.onReset != nil { m.onReset() } + if m.suppressSessionResetResponse { + continue + } + if m.resetResponseDelay > 0 { + time.Sleep(m.resetResponseDelay) + } + if m.resetLateEvent != nil { + eventData, _ := json.Marshal(m.resetLateEvent) + if err := conn.WriteMessage(websocket.TextMessage, eventData); err != nil { + return + } + if m.resetLateEventSent != nil { + close(m.resetLateEventSent) + } + } if m.failSessionReset { resp, _ = ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ "success": false, diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_startup.go b/apps/backend/internal/agent/runtime/lifecycle/manager_startup.go index d64f80f0e1..124f169a0c 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_startup.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_startup.go @@ -104,6 +104,9 @@ func (m *Manager) startAgentProcess(ctx context.Context, executionID string) (re if !exists { return fmt.Errorf("execution %q not found", executionID) } + if err := execution.contextResetAdmissionError(); err != nil { + return err + } defer func() { retErr = wrapBootstrapFailure(execution, retErr) }() @@ -115,6 +118,12 @@ func (m *Manager) startAgentProcess(ctx context.Context, executionID string) (re return err } operationCtx := activityClaim.Context(ctx) + operationRelease, err := execution.acquireContextResetOperation(operationCtx) + if err != nil { + activityClaim.Release() + return err + } + defer operationRelease() defer func() { if retErr != nil { activityClaim.Release() diff --git a/apps/backend/internal/agent/runtime/lifecycle/types.go b/apps/backend/internal/agent/runtime/lifecycle/types.go index 8d32d76b7a..8fb5d2cefb 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/types.go +++ b/apps/backend/internal/agent/runtime/lifecycle/types.go @@ -139,6 +139,21 @@ type AgentExecution struct { agentctlOverride atomic.Pointer[agentctl.Client] agentctlLifecycleMu sync.RWMutex remoteInstanceLifecycleMu sync.Mutex + // contextResetMu owns the reset attempt boundary. While a session reset is + // in flight, fresh-session setup events are retained until the lifecycle + // manager has committed the new ACP session. A timed-out attempt discards + // those events and fences all later provider use until an explicit restart + // succeeds. + // contextResetAdmissionMu is the shared operation lease: ordinary provider + // operations hold a read lease, while a reset holds the write lease through + // session replacement, configuration restoration, and ready publication. + contextResetAdmissionMu sync.RWMutex + contextResetMu sync.Mutex + contextResetInFlight bool + contextResetFenced bool + contextResetFenceReason string + contextResetRecoveryInFlight bool + contextResetEvents []agentctl.AgentEvent // agentctlReady records the successful health check independently of the // agent process and workspace stream lifecycles. Prepared sessions have a // healthy agentctl before either of those is started or attached. diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter.go index d2e1e6f98d..0431d599d5 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter.go @@ -281,6 +281,8 @@ type Adapter struct { // Session transitions use a separate mutex because a reset must keep the // adapter transitionally consistent from session/new through session/close. sessionTransitionMu sync.Mutex + sessionCleanupDone chan struct{} + sessionCleanupWg sync.WaitGroup configChangeMu sync.Mutex configGeneration uint64 contextSamples map[string]contextWindowSample @@ -734,6 +736,7 @@ func (a *Adapter) Close() error { a.mu.Lock() if a.closed { a.mu.Unlock() + a.waitForSessionCleanup() return nil } a.closed = true @@ -756,6 +759,12 @@ func (a *Adapter) Close() error { a.lifetimeCancel() } + // A successful reset returns before its best-effort session/close request, + // but adapter shutdown owns that worker and must drain it before returning. + // Synchronizing with the transition mutex first ensures a reset that is just + // committing its cleanup has registered the wait-group entry. + a.waitForSessionCleanup() + // Wait for the update worker to exit before closing updatesCh. // handleACPUpdate may call sendUpdate, so updatesCh must remain open // until the worker is gone. diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_session.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_session.go index c7bede7159..5cb003c03e 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_session.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_session.go @@ -18,13 +18,75 @@ import ( const kandevMCPServerName = "kandev" +const sessionTransitionPollInterval = 10 * time.Millisecond + // PublishesMCPAttachmentResults reports that this adapter emits attachment // results for the servers that survive its own capability filtering. func (a *Adapter) PublishesMCPAttachmentResults() bool { return true } +// lockSessionTransition waits for the previous reset's cleanup to finish before +// allowing another session transition. The reset response remains independent +// from session/close, while this gate keeps a later load/reset from overtaking +// the cleanup that still owns the superseded session. +func (a *Adapter) lockSessionTransition(ctx context.Context) error { + if ctx == nil { + ctx = context.Background() + } + for { + if a.sessionTransitionMu.TryLock() { + a.mu.RLock() + closed := a.closed + a.mu.RUnlock() + if closed { + a.sessionTransitionMu.Unlock() + return errors.New("adapter is closed") + } + if a.sessionCleanupDone == nil { + return nil + } + if err := ctx.Err(); err != nil { + a.sessionTransitionMu.Unlock() + return err + } + done := a.sessionCleanupDone + a.sessionTransitionMu.Unlock() + select { + case <-done: + case <-ctx.Done(): + return ctx.Err() + } + continue + } + if err := ctx.Err(); err != nil { + return err + } + timer := time.NewTimer(sessionTransitionPollInterval) + select { + case <-ctx.Done(): + if !timer.Stop() { + <-timer.C + } + return ctx.Err() + case <-timer.C: + } + } +} + +func (a *Adapter) waitForSessionCleanup() { + a.sessionTransitionMu.Lock() + cleanupInFlight := a.sessionCleanupDone != nil + a.sessionTransitionMu.Unlock() + if !cleanupInFlight { + return + } + a.sessionCleanupWg.Wait() +} + // NewSession creates a new agent session. func (a *Adapter) NewSession(ctx context.Context, mcpServers []types.McpServer) (string, error) { - a.sessionTransitionMu.Lock() + if err := a.lockSessionTransition(ctx); err != nil { + return "", err + } defer a.sessionTransitionMu.Unlock() return a.newSession(ctx, mcpServers) } @@ -384,7 +446,9 @@ func mapToHTTPHeaders(headers map[string]string) []acp.HttpHeader { // //nolint:funlen // pre-existing length preserved from adapter.go file split func (a *Adapter) LoadSession(ctx context.Context, sessionID string, mcpServers []types.McpServer) error { - a.sessionTransitionMu.Lock() + if err := a.lockSessionTransition(ctx); err != nil { + return err + } defer a.sessionTransitionMu.Unlock() a.mu.Lock() @@ -543,8 +607,9 @@ func (a *Adapter) LoadSession(ctx context.Context, sessionID string, mcpServers // superseded one, so a successful reset closes the outgoing session to release its // resources. The old session is captured before NewSession overwrites a.sessionID. func (a *Adapter) ResetSession(ctx context.Context, mcpServers []types.McpServer) (string, error) { - a.sessionTransitionMu.Lock() - defer a.sessionTransitionMu.Unlock() + if err := a.lockSessionTransition(ctx); err != nil { + return "", err + } a.mu.RLock() previous, conn := a.sessionID, a.acpConn @@ -552,12 +617,35 @@ func (a *Adapter) ResetSession(ctx context.Context, mcpServers []types.McpServer newID, err := a.newSession(ctx, mcpServers) if err != nil { + a.sessionTransitionMu.Unlock() return "", err } - a.closeSupersededSessionLocked(ctx, conn, previous, newID) + // session/new has committed the replacement session. Superseded-session + // cleanup is best effort and has its own bounded context, so it must not + // hold the reset request open while a provider ignores session/close. The + // cleanup remains queued behind the transition mutex so another session + // transition cannot make the still-current check stale. + if previous != "" && previous != newID && conn != nil { + done := make(chan struct{}) + a.sessionCleanupDone = done + a.sessionCleanupWg.Add(1) + go a.runSupersededSessionCleanup(done, conn, previous, newID) + } + a.sessionTransitionMu.Unlock() return newID, nil } +func (a *Adapter) runSupersededSessionCleanup( + done chan struct{}, conn *acp.ClientSideConnection, previous, newID string, +) { + defer a.sessionCleanupWg.Done() + a.sessionTransitionMu.Lock() + defer a.sessionTransitionMu.Unlock() + a.closeSupersededSessionLocked(context.Background(), conn, previous, newID) + close(done) + a.sessionCleanupDone = nil +} + // closeSupersededSessionTimeout bounds session/close after a reset so cleanup // can't hang the caller when the agent doesn't respond. const closeSupersededSessionTimeout = 10 * time.Second @@ -572,14 +660,16 @@ const closeSupersededSessionTimeout = 10 * time.Second // running. A close error is logged and never surfaces to the caller, since the // reset itself already succeeded. func (a *Adapter) closeSupersededSession(ctx context.Context, conn *acp.ClientSideConnection, previous, newID string) { - a.sessionTransitionMu.Lock() + if err := a.lockSessionTransition(ctx); err != nil { + return + } defer a.sessionTransitionMu.Unlock() a.closeSupersededSessionLocked(ctx, conn, previous, newID) } // closeSupersededSessionLocked requires sessionTransitionMu to be held so the // still-current check and session/close request form one transition. -func (a *Adapter) closeSupersededSessionLocked(ctx context.Context, conn *acp.ClientSideConnection, previous, newID string) { +func (a *Adapter) closeSupersededSessionLocked(_ context.Context, conn *acp.ClientSideConnection, previous, newID string) { if previous == "" || previous == newID || conn == nil { return } @@ -591,7 +681,11 @@ func (a *Adapter) closeSupersededSessionLocked(ctx context.Context, conn *acp.Cl return } - closeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), closeSupersededSessionTimeout) + cleanupBase := a.lifetimeCtx + if cleanupBase == nil { + cleanupBase = context.Background() + } + closeCtx, cancel := context.WithTimeout(context.WithoutCancel(cleanupBase), closeSupersededSessionTimeout) defer cancel() if _, err := conn.CloseSession(closeCtx, acp.CloseSessionRequest{ SessionId: acp.SessionId(previous), diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_session_test.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_session_test.go index 24e23bc25f..7e70058823 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_session_test.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_session_test.go @@ -19,11 +19,14 @@ type sessionRequestCaptureAgent struct { loadRequest acpsdk.LoadSessionRequest newStarted chan struct{} releaseNew chan struct{} + newCalls chan struct{} closeStarted chan struct{} releaseClose chan struct{} + closeCalls chan struct{} loadStarted chan struct{} mu sync.Mutex + closeStartedOnce sync.Once sessionCounter int failNewSessionOnAttempt int closeRequests []acpsdk.CloseSessionRequest @@ -63,7 +66,10 @@ func (*sessionRequestCaptureAgent) Cancel(context.Context, acpsdk.CancelNotifica func (a *sessionRequestCaptureAgent) CloseSession(_ context.Context, req acpsdk.CloseSessionRequest) (acpsdk.CloseSessionResponse, error) { if a.closeStarted != nil { - close(a.closeStarted) + a.closeStartedOnce.Do(func() { close(a.closeStarted) }) + } + if a.closeCalls != nil { + a.closeCalls <- struct{}{} } if a.releaseClose != nil { <-a.releaseClose @@ -84,6 +90,9 @@ func (*sessionRequestCaptureAgent) ListSessions(context.Context, acpsdk.ListSess func (a *sessionRequestCaptureAgent) NewSession(_ context.Context, request acpsdk.NewSessionRequest) (acpsdk.NewSessionResponse, error) { a.newRequest = request + if a.newCalls != nil { + a.newCalls <- struct{}{} + } if a.newStarted != nil { close(a.newStarted) <-a.releaseNew @@ -258,15 +267,21 @@ func TestResetSessionClosesSupersededSessionWhenAdvertised(t *testing.T) { t.Fatalf("NewSession: %v", err) } + capture.closeStarted = make(chan struct{}) secondID, err := adapter.ResetSession(context.Background(), nil) if err != nil { t.Fatalf("ResetSession: %v", err) } + select { + case <-capture.closeStarted: + case <-time.After(time.Second): + t.Fatal("reset cleanup did not reach session/close") + } if secondID == firstID { t.Fatalf("ResetSession returned the same session id %q twice", secondID) } - closes := capture.recordedCloseRequests() + closes := waitForCloseRequests(t, capture, 1) if len(closes) != 1 { t.Fatalf("CloseSession called %d times, want 1", len(closes)) } @@ -353,6 +368,7 @@ func TestResetSessionSucceedsWhenCloseSessionErrors(t *testing.T) { adapter, capture := newSessionRequestCaptureAdapter(t, acpsdk.McpCapabilities{}) adapter.capabilities.SessionCapabilities.Close = &acpsdk.SessionCloseCapabilities{} capture.closeErr = errors.New("close failed") + capture.closeStarted = make(chan struct{}) firstID, err := adapter.NewSession(context.Background(), nil) if err != nil { @@ -366,7 +382,12 @@ func TestResetSessionSucceedsWhenCloseSessionErrors(t *testing.T) { if newID == "" { t.Fatal("ResetSession returned an empty session id") } - closes := capture.recordedCloseRequests() + select { + case <-capture.closeStarted: + case <-time.After(time.Second): + t.Fatal("reset cleanup did not reach session/close") + } + closes := waitForCloseRequests(t, capture, 1) if len(closes) != 1 { t.Fatalf("CloseSession called %d times, want 1", len(closes)) } @@ -399,6 +420,15 @@ func TestResetSessionSerializesConcurrentLoadDuringClose(t *testing.T) { t.Fatal("reset did not reach session/close") } + select { + case err := <-resetDone: + if err != nil { + t.Fatalf("ResetSession: %v", err) + } + case <-time.After(time.Second): + t.Fatal("ResetSession remained blocked by superseded-session cleanup") + } + loadDone := make(chan error, 1) go func() { loadDone <- adapter.LoadSession(context.Background(), firstID, nil) @@ -412,14 +442,6 @@ func TestResetSessionSerializesConcurrentLoadDuringClose(t *testing.T) { close(capture.releaseClose) select { - case err := <-resetDone: - if err != nil { - t.Fatalf("ResetSession: %v", err) - } - case <-time.After(time.Second): - t.Fatal("reset did not complete") - } - select { case <-capture.loadStarted: case <-time.After(time.Second): t.Fatal("LoadSession did not reach the provider after reset cleanup") @@ -434,6 +456,139 @@ func TestResetSessionSerializesConcurrentLoadDuringClose(t *testing.T) { } } +func TestResetSessionSerializesConsecutiveResetsDuringCleanup(t *testing.T) { + type resetResult struct { + id string + err error + } + + adapter, capture := newSessionRequestCaptureAdapter(t, acpsdk.McpCapabilities{}) + adapter.capabilities.SessionCapabilities.Close = &acpsdk.SessionCloseCapabilities{} + + firstID, err := adapter.NewSession(context.Background(), nil) + if err != nil { + t.Fatalf("NewSession: %v", err) + } + capture.closeCalls = make(chan struct{}, 2) + capture.releaseClose = make(chan struct{}) + + firstResetResult := make(chan resetResult, 1) + go func() { + id, resetErr := adapter.ResetSession(context.Background(), nil) + firstResetResult <- resetResult{id: id, err: resetErr} + }() + select { + case <-capture.closeCalls: + case <-time.After(time.Second): + t.Fatal("first reset did not start superseded-session cleanup") + } + select { + case result := <-firstResetResult: + if result.err != nil { + t.Fatalf("first ResetSession: %v", result.err) + } + if result.id == firstID { + t.Fatalf("first reset returned the original session id %q", result.id) + } + case <-time.After(time.Second): + t.Fatal("first reset remained blocked by superseded-session cleanup") + } + + capture.newCalls = make(chan struct{}, 1) + secondResetResult := make(chan resetResult, 1) + go func() { + id, resetErr := adapter.ResetSession(context.Background(), nil) + secondResetResult <- resetResult{id: id, err: resetErr} + }() + select { + case <-capture.newCalls: + t.Fatal("second reset reached session/new while first cleanup was blocked") + case <-time.After(50 * time.Millisecond): + } + + close(capture.releaseClose) + var secondID string + select { + case result := <-secondResetResult: + if result.err != nil { + t.Fatalf("second ResetSession: %v", result.err) + } + secondID = result.id + case <-time.After(time.Second): + t.Fatal("second reset did not complete after first cleanup released") + } + if secondID == firstID { + t.Fatalf("second reset returned the original session id %q", secondID) + } + + closes := waitForCloseRequests(t, capture, 2) + if len(closes) != 2 { + t.Fatalf("CloseSession called %d times, want 2", len(closes)) + } + if string(closes[0].SessionId) != firstID || string(closes[1].SessionId) == string(closes[0].SessionId) { + t.Fatalf("closed session ids = %v, want the two superseded sessions in order", closes) + } +} + +func TestAdapterCloseWaitsForSupersededSessionCleanup(t *testing.T) { + adapter, capture := newSessionRequestCaptureAdapter(t, acpsdk.McpCapabilities{}) + adapter.capabilities.SessionCapabilities.Close = &acpsdk.SessionCloseCapabilities{} + if _, err := adapter.NewSession(context.Background(), nil); err != nil { + t.Fatalf("NewSession: %v", err) + } + capture.closeCalls = make(chan struct{}, 1) + capture.releaseClose = make(chan struct{}) + + resetDone := make(chan error, 1) + go func() { + _, err := adapter.ResetSession(context.Background(), nil) + resetDone <- err + }() + select { + case <-capture.closeCalls: + case <-time.After(time.Second): + t.Fatal("reset did not start superseded-session cleanup") + } + select { + case err := <-resetDone: + if err != nil { + t.Fatalf("ResetSession: %v", err) + } + case <-time.After(time.Second): + t.Fatal("ResetSession remained blocked by superseded-session cleanup") + } + + closeDone := make(chan error, 1) + go func() { closeDone <- adapter.Close() }() + select { + case err := <-closeDone: + t.Fatalf("Adapter.Close returned before session cleanup drained: %v", err) + case <-time.After(50 * time.Millisecond): + } + + close(capture.releaseClose) + select { + case err := <-closeDone: + if err != nil { + t.Fatalf("Adapter.Close: %v", err) + } + case <-time.After(time.Second): + t.Fatal("Adapter.Close did not return after session cleanup released") + } +} + +func waitForCloseRequests(t *testing.T, capture *sessionRequestCaptureAgent, want int) []acpsdk.CloseSessionRequest { + t.Helper() + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) { + if closes := capture.recordedCloseRequests(); len(closes) >= want { + return closes + } + time.Sleep(time.Millisecond) + } + return capture.recordedCloseRequests() +} + func newSessionRequestCaptureAdapter(t *testing.T, capabilities acpsdk.McpCapabilities) (*Adapter, *sessionRequestCaptureAgent) { t.Helper() clientToAgentReader, clientToAgentWriter := io.Pipe() diff --git a/apps/backend/internal/integration/simulated_agent_manager_test.go b/apps/backend/internal/integration/simulated_agent_manager_test.go index ee8ac210ed..3f48e667f0 100644 --- a/apps/backend/internal/integration/simulated_agent_manager_test.go +++ b/apps/backend/internal/integration/simulated_agent_manager_test.go @@ -39,6 +39,9 @@ type SimulatedAgentManagerClient struct { shouldFail bool failAfter int // Fail after N successful launches launchCount int32 + promptCalls atomic.Int32 + restartProcessCalls atomic.Int32 + resetContextCalls atomic.Int32 acpMessageFn func(taskID, executionID string) []protocol.Message // Custom ACP messages stopCh chan struct{} } @@ -302,6 +305,7 @@ func (s *SimulatedAgentManagerClient) StopAgentWithReason(ctx context.Context, a // PromptAgent sends a follow-up prompt to a running agent // Note: attachments parameter is accepted but not used in simulation func (s *SimulatedAgentManagerClient) PromptAgent(ctx context.Context, agentExecutionID string, prompt string, _ []v1.MessageAttachment, _ bool) (*executor.PromptResult, error) { + s.promptCalls.Add(1) s.mu.Lock() execution, exists := s.instances[agentExecutionID] s.mu.Unlock() @@ -487,6 +491,12 @@ func (s *SimulatedAgentManagerClient) GetLaunchCount() int { return int(atomic.LoadInt32(&s.launchCount)) } +// PromptCallCount returns the number of follow-up prompt dispatches. Launch +// count alone cannot prove that an existing execution was not prompted. +func (s *SimulatedAgentManagerClient) PromptCallCount() int { + return int(s.promptCalls.Load()) +} + // Close stops all simulated agents func (s *SimulatedAgentManagerClient) Close() { close(s.stopCh) @@ -585,15 +595,25 @@ func (s *SimulatedAgentManagerClient) ResolveAgentProfile(ctx context.Context, p } func (s *SimulatedAgentManagerClient) RestartAgentProcess(ctx context.Context, agentExecutionID string) error { + s.restartProcessCalls.Add(1) s.logger.Info("simulated: restarting agent process", zap.String("agent_execution_id", agentExecutionID)) return nil } func (s *SimulatedAgentManagerClient) ResetAgentContext(ctx context.Context, agentExecutionID string) error { + s.resetContextCalls.Add(1) return s.RestartAgentProcess(ctx, agentExecutionID) } +func (s *SimulatedAgentManagerClient) ResetContextCallCount() int { + return int(s.resetContextCalls.Load()) +} + +func (s *SimulatedAgentManagerClient) RestartProcessCallCount() int { + return int(s.restartProcessCalls.Load()) +} + func (s *SimulatedAgentManagerClient) SetSessionModelBySessionID(_ context.Context, _, _ string) error { return fmt.Errorf("not supported") } diff --git a/apps/backend/internal/integration/test_server_test.go b/apps/backend/internal/integration/test_server_test.go index fcf67406f5..8ae9782704 100644 --- a/apps/backend/internal/integration/test_server_test.go +++ b/apps/backend/internal/integration/test_server_test.go @@ -28,6 +28,7 @@ import ( "github.com/kandev/kandev/internal/workflow" workflowcontroller "github.com/kandev/kandev/internal/workflow/controller" workflowhandlers "github.com/kandev/kandev/internal/workflow/handlers" + wfmodels "github.com/kandev/kandev/internal/workflow/models" workflowservice "github.com/kandev/kandev/internal/workflow/service" "github.com/kandev/kandev/internal/worktree" v1 "github.com/kandev/kandev/pkg/api/v1" @@ -54,6 +55,61 @@ type taskRepositoryAdapter struct { svc *taskservice.Service } +type orchestratorWorkflowStepGetter struct { + svc *workflowservice.Service +} + +type integrationWorkflowProvider struct { + svc *taskservice.Service +} + +func (a *integrationWorkflowProvider) ListWorkflows(ctx context.Context, workspaceID string, includeHidden bool) ([]*models.Workflow, error) { + return a.svc.ListWorkflows(ctx, workspaceID, includeHidden) +} + +func (a *integrationWorkflowProvider) GetWorkflow(ctx context.Context, id string) (*models.Workflow, error) { + return a.svc.GetWorkflow(ctx, id) +} + +func (a *integrationWorkflowProvider) CreateWorkflow(ctx context.Context, workspaceID, name, description string) (*models.Workflow, error) { + return a.svc.CreateWorkflow(ctx, &taskservice.CreateWorkflowRequest{ + WorkspaceID: workspaceID, + Name: name, + Description: description, + }) +} + +func (a *integrationWorkflowProvider) UpdateWorkflow(ctx context.Context, workflow *models.Workflow) error { + prompt := workflow.Prompt + _, err := a.svc.UpdateWorkflow(ctx, workflow.ID, &taskservice.UpdateWorkflowRequest{ + Name: &workflow.Name, + Description: &workflow.Description, + Prompt: &prompt, + AgentProfileID: &workflow.AgentProfileID, + }) + return err +} + +func (a orchestratorWorkflowStepGetter) GetStep(ctx context.Context, stepID string) (*wfmodels.WorkflowStep, error) { + return a.svc.GetStep(ctx, stepID) +} + +func (a orchestratorWorkflowStepGetter) GetNextStepByPosition(ctx context.Context, workflowID string, position int) (*wfmodels.WorkflowStep, error) { + return a.svc.GetNextStepByPosition(ctx, workflowID, position) +} + +func (a orchestratorWorkflowStepGetter) GetPreviousStepByPosition(ctx context.Context, workflowID string, position int) (*wfmodels.WorkflowStep, error) { + return a.svc.GetPreviousStepByPosition(ctx, workflowID, position) +} + +func (a orchestratorWorkflowStepGetter) GetWorkflowMeta(ctx context.Context, workflowID string) (orchestrator.WorkflowMeta, error) { + meta, err := a.svc.GetWorkflowMeta(ctx, workflowID) + if err != nil { + return orchestrator.WorkflowMeta{}, err + } + return orchestrator.WorkflowMeta{AgentProfileID: meta.AgentProfileID, Prompt: meta.Prompt}, nil +} + func (a *taskRepositoryAdapter) GetTask(ctx context.Context, taskID string) (*v1.Task, error) { task, err := a.repo.GetTask(ctx, taskID) if err != nil { @@ -367,6 +423,7 @@ func NewOrchestratorTestServer(t *testing.T) *OrchestratorTestServer { taskSvc.SetWorkflowStepCreator(workflowSvc) taskSvc.SetWorkflowStepGetter(workflowSvc) taskSvc.SetWorkspaceBootstrapper(taskRepo) + workflowSvc.SetWorkflowProvider(&integrationWorkflowProvider{svc: taskSvc}) // Create simulated agent manager agentManager := NewSimulatedAgentManager(eventBus, log) @@ -383,6 +440,7 @@ func NewOrchestratorTestServer(t *testing.T) *OrchestratorTestServer { msgCreator := &testMessageCreatorAdapter{svc: taskSvc} orchestratorSvc.SetMessageCreator(msgCreator) orchestratorSvc.SetTurnService(&testTurnServiceAdapter{svc: taskSvc}) + orchestratorSvc.SetWorkflowStepGetter(orchestratorWorkflowStepGetter{svc: workflowSvc}) // Create WebSocket gateway gateway := gateways.NewGateway(log) diff --git a/apps/backend/internal/integration/workflow_context_reset_failure_test.go b/apps/backend/internal/integration/workflow_context_reset_failure_test.go new file mode 100644 index 0000000000..b1cccf62fd --- /dev/null +++ b/apps/backend/internal/integration/workflow_context_reset_failure_test.go @@ -0,0 +1,131 @@ +package integration + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/kandev/kandev/internal/task/models" + wfmodels "github.com/kandev/kandev/internal/workflow/models" + ws "github.com/kandev/kandev/pkg/websocket" +) + +func TestWorkflowResetEscalationLeavesSessionDeletable(t *testing.T) { + ts := NewOrchestratorTestServer(t) + defer ts.Close() + + ctx := context.Background() + taskID := ts.CreateTestTask(t, "augment-agent", 2) + task, err := ts.TaskRepo.GetTask(ctx, taskID) + require.NoError(t, err) + + steps, err := ts.WorkflowSvc.ListStepsByWorkflow(ctx, task.WorkflowID) + require.NoError(t, err) + require.GreaterOrEqual(t, len(steps), 2) + var target *wfmodels.WorkflowStep + for _, step := range steps { + if step.ID != task.WorkflowStepID { + target = step + break + } + } + require.NotNil(t, target) + target.Prompt = "distinct reset failure workflow prompt" + target.Events.OnEnter = []wfmodels.OnEnterAction{ + {Type: wfmodels.OnEnterResetAgentContext}, + {Type: wfmodels.OnEnterAutoStartAgent}, + } + require.NoError(t, ts.WorkflowSvc.UpdateStep(ctx, target)) + + // Keep the initial simulated turn active while the workflow move runs. + ts.AgentManager.SetExecutionTime(30 * time.Second) + client := NewOrchestratorWSClient(t, ts.Server.URL) + defer client.Close() + + startResp, err := client.SendRequest("start-1", ws.ActionSessionLaunch, map[string]interface{}{ + "task_id": taskID, + "agent_profile_id": "augment-agent", + }) + require.NoError(t, err) + var startPayload map[string]interface{} + require.NoError(t, startResp.ParsePayload(&startPayload)) + sessionID, ok := startPayload["session_id"].(string) + require.True(t, ok) + require.NotEmpty(t, sessionID) + + // StartTask creates the active turn before the launch response. Wait for + // the durable state so the move exercises reset's active-turn admission. + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + session, getErr := ts.TaskRepo.GetTaskSession(ctx, sessionID) + if getErr == nil && session.State == models.TaskSessionStateRunning { + break + } + time.Sleep(25 * time.Millisecond) + } + session, err := ts.TaskRepo.GetTaskSession(ctx, sessionID) + require.NoError(t, err) + require.Equal(t, models.TaskSessionStateRunning, session.State) + promptCallsBeforeMove := ts.AgentManager.PromptCallCount() + messagesBeforeMove, err := ts.TaskRepo.ListMessages(ctx, sessionID) + require.NoError(t, err) + userMessagesBeforeMove := countUserMessages(messagesBeforeMove) + + ts.AgentManager.MarkAgentHungForSession(sessionID) + launchCountBeforeMove := ts.AgentManager.GetLaunchCount() + moveResp, err := client.SendRequest("move-1", ws.ActionTaskMove, map[string]interface{}{ + "id": taskID, + "workflow_id": task.WorkflowID, + "workflow_step_id": target.ID, + "position": target.Position, + }) + require.NoError(t, err) + require.Equal(t, ws.MessageTypeResponse, moveResp.Type) + + var lastError models.LastAgentError + deadline = time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + session, getErr := ts.TaskRepo.GetTaskSession(ctx, sessionID) + if getErr == nil { + if candidate, found := models.LoadLastAgentError(session.Metadata); found { + lastError = candidate + break + } + } + time.Sleep(25 * time.Millisecond) + } + require.NotEmpty(t, lastError.Message) + require.Contains(t, strings.ToLower(lastError.Message), "context reset") + require.Contains(t, strings.ToLower(lastError.Message), "step prompt") + require.Contains(t, strings.ToLower(lastError.Details), "provider cancellation escalated") + require.Equal(t, 0, ts.AgentManager.ResetContextCallCount()) + require.Equal(t, 0, ts.AgentManager.RestartProcessCallCount()) + require.Equal(t, launchCountBeforeMove, ts.AgentManager.GetLaunchCount()) + require.Equal(t, promptCallsBeforeMove, ts.AgentManager.PromptCallCount()) + messagesAfterMove, err := ts.TaskRepo.ListMessages(ctx, sessionID) + require.NoError(t, err) + require.Equal(t, userMessagesBeforeMove, countUserMessages(messagesAfterMove)) + + deleteResp, err := client.SendRequest("delete-1", ws.ActionSessionDelete, map[string]interface{}{ + "session_id": sessionID, + }) + require.NoError(t, err) + require.Equal(t, ws.MessageTypeResponse, deleteResp.Type) + _, err = ts.TaskRepo.GetTaskSession(ctx, sessionID) + require.Error(t, err) + require.True(t, errors.Is(err, models.ErrTaskSessionNotFound), "session delete error = %v", err) +} + +func countUserMessages(messages []*models.Message) int { + count := 0 + for _, message := range messages { + if message != nil && message.AuthorType == models.MessageAuthorUser { + count++ + } + } + return count +} diff --git a/apps/backend/internal/orchestrator/event_handlers_agent.go b/apps/backend/internal/orchestrator/event_handlers_agent.go index f39c7dd947..303038eeaa 100644 --- a/apps/backend/internal/orchestrator/event_handlers_agent.go +++ b/apps/backend/internal/orchestrator/event_handlers_agent.go @@ -2794,7 +2794,7 @@ func (s *Service) handleRecoverableFailureLockedState(ctx context.Context, data } } -func (s *Service) persistLastAgentError(ctx context.Context, data watcher.AgentEventData) { +func (s *Service) persistLastAgentError(ctx context.Context, data watcher.AgentEventData) error { errMsg := data.ErrorMessage if errMsg == "" { errMsg = defaultAgentFailedMessage @@ -2821,7 +2821,7 @@ func (s *Service) persistLastAgentError(ctx context.Context, data watcher.AgentE zap.String("task_id", data.TaskID), zap.String("session_id", data.SessionID), zap.Error(err)) - return + return err } if s.eventBus != nil { eventData := map[string]interface{}{ @@ -2861,8 +2861,10 @@ func (s *Service) persistLastAgentError(ctx context.Context, data watcher.AgentE zap.String("task_id", data.TaskID), zap.String("session_id", data.SessionID), zap.Error(err)) + return err } } + return nil } // clearRecoveredAgentError drops a session's stored agent failure once the agent diff --git a/apps/backend/internal/orchestrator/event_handlers_clarification.go b/apps/backend/internal/orchestrator/event_handlers_clarification.go index ad049f42ed..bac68eca9f 100644 --- a/apps/backend/internal/orchestrator/event_handlers_clarification.go +++ b/apps/backend/internal/orchestrator/event_handlers_clarification.go @@ -884,7 +884,7 @@ func (s *Service) runSilentCancellationOwned(ctx context.Context, taskID, sessio return ErrSendNowTurnChanged } s.setCancellationIdentity(sessionID, operation, identity) - if err := s.cancelAgentWhileUnlocked(ctx, sessionID, guard.unlock, guard.relock); err != nil { + if err := s.cancelAgentWhileUnlocked(ctx, sessionID, operation, guard.unlock, guard.relock); err != nil { return err } session, err := s.repo.GetTaskSession(ctx, sessionID) @@ -1032,24 +1032,25 @@ func (s *Service) cancelAgentSilentWithGuardActionKindExclusiveConflict( kind cancellationKind, expectedTurnID string, conflictErr error, -) (bool, error) { +) (*cancelOperation, bool, error) { operation, registered, err := s.startExclusiveSilentCancellation( ctx, taskID, sessionID, action, kind, expectedTurnID, conflictErr, ) if err != nil { - return false, err + return nil, false, err } if unlockGuard != nil { unlockGuard() defer relockGuard() } if err := operation.wait(ctx); err != nil { - return false, err + return operation, false, err } if registered == nil { - return false, nil + return operation, false, nil } - return registered.wait(ctx) + dispatched, err := registered.wait(ctx) + return operation, dispatched, err } func (s *Service) logSilentCancelReconciled(taskID, sessionID string, err error) { diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index a1c0085420..c5b2a87416 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go @@ -16,6 +16,7 @@ import ( "go.uber.org/zap" "github.com/kandev/kandev/internal/agent/agents" + agentruntime "github.com/kandev/kandev/internal/agent/runtime" "github.com/kandev/kandev/internal/entityrefs" "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/events/bus" @@ -49,6 +50,16 @@ var ( errSessionAttachmentTransferUnavailable = errors.New("session attachment transfer service is unavailable") ) +const ( + workflowResetFailureCode = "workflow_context_reset_failed" + workflowResetFailureMessage = "Context reset failed. The workflow step prompt did not start." +) + +// workflowResetFailureCleanupTimeout bounds each failure-settlement stage. It +// remains a variable so tests can force a persistence timeout without waiting +// for the production budget. +var workflowResetFailureCleanupTimeout = 5 * time.Second + func sessionAttachmentTransfererAvailable(transfer SessionAttachmentTransferer) bool { if transfer == nil { return false @@ -4019,9 +4030,10 @@ func (s *Service) processOnEnter(ctx context.Context, taskID string, session *mo // skips the actual reset and no "Context reset" divider should show. // Capture this before the call, which may flip session.State. hadConversation := session.State != models.TaskSessionStateCreated - if !s.resetAgentContext(ctx, taskID, session, step.Name) { - s.setSessionWaitingForInput(ctx, taskID, sessionID, session) - s.publishSessionWaitingEvent(ctx, taskID, sessionID, step.ID, session) + _, resetErr := s.resetAgentContextWithError(ctx, taskID, session, step.Name, func(executionID string, err error) { + s.persistWorkflowResetFailure(ctx, taskID, sessionID, step.ID, step.Name, executionID, err) + }) + if resetErr != nil { return } s.markIdleAfterReset(ctx, taskID, sessionID, session, step, isPassthrough) @@ -6120,7 +6132,29 @@ func (s *Service) markIdleAfterReset( // resetAgentContext restarts the agent subprocess with a fresh ACP session, clearing // the agent's conversation context. The workspace environment is preserved. func (s *Service) resetAgentContext(ctx context.Context, taskID string, session *models.TaskSession, stepName string) bool { + _, err := s.resetAgentContextWithError(ctx, taskID, session, stepName) + return err == nil +} + +type workflowResetFailureHandler func(executionID string, err error) + +// resetAgentContextWithError performs a workflow context reset and retains the +// execution identity for callers that need to report a failed reset. The +// boolean wrapper above keeps manual and callback callers focused on the +// existing success contract. +// +//nolint:funlen // Reset settlement must keep provider, persistence, and guard stages together. +func (s *Service) resetAgentContextWithError( + ctx context.Context, taskID string, session *models.TaskSession, stepName string, + failureHandlers ...workflowResetFailureHandler, +) (string, error) { sessionID := session.ID + settleFailure := func(executionID string, err error) (string, error) { + if len(failureHandlers) > 0 && failureHandlers[0] != nil { + failureHandlers[0](executionID, err) + } + return executionID, err + } // A CREATED session has never been prompted, so there is no agent // conversation to clear. Its execution may still be workspace-only @@ -6135,7 +6169,7 @@ func (s *Service) resetAgentContext(ctx context.Context, taskID string, session s.logger.Debug("session has no agent context to reset, skipping", zap.String("session_id", sessionID), zap.String("step_name", stepName)) - return true + return "", nil } releaseLifecycleLock := s.acquireSessionLifecycleLock(sessionID) @@ -6155,7 +6189,7 @@ func (s *Service) resetAgentContext(ctx context.Context, taskID string, session zap.String("session_id", sessionID), zap.String("step_name", stepName), zap.Error(err)) - return false + return settleFailure(session.AgentExecutionID, fmt.Errorf("quiesce active turn: %w", err)) } executionID, err := s.agentManager.GetExecutionIDForSession(ctx, sessionID) @@ -6178,10 +6212,10 @@ func (s *Service) resetAgentContext(ctx context.Context, taskID string, session zap.String("session_id", sessionID), zap.String("step_name", stepName), zap.Error(err)) - return false + return settleFailure("", fmt.Errorf("clear lazy resume token: %w", err)) } s.clearPersistedResetState(ctx, sessionID, session) - return true + return "", nil } s.logger.Info("resetting agent context for workflow step", @@ -6191,14 +6225,23 @@ func (s *Service) resetAgentContext(ctx context.Context, taskID string, session zap.String("agent_execution_id", executionID)) previousACPSessionID := s.currentACPSessionID(sessionID) - if err := s.agentManager.ResetAgentContext(ctx, executionID); err != nil { + // The lifecycle manager synchronously republishes setup events that arrived + // during session replacement. Those publications enter the orchestrator's + // stream handler, which acquires this same guard. Keep reset admission fenced + // by the reset marker and lifecycle lock, but yield the stream guard while the + // provider operation and its event replay run; reacquire it before any result + // reconciliation or failure settlement. + resetGuard.unlock() + resetErr := s.agentManager.ResetAgentContext(ctx, executionID) + resetGuard.relock() + if resetErr != 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)) + zap.Error(resetErr)) s.reconcileFailedContextReset(ctx, taskID, session, executionID, previousACPSessionID) - return false + return settleFailure(executionID, fmt.Errorf("provider context reset: %w", resetErr)) } // Clear the old resume token only after the provider reset succeeds. This @@ -6211,7 +6254,7 @@ func (s *Service) resetAgentContext(ctx context.Context, taskID string, session zap.String("session_id", sessionID), zap.String("step_name", stepName), zap.Error(err)) - return false + return settleFailure(executionID, fmt.Errorf("clear resume token after context reset: %w", err)) } if acpSessionID := s.currentACPSessionID(sessionID); acpSessionID != "" { s.storeResumeToken(ctx, taskID, sessionID, executionID, acpSessionID, "") @@ -6220,7 +6263,51 @@ func (s *Service) resetAgentContext(ctx context.Context, taskID string, session // Clear the remaining persisted state (ACP session metadata, context window) // after the provider reset succeeds. The token is handled explicitly above. s.clearPersistedResetState(ctx, sessionID, session) - return true + return executionID, nil +} + +// persistWorkflowResetFailure settles the visible failure while the caller's +// session lifecycle lock and cancel guard still own reset admission. A queued +// successor or deletion therefore cannot observe the reset error midway +// through its metadata/state publication and overwrite the successor turn. +func (s *Service) persistWorkflowResetFailure( + ctx context.Context, + taskID, sessionID, stepID, stepName, executionID string, + resetErr error, +) { + failureCtx, cancelFailure := context.WithTimeout( + context.WithoutCancel(ctx), workflowResetFailureCleanupTimeout, + ) + defer cancelFailure() + if executionID == "" { + if session, err := s.repo.GetTaskSession(failureCtx, sessionID); err == nil && session != nil { + executionID = session.AgentExecutionID + } + } + failureData := watcher.AgentEventData{ + TaskID: taskID, + SessionID: sessionID, + AgentExecutionID: executionID, + ErrorMessage: workflowResetFailureMessage, + FailureCode: workflowResetFailureCode, + FailureDetails: fmt.Sprintf("workflow step %q: %s", stepName, resetErr), + } + if persistErr := s.persistLastAgentError(failureCtx, failureData); persistErr != nil { + s.logger.Error("failed to persist workflow context reset failure", + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.String("step_name", stepName), + zap.Error(persistErr)) + } + cancelFailure() + settlementCtx, cancelSettlement := context.WithTimeout( + context.WithoutCancel(ctx), workflowResetFailureCleanupTimeout, + ) + defer cancelSettlement() + // Do not pass the pre-reset session snapshot to either operation. The + // metadata write above must be visible in the waiting-state projection. + s.setSessionWaitingForInput(settlementCtx, taskID, sessionID) + s.publishSessionWaitingEvent(settlementCtx, taskID, sessionID, stepID) } // quiesceActiveResetTurn stops an in-flight turn through the internal silent @@ -6248,12 +6335,20 @@ func (s *Service) quiesceActiveResetTurn( // best-effort cancellation behavior without an expected durable ID. turnID = "" } - if _, err := s.cancelAgentSilentWithGuardActionKindExclusiveConflict( + operation, _, err := s.cancelAgentSilentWithGuardActionKindExclusiveConflict( ctx, taskID, sessionID, resetGuard.unlock, resetGuard.relock, nil, cancellationKindInternal, turnID, errContextResetCancellationConflict, - ); err != nil { + ) + if err != nil { return fmt.Errorf("cancel active turn for context reset at %s: %w", stepName, err) } + providerErr, outcomeReady := s.cancellationProviderOutcomeSnapshot(operation) + if !outcomeReady { + return fmt.Errorf("cancel active turn for context reset at %s: provider outcome unavailable", stepName) + } + if errors.Is(providerErr, agentruntime.ErrCancelEscalated) { + return fmt.Errorf("cancel active turn for context reset at %s: provider cancellation escalated: %w", stepName, providerErr) + } return nil } diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow_reset_failure_test.go b/apps/backend/internal/orchestrator/event_handlers_workflow_reset_failure_test.go new file mode 100644 index 0000000000..b27026fb31 --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_workflow_reset_failure_test.go @@ -0,0 +1,494 @@ +package orchestrator + +import ( + "context" + "errors" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "go.uber.org/zap/zaptest/observer" + + "github.com/kandev/kandev/internal/agent/runtime/lifecycle" + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/orchestrator/executor" + "github.com/kandev/kandev/internal/task/models" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" + wfmodels "github.com/kandev/kandev/internal/workflow/models" +) + +type workflowResetFailureBarrierRepo struct { + repoStore + entered chan struct{} + release chan struct{} + once sync.Once +} + +type blockingWorkflowResetAgentManager struct { + *mockAgentManager + resetStarted chan struct{} + resetRelease chan struct{} + resetOnce sync.Once +} + +func (m *blockingWorkflowResetAgentManager) ResetAgentContext(ctx context.Context, _ string) error { + m.resetOnce.Do(func() { close(m.resetStarted) }) + select { + case <-m.resetRelease: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func (r *workflowResetFailureBarrierRepo) SetSessionMetadataKey( + ctx context.Context, sessionID, key string, value interface{}, +) error { + if key == models.SessionMetaKeyLastAgentError { + r.once.Do(func() { close(r.entered) }) + select { + case <-r.release: + case <-ctx.Done(): + return ctx.Err() + } + } + return r.repoStore.SetSessionMetadataKey(ctx, sessionID, key, value) +} + +func TestProcessOnEnter_ResetFailurePersistsNotice(t *testing.T) { + tests := []struct { + name string + newAgentManager func() *mockAgentManager + prepare func(*Service, *models.TaskSession) + expectedExecutionID string + }{ + { + name: "provider reset failure", + newAgentManager: func() *mockAgentManager { + return &mockAgentManager{ + restartProcessErr: errors.New("provider reset failed"), + } + }, + expectedExecutionID: "exec-reset", + }, + { + name: "active cancellation escalation", + newAgentManager: func() *mockAgentManager { + return &mockAgentManager{cancelAgentErr: lifecycle.ErrCancelEscalated} + }, + prepare: func(svc *Service, session *models.TaskSession) { + session.AgentExecutionID = "exec-reset" + svc.activeTurns.Store(session.ID, "turn-reset") + }, + expectedExecutionID: "exec-reset", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + session, err := repo.GetTaskSession(context.Background(), "s1") + if err != nil { + t.Fatalf("load session: %v", err) + } + seedExecutorRunning(t, repo, session.ID, session.TaskID, "exec-reset") + // processOnEnter receives this object from the transition handler. Keep + // an intentionally stale copy here to prove the final event reloads the + // metadata written by persistLastAgentError. + session.Metadata = map[string]interface{}{ + models.SessionMetaKeyLastAgentError: models.LastAgentError{Message: "stale reset notice"}, + } + + eventBus := &mockEventBus{} + agentManager := tt.newAgentManager() + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentManager) + svc.eventBus = eventBus + messages := &mockMessageCreator{} + svc.messageCreator = messages + if tt.prepare != nil { + tt.prepare(svc, session) + } + + svc.processOnEnter(context.Background(), "t1", session, workflowResetFailureTestStep(), "review task", 0, nil) + assertResetFailureFeedback(t, repo, eventBus, agentManager, messages, tt.expectedExecutionID) + }) + } +} + +func TestProcessOnEnter_ResetFailureIgnoresCallerCancellation(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + session, err := repo.GetTaskSession(context.Background(), "s1") + if err != nil { + t.Fatalf("load session: %v", err) + } + seedExecutorRunning(t, repo, session.ID, session.TaskID, "exec-reset") + + eventBus := &mockEventBus{} + agentManager := &mockAgentManager{restartProcessErr: errors.New("provider reset failed")} + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentManager) + svc.eventBus = eventBus + messages := &mockMessageCreator{} + svc.messageCreator = messages + step := workflowResetFailureTestStep() + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + svc.processOnEnter(ctx, "t1", session, step, "review task", 0, nil) + + assertResetFailureFeedback(t, repo, eventBus, agentManager, messages, "exec-reset") +} + +func TestProcessOnEnter_ResetFailureLogsPersistenceFailure(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + session, err := repo.GetTaskSession(context.Background(), "s1") + if err != nil { + t.Fatalf("load session: %v", err) + } + seedExecutorRunning(t, repo, session.ID, session.TaskID, "exec-reset") + + core, logs := observer.New(zapcore.DebugLevel) + log, err := logger.NewFromZap(zap.New(core)) + if err != nil { + t.Fatalf("NewFromZap: %v", err) + } + eventBus := &mockEventBus{} + agentManager := &mockAgentManager{restartProcessErr: errors.New("provider reset failed")} + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentManager) + svc.logger = log + svc.repo = failSetSessionMetadataRepo{repoStore: repo} + svc.eventBus = eventBus + svc.messageCreator = &mockMessageCreator{} + + svc.processOnEnter(context.Background(), "t1", session, workflowResetFailureTestStep(), "review task", 0, nil) + + if logs.FilterMessage("failed to persist workflow context reset failure").Len() == 0 { + t.Fatal("expected a persistence failure log") + } + stored, err := repo.GetTaskSession(context.Background(), "s1") + if err != nil { + t.Fatalf("reload session: %v", err) + } + if stored.State != models.TaskSessionStateWaitingForInput { + t.Fatalf("session state = %s, want WAITING_FOR_INPUT after persistence failure", stored.State) + } + for _, published := range eventBus.published() { + if published.Subject == events.TaskSessionErrorChanged { + t.Fatal("persistence failure must not publish a false active error event") + } + } +} + +func TestProcessOnEnter_ResetFailureSettlesStateAfterPersistenceBudgetExpires(t *testing.T) { + previousTimeout := workflowResetFailureCleanupTimeout + workflowResetFailureCleanupTimeout = 25 * time.Millisecond + t.Cleanup(func() { workflowResetFailureCleanupTimeout = previousTimeout }) + + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + seedSessionRecord, err := repo.GetTaskSession(context.Background(), "s1") + require.NoError(t, err) + seedExecutorRunning(t, repo, seedSessionRecord.ID, seedSessionRecord.TaskID, "exec-reset") + + barrier := &workflowResetFailureBarrierRepo{ + repoStore: repo, + entered: make(chan struct{}), + release: make(chan struct{}), + } + manager := &mockAgentManager{restartProcessErr: errors.New("provider reset failed")} + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), manager) + svc.repo = barrier + svc.eventBus = &mockEventBus{} + svc.executor = executor.NewExecutor(manager, repo, testLogger(), executor.ExecutorConfig{}) + + resetDone := make(chan struct{}) + go func() { + svc.processOnEnter( + context.Background(), "t1", seedSessionRecord, workflowResetFailureTestStep(), + "review task", 0, nil, + ) + close(resetDone) + }() + waitForWorkflowResetBarrier(t, barrier.entered) + + select { + case <-resetDone: + case <-time.After(time.Second): + t.Fatal("workflow reset failure did not settle after metadata persistence timed out") + } + + stored, err := repo.GetTaskSession(context.Background(), "s1") + require.NoError(t, err) + require.Equal(t, models.TaskSessionStateWaitingForInput, stored.State) +} + +func TestProcessOnEnter_SuccessfulResetDispatchesAutoStartPrompt(t *testing.T) { + svc, repo, manager, session := newActiveResetTestService(t) + manager.isAgentRunning = true + svc.executor = executor.NewExecutor(manager, repo, testLogger(), executor.ExecutorConfig{}) + messages := &mockMessageCreator{} + svc.messageCreator = messages + + svc.processOnEnter( + context.Background(), "task1", session, workflowResetFailureTestStep(), + "review task", 0, nil, + ) + + requireResetEvents(t, manager.events, "cancel", "reset") + if len(manager.capturedPrompts) != 1 { + t.Fatalf("successful workflow entry dispatched %d prompts, want 1", len(manager.capturedPrompts)) + } + if !strings.Contains(manager.capturedPrompts[0], "distinct reset failure workflow prompt") { + t.Fatalf("dispatched prompt = %q, want the distinct step prompt", manager.capturedPrompts[0]) + } + if len(messages.userMessages) != 1 { + t.Fatalf("successful workflow entry persisted %d user messages, want 1", len(messages.userMessages)) + } +} + +func TestProcessOnEnter_ResetFailureSettlementOwnsSuccessorAdmission(t *testing.T) { + t.Run("successor admission waits for failure settlement", func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + seedExecutorRunning(t, repo, "s1", "t1", "exec-reset") + session, err := repo.GetTaskSession(context.Background(), "s1") + require.NoError(t, err) + session.State = models.TaskSessionStateWaitingForInput + require.NoError(t, repo.UpdateTaskSession(context.Background(), session)) + session, err = repo.GetTaskSession(context.Background(), "s1") + require.NoError(t, err) + + barrier := &workflowResetFailureBarrierRepo{ + repoStore: repo, + entered: make(chan struct{}), + release: make(chan struct{}), + } + manager := &mockAgentManager{restartProcessErr: errors.New("provider reset failed")} + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), manager) + svc.repo = barrier + svc.eventBus = &mockEventBus{} + svc.executor = executor.NewExecutor(manager, repo, testLogger(), executor.ExecutorConfig{}) + + resetDone := make(chan struct{}) + go func() { + svc.processOnEnter(context.Background(), "t1", session, workflowResetFailureTestStep(), "review task", 0, nil) + close(resetDone) + }() + waitForWorkflowResetBarrier(t, barrier.entered) + + successorAdmitted := make(chan struct{}) + successorDone := make(chan error, 1) + go func() { + lock, release := svc.acquireCancelInFlightGuard("s1") + lock.Lock() + close(successorAdmitted) + successorDone <- repo.UpdateTaskSessionState( + context.Background(), "s1", models.TaskSessionStateRunning, "successor turn", + ) + lock.Unlock() + release() + }() + assertWorkflowResetChannelBlocked(t, successorAdmitted, "successor admission") + + close(barrier.release) + <-resetDone + require.NoError(t, <-successorDone) + stored, err := repo.GetTaskSession(context.Background(), "s1") + require.NoError(t, err) + require.Equal(t, models.TaskSessionStateRunning, stored.State) + if _, ok := models.LoadLastAgentError(stored.Metadata); !ok { + t.Fatal("failure settlement lost the durable error before successor admission") + } + }) + + t.Run("deletion waits for failure settlement", func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + seedExecutorRunning(t, repo, "s1", "t1", "exec-reset") + session, err := repo.GetTaskSession(context.Background(), "s1") + require.NoError(t, err) + session.State = models.TaskSessionStateWaitingForInput + require.NoError(t, repo.UpdateTaskSession(context.Background(), session)) + session, err = repo.GetTaskSession(context.Background(), "s1") + require.NoError(t, err) + + barrier := &workflowResetFailureBarrierRepo{ + repoStore: repo, + entered: make(chan struct{}), + release: make(chan struct{}), + } + manager := &mockAgentManager{restartProcessErr: errors.New("provider reset failed")} + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), manager) + svc.repo = barrier + svc.eventBus = &mockEventBus{} + svc.executor = executor.NewExecutor(manager, repo, testLogger(), executor.ExecutorConfig{}) + + resetDone := make(chan struct{}) + go func() { + svc.processOnEnter(context.Background(), "t1", session, workflowResetFailureTestStep(), "review task", 0, nil) + close(resetDone) + }() + waitForWorkflowResetBarrier(t, barrier.entered) + + deleteDone := make(chan error, 1) + go func() { deleteDone <- svc.DeleteSession(context.Background(), "s1") }() + assertWorkflowResetChannelBlocked(t, deleteDone, "session deletion") + + close(barrier.release) + <-resetDone + require.NoError(t, <-deleteDone) + _, err = repo.GetTaskSession(context.Background(), "s1") + require.ErrorIs(t, err, models.ErrTaskSessionNotFound) + }) +} + +func TestResetAgentContext_BlocksDeletionWhileProviderCallIsInFlight(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + seedExecutorRunning(t, repo, "s1", "t1", "exec-reset") + session, err := repo.GetTaskSession(context.Background(), "s1") + require.NoError(t, err) + session.State = models.TaskSessionStateWaitingForInput + require.NoError(t, repo.UpdateTaskSession(context.Background(), session)) + + manager := &blockingWorkflowResetAgentManager{ + mockAgentManager: &mockAgentManager{repoForExecutionLookup: repo}, + resetStarted: make(chan struct{}), + resetRelease: make(chan struct{}), + } + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), manager) + svc.executor = executor.NewExecutor(manager, repo, testLogger(), executor.ExecutorConfig{}) + + resetDone := make(chan error, 1) + go func() { + _, resetErr := svc.resetAgentContextWithError( + context.Background(), "t1", session, "Review Step", + ) + resetDone <- resetErr + }() + waitForWorkflowResetBarrier(t, manager.resetStarted) + + deleteErr := svc.DeleteSession(context.Background(), "s1") + require.ErrorIs(t, deleteErr, ErrSessionResetInProgress) + _, err = repo.GetTaskSession(context.Background(), "s1") + require.NoError(t, err, "deletion must not remove a session during provider reset") + + close(manager.resetRelease) + require.NoError(t, <-resetDone) + require.NoError(t, svc.DeleteSession(context.Background(), "s1")) + _, err = repo.GetTaskSession(context.Background(), "s1") + require.ErrorIs(t, err, models.ErrTaskSessionNotFound) +} + +func waitForWorkflowResetBarrier(t *testing.T, entered <-chan struct{}) { + t.Helper() + select { + case <-entered: + case <-time.After(time.Second): + t.Fatal("timed out waiting for workflow reset failure persistence barrier") + } +} + +func assertWorkflowResetChannelBlocked[T any](t *testing.T, channel <-chan T, operation string) { + t.Helper() + select { + case value := <-channel: + t.Fatalf("%s completed before reset failure settlement: %#v", operation, value) + case <-time.After(50 * time.Millisecond): + } +} + +func workflowResetFailureTestStep() *wfmodels.WorkflowStep { + return &wfmodels.WorkflowStep{ + ID: "step2", WorkflowID: "wf1", Name: "Review Step", + Prompt: "distinct reset failure workflow prompt", + Events: wfmodels.StepEvents{OnEnter: []wfmodels.OnEnterAction{ + {Type: wfmodels.OnEnterResetAgentContext}, + {Type: wfmodels.OnEnterAutoStartAgent}, + }}, + } +} + +func assertResetFailureFeedback( + t *testing.T, + repo *sqliterepo.Repository, + eventBus *mockEventBus, + agentManager *mockAgentManager, + messages *mockMessageCreator, + expectedExecutionID string, +) { + t.Helper() + ctx := context.Background() + stored, err := repo.GetTaskSession(ctx, "s1") + if err != nil { + t.Fatalf("reload session: %v", err) + } + lastError, ok := models.LoadLastAgentError(stored.Metadata) + if !ok { + t.Fatal("expected a durable workflow reset error notice") + } + message := strings.ToLower(lastError.Message) + if !strings.Contains(message, "context reset") || !strings.Contains(message, "step prompt") { + t.Fatalf("reset error message = %q, want reset and prompt details", lastError.Message) + } + if lastError.AgentExecutionID != expectedExecutionID { + t.Fatalf("error execution ID = %q, want %q", lastError.AgentExecutionID, expectedExecutionID) + } + if lastError.Code != workflowResetFailureCode { + t.Fatalf("error code = %q, want %q", lastError.Code, workflowResetFailureCode) + } + if strings.Contains(lastError.Details, "stale reset notice") { + t.Fatalf("failure details retained stale snapshot data: %q", lastError.Details) + } + + if len(agentManager.capturedPrompts) != 0 { + t.Fatalf("failed workflow entry dispatched %d prompts", len(agentManager.capturedPrompts)) + } + if len(messages.userMessages) != 0 { + t.Fatalf("failed workflow entry persisted %d user messages", len(messages.userMessages)) + } + + var errorEvent, finalWaitingEvent *publishedEvent + publishedEvents := eventBus.published() + for i := range publishedEvents { + published := publishedEvents[i] + switch published.Subject { + case events.TaskSessionErrorChanged: + errorEvent = &published + case events.TaskSessionStateChanged: + data, _ := published.Event.Data.(map[string]interface{}) + if data["workflow_step_id"] == "step2" { + finalWaitingEvent = &published + } + } + } + if errorEvent == nil { + t.Fatal("expected a published last-agent-error event") + } + errorData, _ := errorEvent.Event.Data.(map[string]interface{}) + if active, _ := errorData["active"].(bool); !active { + t.Fatalf("last-agent-error event data = %#v, want active=true", errorData) + } + if errorData["code"] != workflowResetFailureCode { + t.Fatalf("last-agent-error event code = %#v, want %q", errorData["code"], workflowResetFailureCode) + } + if finalWaitingEvent == nil { + t.Fatal("expected a final waiting-state event for the destination step") + } + waitingData, _ := finalWaitingEvent.Event.Data.(map[string]interface{}) + metadata, ok := waitingData["session_metadata"].(map[string]interface{}) + if !ok { + t.Fatalf("waiting event metadata = %#v, want fresh session metadata", waitingData["session_metadata"]) + } + waitingError, ok := models.LoadLastAgentError(metadata) + if !ok || waitingError.Message != lastError.Message { + t.Fatalf("waiting event omitted the persisted reset error: %#v", metadata) + } +} diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow_reset_quiescence_test.go b/apps/backend/internal/orchestrator/event_handlers_workflow_reset_quiescence_test.go index e67c0bb8c3..8f40a251e9 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow_reset_quiescence_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow_reset_quiescence_test.go @@ -7,6 +7,7 @@ import ( "testing" "time" + "github.com/kandev/kandev/internal/agent/runtime/lifecycle" "github.com/kandev/kandev/internal/orchestrator/executor" "github.com/kandev/kandev/internal/task/models" sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" @@ -245,6 +246,108 @@ func TestResetAgentContext_CancelFailureStopsProviderReset(t *testing.T) { } } +func TestResetAgentContext_EscalatedCancelStopsProviderReset(t *testing.T) { + svc, _, manager, session := newActiveResetTestService(t) + manager.cancelAgentErr = lifecycle.ErrCancelEscalated + + if svc.resetAgentContext(context.Background(), "task1", session, "Successor") { + t.Fatal("resetAgentContext returned true after cancellation escalation") + } + + requireResetEvents(t, manager.events, "cancel") + select { + case event := <-manager.events: + t.Fatalf("provider operation %q started after cancellation escalation", event) + default: + } +} + +func TestResetAgentContext_WrappedEscalatedCancelStopsProviderReset(t *testing.T) { + svc, _, manager, session := newActiveResetTestService(t) + manager.cancelAgentErr = errors.Join(errors.New("provider timeout"), lifecycle.ErrCancelEscalated) + + if svc.resetAgentContext(context.Background(), "task1", session, "Successor") { + t.Fatal("resetAgentContext returned true after wrapped cancellation escalation") + } + + requireResetEvents(t, manager.events, "cancel") + select { + case event := <-manager.events: + t.Fatalf("provider operation %q started after wrapped cancellation escalation", event) + default: + } +} + +func TestResetAgentContext_MissingExecutionStillReconcilesBeforeProviderReset(t *testing.T) { + svc, _, manager, session := newActiveResetTestService(t) + manager.cancelAgentErr = lifecycle.ErrNoExecutionForSession + + if !svc.resetAgentContext(context.Background(), "task1", session, "Successor") { + t.Fatal("resetAgentContext returned false after missing execution cancellation") + } + + requireResetEvents(t, manager.events, "cancel", "reset") + if active, err := svc.hasActiveResetTurn(context.Background(), session.ID); err != nil { + t.Fatalf("hasActiveResetTurn: %v", err) + } else if active { + t.Fatal("cancelled turn remained active after missing execution reconciliation") + } +} + +func TestResetAgentContext_JoinedExplicitCancelKeepsSuccessfulCancellationSemantics(t *testing.T) { + svc, _, manager, session := newActiveResetTestService(t) + manager.cancelAgentErr = lifecycle.ErrCancelEscalated + svc.messageCreator = &mockMessageCreator{} + cancelEntered := make(chan struct{}, 1) + cancelRelease := make(chan struct{}) + manager.cancelAgentEntered = cancelEntered + manager.cancelAgentBlock = cancelRelease + + resetDone := make(chan bool, 1) + go func() { + resetDone <- svc.resetAgentContext(context.Background(), session.TaskID, session, "Successor") + }() + select { + case <-cancelEntered: + case <-time.After(time.Second): + t.Fatal("timed out waiting for reset cancellation") + } + + joinedDone := make(chan error, 1) + go func() { + joinedDone <- svc.CancelAgent(context.Background(), session.ID) + }() + operation := svc.currentCancellation(session.ID) + if operation == nil { + t.Fatal("expected reset cancellation operation to remain registered") + } + select { + case <-operation.joined: + case <-time.After(time.Second): + t.Fatal("timed out waiting for explicit cancellation to join reset cancellation") + } + + close(cancelRelease) + if resetOK := <-resetDone; resetOK { + t.Fatal("resetAgentContext returned true after cancellation escalation") + } + select { + case err := <-joinedDone: + if err != nil { + t.Fatalf("joined explicit cancellation returned error: %v", err) + } + case <-time.After(time.Second): + t.Fatal("timed out waiting for joined explicit cancellation") + } + + requireResetEvents(t, manager.events, "cancel") + select { + case event := <-manager.events: + t.Fatalf("provider operation %q started after cancellation escalation", event) + default: + } +} + // TestResetAgentContext_ResetMarkerPrecedesCancellationWait fixes the ordering // that made TestResetAgentContext_SerializesPromptAdmission flaky. Once reset // has published its marker and yielded the session guard for cancellation, a diff --git a/apps/backend/internal/orchestrator/stuck_signal_watchdog.go b/apps/backend/internal/orchestrator/stuck_signal_watchdog.go index 5c839ca787..f645c41410 100644 --- a/apps/backend/internal/orchestrator/stuck_signal_watchdog.go +++ b/apps/backend/internal/orchestrator/stuck_signal_watchdog.go @@ -310,7 +310,7 @@ func (s *Service) reclaimStuckSignalSessionOwned( // (ErrNoExecutionForSession, ErrCancelEscalated) means the execution // could not be confirmed settled; fail closed and skip this tick rather // than force-closing the turn anyway. - if err := s.cancelAgentWhileUnlockedForPrompt(ctx, session.ID, identity, guard.unlock, guard.relock); err != nil { + if err := s.cancelAgentWhileUnlockedForPrompt(ctx, session.ID, identity, operation, guard.unlock, guard.relock); err != nil { s.logger.Warn("stuck-signal watchdog: failed to settle stuck execution before reclaim; skipping this tick", zap.String("task_id", task.ID), zap.String("session_id", session.ID), @@ -663,10 +663,11 @@ func (s *Service) cancelAgentWhileUnlockedForPrompt( ctx context.Context, sessionID string, identity cancellationIdentity, + operation *cancelOperation, unlockGuard, relockGuard func(), ) error { if identity.executionID == "" { - return s.cancelAgentWhileUnlocked(ctx, sessionID, unlockGuard, relockGuard) + return s.cancelAgentWhileUnlocked(ctx, sessionID, operation, unlockGuard, relockGuard) } canceller, ok := s.agentManager.(promptActivityCanceller) if !ok { @@ -684,6 +685,7 @@ func (s *Service) cancelAgentWhileUnlockedForPrompt( identity.activityEpoch, ) relockGuard() + s.setCancellationProviderOutcome(sessionID, operation, cancelErr) return s.normalizeCancelAgentError(cancelErr) } diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index 0ae42a3a6b..4b6f71d569 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -4318,6 +4318,9 @@ func (s *Service) DeleteSession(ctx context.Context, sessionID string) error { defer release() lock.Lock() defer lock.Unlock() + if s.isSessionResetInProgress(sessionID) { + return ErrSessionResetInProgress + } session, err := s.repo.GetTaskSession(ctx, sessionID) if err != nil { @@ -7306,6 +7309,8 @@ type cancelOperation struct { joined chan struct{} joinObserved bool err error + providerCancelErr error + providerCancelOutcomeReady bool projectionRelease func() kind cancellationKind identity cancellationIdentity @@ -7575,6 +7580,24 @@ func (s *Service) cancellationPreparationSnapshot(operation *cancelOperation) (c return operation.identity, operation.completionEligible, operation.completionEligibilityReady } +func (s *Service) setCancellationProviderOutcome(sessionID string, operation *cancelOperation, err error) { + s.cancellationOperationsMu.Lock() + defer s.cancellationOperationsMu.Unlock() + if current := s.cancellationOperations[sessionID]; current == operation { + operation.providerCancelErr = err + operation.providerCancelOutcomeReady = true + } +} + +func (s *Service) cancellationProviderOutcomeSnapshot(operation *cancelOperation) (error, bool) { + s.cancellationOperationsMu.Lock() + defer s.cancellationOperationsMu.Unlock() + if operation == nil { + return nil, false + } + return operation.providerCancelErr, operation.providerCancelOutcomeReady +} + func (s *Service) finishCancellation(sessionID string, operation *cancelOperation, err error) { s.finishCancellationWithActions(context.Background(), sessionID, operation, err) } @@ -8146,7 +8169,7 @@ func (s *Service) runExplicitCancellationOwned(ctx context.Context, sessionID st } s.setCancellationIdentity(sessionID, operation, prepared.identity) s.setCancellationCompletionEligible(sessionID, operation, prepared.completionEligible) - if err := s.cancelAgentWhileUnlocked(ctx, sessionID, unlockGuard, relockGuard); err != nil { + if err := s.cancelAgentWhileUnlocked(ctx, sessionID, operation, unlockGuard, relockGuard); err != nil { return err } if err := s.finishCancelledAgentTurn(ctx, sessionID, prepared); err != nil { @@ -8299,7 +8322,12 @@ func (s *Service) cancelTurnCompletionEligible(ctx context.Context, session *mod } } -func (s *Service) cancelAgentWhileUnlocked(ctx context.Context, sessionID string, unlockGuard, relockGuard func()) error { +func (s *Service) cancelAgentWhileUnlocked( + ctx context.Context, + sessionID string, + operation *cancelOperation, + unlockGuard, relockGuard func(), +) error { if s.agentManager == nil { return nil } @@ -8310,6 +8338,7 @@ func (s *Service) cancelAgentWhileUnlocked(ctx context.Context, sessionID string unlockGuard() cancelErr := s.agentManager.CancelAgent(ctx, sessionID) relockGuard() + s.setCancellationProviderOutcome(sessionID, operation, cancelErr) if cancelErr == nil { return nil } diff --git a/apps/backend/internal/task/repository/sqlite/plan_comment_test.go b/apps/backend/internal/task/repository/sqlite/plan_comment_test.go index ff57628017..edbf151c00 100644 --- a/apps/backend/internal/task/repository/sqlite/plan_comment_test.go +++ b/apps/backend/internal/task/repository/sqlite/plan_comment_test.go @@ -1,3 +1,5 @@ +//go:build cgo + package sqlite import ( diff --git a/apps/web/components/task/chat/message-list-shared.test.tsx b/apps/web/components/task/chat/message-list-shared.test.tsx index 2e71fad32f..67e7af3737 100644 --- a/apps/web/components/task/chat/message-list-shared.test.tsx +++ b/apps/web/components/task/chat/message-list-shared.test.tsx @@ -349,7 +349,12 @@ describe("MessageItem agent error notice", () => { const REMEDIATION_URL = "https://opencode.ai/workspace/wrk_01KQM7K5CYT715264YKKFB17ZY/go"; - function renderNotice(error: { message: string; occurredAt?: string; remediationUrl?: string }) { + function renderNotice(error: { + message: string; + occurredAt?: string; + remediationUrl?: string; + details?: string; + }) { render( { ); }); + it("uses a touch-sized dismiss target on coarse pointers", () => { + renderNotice({ message: AGENT_ERROR_MESSAGE }); + + expect(screen.getByRole("button").className).toContain("[@media(pointer:coarse)]:h-11"); + expect(screen.getByRole("button").className).toContain("[@media(pointer:coarse)]:w-11"); + }); + + it("reveals the sanitized failure cause in a collapsed technical-details disclosure", () => { + const details = 'workflow step "Review Step": provider context reset: provider reset timed out'; + renderNotice({ message: "context reset failed", details }); + + const disclosure = screen.getByTestId("last-agent-error-details") as HTMLDetailsElement; + expect(disclosure.querySelector("summary")?.textContent).toBe("Technical details"); + expect(disclosure.open).toBe(false); + + fireEvent.click(screen.getByText("Technical details")); + expect(disclosure.open).toBe(true); + expect(disclosure.textContent).toContain(details); + }); + it("renders a validated remediation link, and nothing for an invalid URL", () => { renderNotice({ message: "usage limit reached", diff --git a/apps/web/components/task/chat/message-list-shared.tsx b/apps/web/components/task/chat/message-list-shared.tsx index 40ce20c167..7e7923e42c 100644 --- a/apps/web/components/task/chat/message-list-shared.tsx +++ b/apps/web/components/task/chat/message-list-shared.tsx @@ -438,6 +438,14 @@ export function LastAgentErrorNotice({ 
             {error.message}
+ {error.details && ( +
+ {t("task:technicalDetails")} + 
+                {error.details}
+
+
+ )} {error.remediationUrl && (
@@ -446,7 +454,7 @@ export function LastAgentErrorNotice({