From 8a2404251db79ac8ff6f2d4e04655a95fdd43472 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Sat, 11 Jul 2026 12:31:21 +0100 Subject: [PATCH] fix: guard agent profile orphan cleanup (#1659) --- .../internal/agent/registry/provider.go | 1 + .../internal/agent/registry/provider_test.go | 3 + .../internal/agent/registry/registry.go | 16 + .../internal/agent/registry/registry_test.go | 9 + .../agent/runtime/lifecycle/session.go | 15 +- .../agent/runtime/lifecycle/session_test.go | 60 ++++ .../settings/controller/agent_install_test.go | 18 +- .../agent/settings/controller/reconciler.go | 156 ++++++++- .../settings/controller/reconciler_test.go | 297 ++++++++++++++++-- .../adapter/transport/acp/adapter_prompt.go | 13 +- .../acp/adapter_prompt_cancel_test.go | 26 ++ .../server/adapter/transport/acp/errors.go | 16 +- .../internal/agentctl/server/api/agent.go | 5 + .../agentctl/server/api/agent_test.go | 109 +++++++ .../server/process/testing_helpers.go | 14 +- .../orchestrator/event_handlers_agent.go | 4 +- .../orchestrator/event_handlers_queue_test.go | 45 +++ .../orchestrator/event_handlers_test.go | 7 +- .../internal/orchestrator/task_operations.go | 5 + .../orchestrator/task_operations_test.go | 42 +++ 20 files changed, 817 insertions(+), 44 deletions(-) diff --git a/apps/backend/internal/agent/registry/provider.go b/apps/backend/internal/agent/registry/provider.go index 96c495b2df..6c89a7ec89 100644 --- a/apps/backend/internal/agent/registry/provider.go +++ b/apps/backend/internal/agent/registry/provider.go @@ -56,6 +56,7 @@ func Provide(log *logger.Logger) (*Registry, func() error, error) { } registerRealProvidersProber(reg, log) } + reg.MarkLoaded() return reg, func() error { return nil }, nil } diff --git a/apps/backend/internal/agent/registry/provider_test.go b/apps/backend/internal/agent/registry/provider_test.go index e0c55a37be..7e3d13de24 100644 --- a/apps/backend/internal/agent/registry/provider_test.go +++ b/apps/backend/internal/agent/registry/provider_test.go @@ -53,6 +53,9 @@ func TestProvide_MockAgentModes(t *testing.T) { t.Fatalf("Provide() error: %v", err) } defer cleanup() //nolint:errcheck + if !reg.IsLoaded() { + t.Fatal("Provide should return a loaded registry") + } // Check mock-agent presence and enabled state mock, hasMock := reg.Get("mock-agent") diff --git a/apps/backend/internal/agent/registry/registry.go b/apps/backend/internal/agent/registry/registry.go index dad9224074..084b917573 100644 --- a/apps/backend/internal/agent/registry/registry.go +++ b/apps/backend/internal/agent/registry/registry.go @@ -29,6 +29,7 @@ type ( // Registry manages agent configurations type Registry struct { agents map[string]agents.Agent + loaded bool mu sync.RWMutex logger *logger.Logger } @@ -73,6 +74,21 @@ func (r *Registry) LoadDefaults() { r.agents[ag.ID()] = ag r.logger.Debug("loaded default agent type", zap.String("id", ag.ID())) } + r.loaded = true +} + +// MarkLoaded records that the initial registry population completed. +func (r *Registry) MarkLoaded() { + r.mu.Lock() + defer r.mu.Unlock() + r.loaded = true +} + +// IsLoaded reports whether the initial registry population completed. +func (r *Registry) IsLoaded() bool { + r.mu.RLock() + defer r.mu.RUnlock() + return r.loaded } // Get returns an agent by ID diff --git a/apps/backend/internal/agent/registry/registry_test.go b/apps/backend/internal/agent/registry/registry_test.go index f01d0b2e10..207773f51b 100644 --- a/apps/backend/internal/agent/registry/registry_test.go +++ b/apps/backend/internal/agent/registry/registry_test.go @@ -75,6 +75,9 @@ func TestNewRegistry(t *testing.T) { } else if len(reg.agents) != 0 { t.Errorf("expected empty agents map, got %d", len(reg.agents)) } + if reg.IsLoaded() { + t.Error("new registry should not be marked loaded") + } } func TestRegistry_Register(t *testing.T) { @@ -94,6 +97,9 @@ func TestRegistry_Register(t *testing.T) { if err == nil { t.Error("expected error for duplicate registration") } + if reg.IsLoaded() { + t.Error("manual registration should not mark initial registry load complete") + } } func TestRegistry_RegisterValidation(t *testing.T) { @@ -233,6 +239,9 @@ func TestRegistry_LoadDefaults(t *testing.T) { reg := NewRegistry(log) reg.LoadDefaults() + if !reg.IsLoaded() { + t.Fatal("LoadDefaults should mark registry loaded") + } // Verify at least some default agents are loaded list := reg.List() diff --git a/apps/backend/internal/agent/runtime/lifecycle/session.go b/apps/backend/internal/agent/runtime/lifecycle/session.go index e305c2e7a8..957e68160b 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/session.go +++ b/apps/backend/internal/agent/runtime/lifecycle/session.go @@ -519,10 +519,10 @@ func (sm *SessionManager) waitForPromptDone(ctx context.Context, execution *Agen sm.logger.Error("prompt completed with error", zap.String("execution_id", execution.ID), zap.String("error", signal.Error)) - // Wrap the cancel-escalation sentinel so PromptTask can identify it and + // Wrap cancel-release sentinels so PromptTask can identify them and // skip the REVIEW task-state transition — the user is cancelling, not // hitting a real agent failure. - if strings.HasPrefix(signal.Error, "cancel escalated") { + if isCancelReleaseError(signal.Error) { return nil, fmt.Errorf("%w: %s: %w", ErrAgentReported, signal.Error, ErrCancelEscalated) } return nil, fmt.Errorf("%w: %s", ErrAgentReported, signal.Error) @@ -563,6 +563,11 @@ func (sm *SessionManager) waitForPromptDone(ctx context.Context, execution *Agen } } +func isCancelReleaseError(msg string) bool { + return strings.HasPrefix(msg, "cancel escalated") || + strings.Contains(msg, "prompt abandoned after cancel") +} + // SendPrompt sends a prompt to an agent execution and waits for completion. // For initial prompts, pass validateStatus=false. For follow-up prompts, pass validateStatus=true. // When dispatchOnly is true, returns once agentctl.Prompt has accepted the prompt @@ -664,6 +669,12 @@ func (sm *SessionManager) SendPrompt( zap.String("execution_id", execution.ID), zap.Error(retryErr)) } + if isCancelReleaseError(err.Error()) { + sm.logger.Info("prompt trigger abandoned after cancel; requeueing", + zap.String("execution_id", execution.ID), + zap.Error(err)) + return nil, fmt.Errorf("failed to trigger prompt: %w: %w", err, ErrCancelEscalated) + } sm.logger.Error("failed to trigger prompt", zap.String("execution_id", execution.ID), zap.Error(err)) diff --git a/apps/backend/internal/agent/runtime/lifecycle/session_test.go b/apps/backend/internal/agent/runtime/lifecycle/session_test.go index 5c8da94639..03a4b9dc6f 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/session_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/session_test.go @@ -3,6 +3,7 @@ package lifecycle import ( "context" "encoding/json" + "errors" "fmt" "net" "net/http" @@ -935,6 +936,65 @@ func TestSendPrompt_DrainsStaleSignalFromPriorDispatchOnly(t *testing.T) { } } +func TestWaitForPromptDone_TreatsPromptAbandonedAfterCancelAsCancelEscalated(t *testing.T) { + log := newSessionTestLogger() + sm := NewSessionManager(log, make(chan struct{})) + execution := &AgentExecution{ + ID: "test-exec", + promptDoneCh: make(chan PromptCompletionSignal, 1), + } + execution.promptDoneCh <- PromptCompletionSignal{ + IsError: true, + Error: "prompt abandoned after cancel", + } + + _, err := sm.waitForPromptDone(context.Background(), execution) + if !errors.Is(err, ErrCancelEscalated) { + t.Fatalf("expected ErrCancelEscalated, got: %v", err) + } + if !errors.Is(err, ErrAgentReported) { + t.Fatalf("expected ErrAgentReported wrapper, got: %v", err) + } +} + +func TestSendPrompt_TriggerTimeCancelReleaseReturnsErrCancelEscalated(t *testing.T) { + mock := newMockAgentServer(t) + defer mock.Close() + mock.handler = func(msg ws.Message) *ws.Message { + if msg.Action == "agent.prompt" { + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "prompt abandoned after cancel", nil) + return resp + } + return mock.defaultHandler(msg) + } + + log := newSessionTestLogger() + sm := NewSessionManager(log, make(chan struct{})) + + client := createTestClient(t, mock.server.URL) + defer client.Close() + + ctx := context.Background() + if err := client.StreamUpdates(ctx, func(event agentctl.AgentEvent) {}, nil, nil); err != nil { + t.Fatalf("failed to connect stream: %v", err) + } + waitForWSConnected(t, mock) + + execution := &AgentExecution{ + ID: "test-exec", + TaskID: "test-task", + SessionID: "test-session", + WorkspacePath: "/workspace", + agentctl: client, + promptDoneCh: make(chan PromptCompletionSignal, 1), + } + + _, err := sm.SendPrompt(ctx, execution, "hello", false, nil, true) + if !errors.Is(err, ErrCancelEscalated) { + t.Fatalf("expected ErrCancelEscalated, got: %v", err) + } +} + // TestBuildEffectivePrompt_DoesNotInjectKandevContextOnResume verifies the // lifecycle layer no longer wraps follow-up prompts with the Kandev system // block. The orchestrator wraps the first prompt only; on resume the agent diff --git a/apps/backend/internal/agent/settings/controller/agent_install_test.go b/apps/backend/internal/agent/settings/controller/agent_install_test.go index a3c50b73d4..38e52be016 100644 --- a/apps/backend/internal/agent/settings/controller/agent_install_test.go +++ b/apps/backend/internal/agent/settings/controller/agent_install_test.go @@ -43,6 +43,22 @@ func (b *captureBroadcaster) actions() []string { return out } +func (b *captureBroadcaster) waitForAction(t *testing.T, action string) []string { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + actions := b.actions() + for _, got := range actions { + if got == action { + return actions + } + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("broadcast action %s did not arrive in time; got %v", action, b.actions()) + return nil +} + // withStubStreamingRunner swaps streamingInstallRunner for the duration of // the test. The stub invokes onChunk synchronously to make ordering deterministic. func withStubStreamingRunner(t *testing.T, fn func(ctx context.Context, script string, onChunk func(string)) error) { @@ -111,7 +127,7 @@ func TestEnqueueInstall_StreamsAndSucceeds(t *testing.T) { // Must have broadcast a started and a finished message; output messages // land in between but exact count depends on the flush timing. - actions := hub.actions() + actions := hub.waitForAction(t, ws.ActionAgentInstallFinished) if len(actions) < 2 { t.Fatalf("expected ≥2 broadcasts, got %v", actions) } diff --git a/apps/backend/internal/agent/settings/controller/reconciler.go b/apps/backend/internal/agent/settings/controller/reconciler.go index ab3264ebfc..b96fbd7931 100644 --- a/apps/backend/internal/agent/settings/controller/reconciler.go +++ b/apps/backend/internal/agent/settings/controller/reconciler.go @@ -16,6 +16,13 @@ import ( "github.com/kandev/kandev/internal/common/logger" ) +const ( + // Orphan cleanup is startup hygiene, not a user-confirmed bulk action. + // Keep automatic batches small so registry bugs fail closed. + maxOrphanCleanupAgentTypesPerRun = 20 + maxOrphanCleanupProfilesPerRun = 50 +) + // CapabilityReader is the minimum surface of the host utility manager that // the reconciler needs. Declared as an interface so tests can inject a fake. type CapabilityReader interface { @@ -39,6 +46,27 @@ type ProfileReconciler struct { log *logger.Logger } +type orphanCleanupCandidate struct { + agent *models.Agent + profile *models.AgentProfile +} + +type orphanCleanupSummary struct { + enabledAgentCount int + dbAgentCount int + orphanAgentCount int + profilesCandidateCount int + profilesCandidatePartial bool + profilesDeletedCount int + profileListFailureCount int + skipped bool + skipReason string + enabledAgentIDs []string + orphanAgentNames []string + maxAgentTypesPerRun int + maxProfilesPerRun int +} + // NewProfileReconciler constructs a reconciler. func NewProfileReconciler( h CapabilityReader, @@ -63,7 +91,8 @@ func (r *ProfileReconciler) Run(ctx context.Context) error { } // Orphan cleanup first: removed agents can't come back, regardless of - // probe state, so we can clean these unconditionally. + // probe state. The cleanup itself fails closed when the registry is empty + // so a transient registry/bootstrap issue cannot mass-delete profiles. r.cleanupOrphans(ctx) // Walk enabled inference agents and reconcile each one. @@ -83,22 +112,82 @@ func (r *ProfileReconciler) Run(ctx context.Context) error { // each DB agent by a UUID in `id`, with the registry-facing identifier stored // in `name` — we match against the registry on `name`. func (r *ProfileReconciler) cleanupOrphans(ctx context.Context) { - enabled := make(map[string]struct{}) - for _, ag := range r.registry.ListEnabled() { - enabled[ag.ID()] = struct{}{} + summary := orphanCleanupSummary{ + maxAgentTypesPerRun: maxOrphanCleanupAgentTypesPerRun, + maxProfilesPerRun: maxOrphanCleanupProfilesPerRun, + } + defer func() { + r.logOrphanCleanupSummary(summary) + }() + + if !r.registry.IsLoaded() { + summary.skipped = true + summary.skipReason = "registry_not_loaded" + r.log.Warn("orphan cleanup skipped: registry is not loaded") + return + } + enabledAgents := r.registry.ListEnabled() + summary.enabledAgentCount = len(enabledAgents) + if len(enabledAgents) == 0 { + summary.skipped = true + summary.skipReason = "enabled_registry_empty" + r.log.Warn("orphan cleanup skipped: enabled agent registry is empty") + return + } + enabled := make(map[string]struct{}, len(enabledAgents)) + for _, ag := range enabledAgents { + id := ag.ID() + enabled[id] = struct{}{} + summary.enabledAgentIDs = append(summary.enabledAgentIDs, id) } dbAgents, err := r.store.ListAgents(ctx) if err != nil { + summary.skipped = true + summary.skipReason = "list_agents_failed" r.log.Warn("orphan cleanup: list agents failed", zap.Error(err)) return } + summary.dbAgentCount = len(dbAgents) + candidates := r.collectOrphanCleanupCandidates(ctx, dbAgents, enabled, &summary) + if summary.profileListFailureCount > 0 { + summary.skipped = true + summary.skipReason = "profile_list_failed" + r.log.Warn("orphan cleanup skipped: profile list failed for one or more orphan agents", + zap.Int("profile_list_failure_count", summary.profileListFailureCount)) + return + } + if exceedsOrphanCleanupLimit(summary.orphanAgentCount, len(candidates)) { + summary.skipped = true + summary.skipReason = "safety_limit_exceeded" + r.log.Warn("orphan cleanup skipped: candidate batch exceeds safety limit", + zap.Int("orphan_agent_count", summary.orphanAgentCount), + zap.Int("profiles_candidate_count", len(candidates)), + zap.Int("max_agent_types_per_run", maxOrphanCleanupAgentTypesPerRun), + zap.Int("max_profiles_per_run", maxOrphanCleanupProfilesPerRun), + zap.Strings("orphan_agents", summary.orphanAgentNames)) + return + } + r.deleteOrphanCleanupCandidates(ctx, candidates, &summary) +} + +func (r *ProfileReconciler) collectOrphanCleanupCandidates( + ctx context.Context, + dbAgents []*models.Agent, + enabled map[string]struct{}, + summary *orphanCleanupSummary, +) []orphanCleanupCandidate { + var candidates []orphanCleanupCandidate for _, dbAgent := range dbAgents { if _, ok := enabled[dbAgent.Name]; ok { continue } + summary.orphanAgentCount++ + summary.orphanAgentNames = append(summary.orphanAgentNames, dbAgent.Name) profiles, err := r.store.ListAgentProfiles(ctx, dbAgent.ID) if err != nil { + summary.profileListFailureCount++ + summary.profilesCandidatePartial = true r.log.Warn("orphan cleanup: list profiles failed", zap.String("agent_id", dbAgent.ID), zap.String("agent_name", dbAgent.Name), @@ -106,16 +195,59 @@ func (r *ProfileReconciler) cleanupOrphans(ctx context.Context) { continue } for _, p := range profiles { - r.log.Info("soft-deleting orphan profile", - zap.String("profile_id", p.ID), - zap.String("agent_id", p.AgentID), - zap.String("agent_name", dbAgent.Name)) - if err := r.store.DeleteAgentProfile(ctx, p.ID); err != nil { - r.log.Warn("orphan cleanup: delete failed", - zap.String("profile_id", p.ID), zap.Error(err)) - } + candidates = append(candidates, orphanCleanupCandidate{ + agent: dbAgent, + profile: p, + }) } } + if summary.profileListFailureCount == 0 { + summary.profilesCandidateCount = len(candidates) + } + return candidates +} + +func exceedsOrphanCleanupLimit(orphanAgentCount, profileCount int) bool { + return orphanAgentCount > maxOrphanCleanupAgentTypesPerRun || + profileCount > maxOrphanCleanupProfilesPerRun +} + +func (r *ProfileReconciler) deleteOrphanCleanupCandidates( + ctx context.Context, + candidates []orphanCleanupCandidate, + summary *orphanCleanupSummary, +) { + for _, candidate := range candidates { + r.log.Info("soft-deleting orphan profile", + zap.String("profile_id", candidate.profile.ID), + zap.String("agent_id", candidate.profile.AgentID), + zap.String("agent_name", candidate.agent.Name)) + if err := r.store.DeleteAgentProfile(ctx, candidate.profile.ID); err != nil { + r.log.Warn("orphan cleanup: delete failed", + zap.String("profile_id", candidate.profile.ID), zap.Error(err)) + continue + } + summary.profilesDeletedCount++ + } +} + +func (r *ProfileReconciler) logOrphanCleanupSummary(summary orphanCleanupSummary) { + fields := []zap.Field{ + zap.Int("enabled_agent_count", summary.enabledAgentCount), + zap.Int("db_agent_count", summary.dbAgentCount), + zap.Int("orphan_agent_count", summary.orphanAgentCount), + zap.Int("profiles_candidate_count", summary.profilesCandidateCount), + zap.Bool("profiles_candidate_partial", summary.profilesCandidatePartial), + zap.Int("profiles_deleted_count", summary.profilesDeletedCount), + zap.Int("profile_list_failure_count", summary.profileListFailureCount), + zap.Bool("skipped", summary.skipped), + zap.String("skip_reason", summary.skipReason), + zap.Int("max_agent_types_per_run", summary.maxAgentTypesPerRun), + zap.Int("max_profiles_per_run", summary.maxProfilesPerRun), + zap.Strings("enabled_agents", summary.enabledAgentIDs), + zap.Strings("orphan_agents", summary.orphanAgentNames), + } + r.log.Info("orphan cleanup summary", fields...) } // reconcileAgent validates or seeds profiles for a single inference agent. diff --git a/apps/backend/internal/agent/settings/controller/reconciler_test.go b/apps/backend/internal/agent/settings/controller/reconciler_test.go index 2c054b4230..efd3226d5a 100644 --- a/apps/backend/internal/agent/settings/controller/reconciler_test.go +++ b/apps/backend/internal/agent/settings/controller/reconciler_test.go @@ -3,8 +3,14 @@ package controller import ( "context" "database/sql" + "errors" + "strconv" "testing" + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "go.uber.org/zap/zaptest/observer" + "github.com/kandev/kandev/internal/agent/agents" "github.com/kandev/kandev/internal/agent/discovery" "github.com/kandev/kandev/internal/agent/hostutility" @@ -26,16 +32,18 @@ func (f *fakeCapReader) Get(agentType string) (hostutility.AgentCapabilities, bo // fakeStore implements just enough of store.Repository for the reconciler. type fakeStore struct { - agents map[string]*models.Agent // keyed by DB ID - byName map[string]*models.Agent // keyed by Name - profiles map[string][]*models.AgentProfile // keyed by DB agent ID (live) - deleted map[string][]*models.AgentProfile // keyed by DB agent ID (soft-deleted) - created []*models.AgentProfile - updated []*models.AgentProfile - softDeleted []string - nextAgentID int - nextProfID int - getByNameErr error + agents map[string]*models.Agent // keyed by DB ID + byName map[string]*models.Agent // keyed by Name + profiles map[string][]*models.AgentProfile // keyed by DB agent ID (live) + deleted map[string][]*models.AgentProfile // keyed by DB agent ID (soft-deleted) + created []*models.AgentProfile + updated []*models.AgentProfile + softDeleted []string + nextAgentID int + nextProfID int + getByNameErr error + listAgentsErr error + listProfErr map[string]error } func newFakeStore() *fakeStore { @@ -49,7 +57,7 @@ func newFakeStore() *fakeStore { func (f *fakeStore) CreateAgent(_ context.Context, a *models.Agent) error { f.nextAgentID++ - a.ID = "agent-" + itoa(f.nextAgentID) + a.ID = "agent-" + strconv.Itoa(f.nextAgentID) f.agents[a.ID] = a f.byName[a.Name] = a return nil @@ -73,6 +81,9 @@ func (f *fakeStore) UpdateAgent(context.Context, *models.Agent) error { return n func (f *fakeStore) DeleteAgent(context.Context, string) error { return nil } func (f *fakeStore) ListAgents(_ context.Context) ([]*models.Agent, error) { + if f.listAgentsErr != nil { + return nil, f.listAgentsErr + } out := make([]*models.Agent, 0, len(f.agents)) for _, a := range f.agents { out = append(out, a) @@ -93,7 +104,7 @@ func (f *fakeStore) UpsertAgentProfileMcpConfig(context.Context, *models.AgentPr func (f *fakeStore) CreateAgentProfile(_ context.Context, p *models.AgentProfile) error { f.nextProfID++ - p.ID = "profile-" + itoa(f.nextProfID) + p.ID = "profile-" + strconv.Itoa(f.nextProfID) f.profiles[p.AgentID] = append(f.profiles[p.AgentID], p) f.created = append(f.created, p) return nil @@ -155,23 +166,16 @@ func (f *fakeStore) GetAgentProfileIncludingDeleted(ctx context.Context, id stri } func (f *fakeStore) ListAgentProfiles(_ context.Context, agentID string) ([]*models.AgentProfile, error) { + if f.listProfErr != nil { + if err, ok := f.listProfErr[agentID]; ok { + return nil, err + } + } return f.profiles[agentID], nil } func (f *fakeStore) Close() error { return nil } -func itoa(n int) string { - if n == 0 { - return "0" - } - out := "" - for n > 0 { - out = string(rune('0'+n%10)) + out - n /= 10 - } - return out -} - // mockInferenceAgent is a minimal fake agent for the registry. type mockInferenceAgent struct { id string @@ -211,9 +215,20 @@ func newReconciler(t *testing.T, st *fakeStore, caps *fakeCapReader, ag agents.A if err := reg.Register(ag); err != nil { t.Fatalf("register: %v", err) } + reg.MarkLoaded() return NewProfileReconciler(caps, reg, st, log) } +func newObserverLogger(t *testing.T) (*logger.Logger, *observer.ObservedLogs) { + t.Helper() + core, logs := observer.New(zapcore.DebugLevel) + log, err := logger.NewFromZap(zap.New(core)) + if err != nil { + t.Fatalf("NewFromZap: %v", err) + } + return log, logs +} + func TestProfileReconciler_SeedsDefaultProfile(t *testing.T) { st := newFakeStore() ag := &mockInferenceAgent{id: "claude-acp", displayName: "Claude", enabled: true} @@ -392,6 +407,240 @@ func TestProfileReconciler_CleansOrphanProfiles(t *testing.T) { } } +func TestProfileReconciler_SkipsOrphanCleanupUntilRegistryReady(t *testing.T) { + st := newFakeStore() + orphanAgent := &models.Agent{Name: "removed-old-agent"} + _ = st.CreateAgent(context.Background(), orphanAgent) + orphanProfile := &models.AgentProfile{ + AgentID: orphanAgent.ID, + Name: "legacy", + Model: "x", + } + _ = st.CreateAgentProfile(context.Background(), orphanProfile) + + log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json"}) + if err != nil { + t.Fatalf("logger: %v", err) + } + reg := registry.NewRegistry(log) + ag := &mockInferenceAgent{id: "claude-acp", displayName: "Claude", enabled: true} + if err := reg.Register(ag); err != nil { + t.Fatalf("register: %v", err) + } + r := NewProfileReconciler(&fakeCapReader{caps: map[string]hostutility.AgentCapabilities{}}, reg, st, log) + if err := r.Run(context.Background()); err != nil { + t.Fatalf("Run: %v", err) + } + if len(st.softDeleted) != 0 { + t.Fatalf("expected not-ready registry cleanup to fail closed, got soft-deletes %v", st.softDeleted) + } +} + +func TestProfileReconciler_SkipsOrphanCleanupWhenEnabledRegistryEmpty(t *testing.T) { + st := newFakeStore() + claudeAgent := &models.Agent{Name: "claude-acp"} + _ = st.CreateAgent(context.Background(), claudeAgent) + claudeProfile := &models.AgentProfile{ + AgentID: claudeAgent.ID, + Name: "Claude", + Model: "claude-sonnet", + } + _ = st.CreateAgentProfile(context.Background(), claudeProfile) + codexAgent := &models.Agent{Name: "codex-acp"} + _ = st.CreateAgent(context.Background(), codexAgent) + codexProfile := &models.AgentProfile{ + AgentID: codexAgent.ID, + Name: "Codex", + Model: "gpt-5", + } + _ = st.CreateAgentProfile(context.Background(), codexProfile) + + log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json"}) + if err != nil { + t.Fatalf("logger: %v", err) + } + reg := registry.NewRegistry(log) + reg.MarkLoaded() + r := NewProfileReconciler(&fakeCapReader{caps: map[string]hostutility.AgentCapabilities{}}, reg, st, log) + if err := r.Run(context.Background()); err != nil { + t.Fatalf("Run: %v", err) + } + if len(st.softDeleted) != 0 { + t.Fatalf("expected empty registry cleanup to fail closed, got soft-deletes %v", st.softDeleted) + } + if len(st.profiles[claudeAgent.ID]) != 1 || len(st.profiles[codexAgent.ID]) != 1 { + t.Fatalf("expected live profiles to remain, got claude=%d codex=%d", + len(st.profiles[claudeAgent.ID]), len(st.profiles[codexAgent.ID])) + } +} + +func TestProfileReconciler_SkipsOrphanCleanupWhenAgentListFails(t *testing.T) { + st := newFakeStore() + st.listAgentsErr = errors.New("database locked") + orphanAgent := &models.Agent{Name: "removed-old-agent"} + _ = st.CreateAgent(context.Background(), orphanAgent) + orphanProfile := &models.AgentProfile{ + AgentID: orphanAgent.ID, + Name: "legacy", + Model: "x", + } + _ = st.CreateAgentProfile(context.Background(), orphanProfile) + + log, logs := newObserverLogger(t) + reg := registry.NewRegistry(log) + reg.LoadDefaults() + r := NewProfileReconciler(&fakeCapReader{caps: map[string]hostutility.AgentCapabilities{}}, reg, st, log) + if err := r.Run(context.Background()); err != nil { + t.Fatalf("Run: %v", err) + } + if len(st.softDeleted) != 0 { + t.Fatalf("expected agent-list failure to fail closed, got soft-deletes %v", st.softDeleted) + } + entries := logs.FilterMessage("orphan cleanup summary").All() + if len(entries) != 1 { + t.Fatalf("expected 1 summary log, got %d", len(entries)) + } + fields := entries[0].ContextMap() + if fields["skip_reason"] != "list_agents_failed" { + t.Errorf("skip_reason = %v, want list_agents_failed", fields["skip_reason"]) + } +} + +func TestProfileReconciler_SkipsOrphanCleanupWhenBatchExceedsSafetyLimit(t *testing.T) { + st := newFakeStore() + for i := 1; i <= 21; i++ { + orphanAgent := &models.Agent{Name: "removed-old-agent-" + strconv.Itoa(i)} + _ = st.CreateAgent(context.Background(), orphanAgent) + orphanProfile := &models.AgentProfile{ + AgentID: orphanAgent.ID, + Name: "legacy", + Model: "x", + } + _ = st.CreateAgentProfile(context.Background(), orphanProfile) + } + + log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json"}) + if err != nil { + t.Fatalf("logger: %v", err) + } + reg := registry.NewRegistry(log) + reg.LoadDefaults() + r := NewProfileReconciler(&fakeCapReader{caps: map[string]hostutility.AgentCapabilities{}}, reg, st, log) + if err := r.Run(context.Background()); err != nil { + t.Fatalf("Run: %v", err) + } + if len(st.softDeleted) != 0 { + t.Fatalf("expected oversized cleanup batch to fail closed, got soft-deletes %v", st.softDeleted) + } +} + +func TestProfileReconciler_SkipsOrphanCleanupWhenProfileCountExceedsSafetyLimit(t *testing.T) { + st := newFakeStore() + orphanAgent := &models.Agent{Name: "removed-old-agent"} + _ = st.CreateAgent(context.Background(), orphanAgent) + for i := 1; i <= 51; i++ { + orphanProfile := &models.AgentProfile{ + AgentID: orphanAgent.ID, + Name: "legacy-" + strconv.Itoa(i), + Model: "x", + } + _ = st.CreateAgentProfile(context.Background(), orphanProfile) + } + + log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json"}) + if err != nil { + t.Fatalf("logger: %v", err) + } + reg := registry.NewRegistry(log) + reg.LoadDefaults() + r := NewProfileReconciler(&fakeCapReader{caps: map[string]hostutility.AgentCapabilities{}}, reg, st, log) + if err := r.Run(context.Background()); err != nil { + t.Fatalf("Run: %v", err) + } + if len(st.softDeleted) != 0 { + t.Fatalf("expected oversized profile batch to fail closed, got soft-deletes %v", st.softDeleted) + } +} + +func TestProfileReconciler_SkipsOrphanCleanupWhenCandidateProfileListFails(t *testing.T) { + st := newFakeStore() + failingAgent := &models.Agent{Name: "removed-old-agent-failing"} + _ = st.CreateAgent(context.Background(), failingAgent) + st.listProfErr = map[string]error{ + failingAgent.ID: errors.New("database locked"), + } + deletableAgent := &models.Agent{Name: "removed-old-agent-deletable"} + _ = st.CreateAgent(context.Background(), deletableAgent) + deletableProfile := &models.AgentProfile{ + AgentID: deletableAgent.ID, + Name: "legacy", + Model: "x", + } + _ = st.CreateAgentProfile(context.Background(), deletableProfile) + + log, logs := newObserverLogger(t) + reg := registry.NewRegistry(log) + reg.LoadDefaults() + r := NewProfileReconciler(&fakeCapReader{caps: map[string]hostutility.AgentCapabilities{}}, reg, st, log) + if err := r.Run(context.Background()); err != nil { + t.Fatalf("Run: %v", err) + } + if len(st.softDeleted) != 0 { + t.Fatalf("expected profile-list failure to fail closed, got soft-deletes %v", st.softDeleted) + } + entries := logs.FilterMessage("orphan cleanup summary").All() + if len(entries) != 1 { + t.Fatalf("expected 1 summary log, got %d", len(entries)) + } + fields := entries[0].ContextMap() + if fields["profiles_candidate_partial"] != true { + t.Errorf("profiles_candidate_partial = %v, want true", fields["profiles_candidate_partial"]) + } + if fields["profiles_candidate_count"] != int64(0) { + t.Errorf("profiles_candidate_count = %v, want 0 for partial collection", fields["profiles_candidate_count"]) + } +} + +func TestProfileReconciler_LogsOrphanCleanupSummary(t *testing.T) { + st := newFakeStore() + orphanAgent := &models.Agent{Name: "removed-old-agent"} + _ = st.CreateAgent(context.Background(), orphanAgent) + orphanProfile := &models.AgentProfile{ + AgentID: orphanAgent.ID, + Name: "legacy", + Model: "x", + } + _ = st.CreateAgentProfile(context.Background(), orphanProfile) + + log, logs := newObserverLogger(t) + reg := registry.NewRegistry(log) + reg.LoadDefaults() + r := NewProfileReconciler(&fakeCapReader{caps: map[string]hostutility.AgentCapabilities{}}, reg, st, log) + if err := r.Run(context.Background()); err != nil { + t.Fatalf("Run: %v", err) + } + entries := logs.FilterMessage("orphan cleanup summary").All() + if len(entries) != 1 { + t.Fatalf("expected 1 summary log, got %d", len(entries)) + } + fields := entries[0].ContextMap() + if fields["db_agent_count"] != int64(1) { + t.Errorf("db_agent_count = %v, want 1", fields["db_agent_count"]) + } + if fields["orphan_agent_count"] != int64(1) { + t.Errorf("orphan_agent_count = %v, want 1", fields["orphan_agent_count"]) + } + if fields["profiles_deleted_count"] != int64(1) { + t.Errorf("profiles_deleted_count = %v, want 1", fields["profiles_deleted_count"]) + } + if fields["profiles_candidate_partial"] != false { + t.Errorf("profiles_candidate_partial = %v, want false", fields["profiles_candidate_partial"]) + } + if fields["skipped"] != false { + t.Errorf("skipped = %v, want false", fields["skipped"]) + } +} + // newDiscoveryController builds a Controller with just the dependencies that // syncAgentFromDiscovery touches: the agent registry, the store, and a logger. func newDiscoveryController(t *testing.T, st *fakeStore, ag agents.Agent) *Controller { diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_prompt.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_prompt.go index acbd7bb1d1..99db86dec8 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_prompt.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_prompt.go @@ -2,6 +2,7 @@ package acp import ( "context" + "errors" "fmt" "github.com/coder/acp-go-sdk" @@ -137,7 +138,7 @@ func (a *Adapter) sendPrompt(ctx context.Context, message string, attachments [] promptSpan.End() if err != nil { - return err + return normalizePromptErrorAfterCancel(promptCtx, err) } // Drain queued ACP notifications before running the post-prompt sweeps and @@ -205,6 +206,16 @@ func (a *Adapter) sendPrompt(ctx context.Context, message string, attachments [] return nil } +func normalizePromptErrorAfterCancel(promptCtx context.Context, err error) error { + if err == nil { + return nil + } + if errors.Is(context.Cause(promptCtx), ErrTurnCancelNotAcknowledged) { + return errPromptAbandonedAfterCancel + } + return err +} + func (a *Adapter) buildPromptContentBlocks(message string, attachments []v1.MessageAttachment) []acp.ContentBlock { contentBlocks := []acp.ContentBlock{acp.TextBlock(message)} diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_prompt_cancel_test.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_prompt_cancel_test.go index e1240cbb92..0ffd386363 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_prompt_cancel_test.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_prompt_cancel_test.go @@ -3,6 +3,7 @@ package acp import ( "context" "errors" + "fmt" "testing" "testing/synctest" "time" @@ -49,6 +50,21 @@ func TestWaitForPromptRPCAfterUserCancel_AbortReleasesWhenRPCStuck(t *testing.T) } } +func TestIsPromptAbandonedAfterCancel(t *testing.T) { + if !IsPromptAbandonedAfterCancel(errPromptAbandonedAfterCancel) { + t.Fatal("expected direct sentinel to match") + } + if !IsPromptAbandonedAfterCancel(fmt.Errorf("wrapped: %w", errPromptAbandonedAfterCancel)) { + t.Fatal("expected wrapped sentinel to match") + } + if !IsPromptAbandonedAfterCancel(errors.New("prompt failed: prompt abandoned after cancel")) { + t.Fatal("expected string-wrapped sentinel to match") + } + if IsPromptAbandonedAfterCancel(errors.New("different prompt failure")) { + t.Fatal("unexpected match for unrelated error") + } +} + func TestWaitForPromptRPCAfterUserCancel_CompletesAfterAbort(t *testing.T) { // synctest lets us deterministically exercise the abort path: close abortCh // first, run waitForPromptRPCAfterUserCancel in a goroutine, synctest.Wait @@ -162,3 +178,13 @@ func TestWaitForPromptRPCAfterCancel_CancelsPromptCtxOnTimeout(t *testing.T) { context.Cause(ctx)) } } + +func TestNormalizePromptErrorAfterCancel_MapsTimeoutCanceledRPCToAbandonedPrompt(t *testing.T) { + ctx, cancel := context.WithCancelCause(context.Background()) + cancel(ErrTurnCancelNotAcknowledged) + + err := normalizePromptErrorAfterCancel(ctx, context.Canceled) + if !errors.Is(err, errPromptAbandonedAfterCancel) { + t.Fatalf("expected errPromptAbandonedAfterCancel, got %v", err) + } +} diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/errors.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/errors.go index a5409d8a12..a374cbd238 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/acp/errors.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/errors.go @@ -1,6 +1,9 @@ package acp -import "errors" +import ( + "errors" + "strings" +) // ErrTurnCancelNotAcknowledged means session/cancel (and local prompt interruption) // were sent but the in-flight session/prompt RPC did not finish within the join @@ -11,3 +14,14 @@ var ErrTurnCancelNotAcknowledged = errors.New("turn cancel not acknowledged") // requested but the session/prompt RPC did not end in time. The prompt gate is // released so a follow-up prompt can be dispatched. var errPromptAbandonedAfterCancel = errors.New("prompt abandoned after cancel") + +// IsPromptAbandonedAfterCancel reports whether err is the prompt-gate release +// sentinel produced after a user cancel. This is not an agent failure; the +// lifecycle cancel path has already reconciled the interrupted turn. +func IsPromptAbandonedAfterCancel(err error) bool { + if err == nil { + return false + } + return errors.Is(err, errPromptAbandonedAfterCancel) || + strings.Contains(err.Error(), errPromptAbandonedAfterCancel.Error()) +} diff --git a/apps/backend/internal/agentctl/server/api/agent.go b/apps/backend/internal/agentctl/server/api/agent.go index b6782b4061..f2c55a8a31 100644 --- a/apps/backend/internal/agentctl/server/api/agent.go +++ b/apps/backend/internal/agentctl/server/api/agent.go @@ -467,6 +467,11 @@ func (s *Server) handleWSPrompt(ctx context.Context, msg *ws.Message) *ws.Messag // the user cancels, or agentctl shuts down. go func() { if err := adapter.Prompt(context.Background(), req.Text, req.Attachments); err != nil { + if acptransport.IsPromptAbandonedAfterCancel(err) { + s.logger.Info("async prompt abandoned after cancel; suppressing stale error event", + zap.Error(err)) + return + } s.logger.Error("async prompt failed", zap.Error(err)) s.procMgr.SendErrorEvent(err.Error()) } diff --git a/apps/backend/internal/agentctl/server/api/agent_test.go b/apps/backend/internal/agentctl/server/api/agent_test.go index 5707dc05c3..ed632fa5f9 100644 --- a/apps/backend/internal/agentctl/server/api/agent_test.go +++ b/apps/backend/internal/agentctl/server/api/agent_test.go @@ -3,16 +3,20 @@ package api import ( "context" "encoding/json" + "errors" + "io" "net/http/httptest" "strings" "testing" "time" "github.com/gorilla/websocket" + "github.com/kandev/kandev/internal/agentctl/server/adapter" "github.com/kandev/kandev/internal/agentctl/server/config" "github.com/kandev/kandev/internal/agentctl/server/process" "github.com/kandev/kandev/internal/agentctl/types" "github.com/kandev/kandev/internal/common/logger" + v1 "github.com/kandev/kandev/pkg/api/v1" ws "github.com/kandev/kandev/pkg/websocket" ) @@ -248,6 +252,44 @@ func TestHandleWSPrompt_NoAdapter(t *testing.T) { } } +func TestHandleWSPrompt_SuppressesPromptAbandonedAfterCancel(t *testing.T) { + s := newTestServer(t) + prompted := make(chan struct{}, 1) + s.procMgr.SetAdapterForTest(&promptErrorAdapter{ + sessionID: "session-123", + err: errors.New("prompt failed: prompt abandoned after cancel"), + prompted: prompted, + }) + + msg, _ := ws.NewRequest("req-1", "agent.prompt", map[string]string{ + "text": "hello", + }) + resp := s.handleWSPrompt(context.Background(), msg) + + if resp.Type != ws.MessageTypeResponse { + t.Fatalf("expected response type, got %q", resp.Type) + } + var result PromptResponse + if err := resp.ParsePayload(&result); err != nil { + t.Fatalf("failed to parse response: %v", err) + } + if !result.Success { + t.Fatal("expected prompt response success") + } + + select { + case <-prompted: + case <-time.After(2 * time.Second): + t.Fatal("prompt was not called") + } + + select { + case event := <-s.procMgr.GetUpdates(): + t.Fatalf("expected no error event for prompt abandoned after cancel, got %+v", event) + case <-time.After(100 * time.Millisecond): + } +} + func TestHandleWSCancel_NoAdapter(t *testing.T) { s := newTestServer(t) ctx := context.Background() @@ -267,6 +309,73 @@ func TestHandleWSCancel_NoAdapter(t *testing.T) { } } +type promptErrorAdapter struct { + sessionID string + err error + prompted chan<- struct{} +} + +func (a *promptErrorAdapter) PrepareEnvironment() (map[string]string, error) { + return nil, nil +} + +func (a *promptErrorAdapter) PrepareCommandArgs() []string { + return nil +} + +func (a *promptErrorAdapter) Connect(_ io.Writer, _ io.Reader) error { + return nil +} + +func (a *promptErrorAdapter) Initialize(_ context.Context) error { + return nil +} + +func (a *promptErrorAdapter) GetAgentInfo() *adapter.AgentInfo { + return nil +} + +func (a *promptErrorAdapter) NewSession(_ context.Context, _ []types.McpServer) (string, error) { + return a.sessionID, nil +} + +func (a *promptErrorAdapter) LoadSession(_ context.Context, sessionID string, _ []types.McpServer) error { + a.sessionID = sessionID + return nil +} + +func (a *promptErrorAdapter) Prompt(_ context.Context, _ string, _ []v1.MessageAttachment) error { + a.prompted <- struct{}{} + return a.err +} + +func (a *promptErrorAdapter) Cancel(_ context.Context) error { + return nil +} + +func (a *promptErrorAdapter) Updates() <-chan adapter.AgentEvent { + return nil +} + +func (a *promptErrorAdapter) GetSessionID() string { + return a.sessionID +} + +func (a *promptErrorAdapter) GetOperationID() string { + return "" +} + +func (a *promptErrorAdapter) SetPermissionHandler(_ adapter.PermissionHandler) { +} + +func (a *promptErrorAdapter) Close() error { + return nil +} + +func (a *promptErrorAdapter) RequiresProcessKill() bool { + return false +} + func TestHandleWSStderr_Empty(t *testing.T) { s := newTestServer(t) ctx := context.Background() diff --git a/apps/backend/internal/agentctl/server/process/testing_helpers.go b/apps/backend/internal/agentctl/server/process/testing_helpers.go index f23b82833c..fb0e2f3b4c 100644 --- a/apps/backend/internal/agentctl/server/process/testing_helpers.go +++ b/apps/backend/internal/agentctl/server/process/testing_helpers.go @@ -1,6 +1,18 @@ package process -import "time" +import ( + "time" + + "github.com/kandev/kandev/internal/agentctl/server/adapter" +) + +// SetAdapterForTest injects an adapter without starting a real agent process. +// Intended for API handler tests that need to exercise adapter interactions. +func (m *Manager) SetAdapterForTest(a adapter.AgentAdapter) { + m.mu.Lock() + defer m.mu.Unlock() + m.adapter = a +} // SetVscodeForTest injects a VscodeManager with the given status and port. // Intended for use in tests where starting a real code-server is not feasible. diff --git a/apps/backend/internal/orchestrator/event_handlers_agent.go b/apps/backend/internal/orchestrator/event_handlers_agent.go index 8e878d81ec..a190f8c517 100644 --- a/apps/backend/internal/orchestrator/event_handlers_agent.go +++ b/apps/backend/internal/orchestrator/event_handlers_agent.go @@ -8,6 +8,7 @@ import ( "go.uber.org/zap" + "github.com/kandev/kandev/internal/agent/runtime/lifecycle" "github.com/kandev/kandev/internal/agent/runtime/routingerr" "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/events/bus" @@ -456,7 +457,8 @@ func (s *Service) executeQueuedMessage(callerSessionID string, queuedMsg *messag zap.String("queue_id", queuedMsg.ID), zap.Error(err)) - if isSessionBusyError(err) || isTransientPromptError(err) || isSessionResetInProgressError(err) { + if isSessionBusyError(err) || isTransientPromptError(err) || + errors.Is(err, lifecycle.ErrCancelEscalated) || isSessionResetInProgressError(err) { s.logger.Warn("queued message execution failed transiently; requeueing", zap.String("session_id", callerSessionID), zap.String("task_id", queuedMsg.TaskID), diff --git a/apps/backend/internal/orchestrator/event_handlers_queue_test.go b/apps/backend/internal/orchestrator/event_handlers_queue_test.go index c8c37b349b..212b4e7beb 100644 --- a/apps/backend/internal/orchestrator/event_handlers_queue_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_queue_test.go @@ -2,10 +2,12 @@ package orchestrator import ( "context" + "fmt" "reflect" "strings" "testing" + "github.com/kandev/kandev/internal/agent/runtime/lifecycle" "github.com/kandev/kandev/internal/github" "github.com/kandev/kandev/internal/orchestrator/executor" "github.com/kandev/kandev/internal/orchestrator/messagequeue" @@ -53,6 +55,49 @@ func TestExecuteQueuedMessage_RequeuesWhenResetInProgress(t *testing.T) { } } +func TestExecuteQueuedMessage_RequeuesCancelReleaseFailure(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + + session, err := repo.GetTaskSession(ctx, "s1") + if err != nil { + t.Fatalf("failed to get session: %v", err) + } + session.State = models.TaskSessionStateWaitingForInput + session.AgentExecutionID = "exec-1" + seedExecutorRunning(t, repo, session.ID, session.TaskID, "exec-1") + if err := repo.UpdateTaskSession(ctx, session); err != nil { + t.Fatalf("failed to update session: %v", err) + } + + taskRepo := newMockTaskRepo() + agentMgr := &mockAgentManager{ + isAgentRunning: true, + promptErr: fmt.Errorf("failed to trigger prompt: prompt abandoned after cancel: %w", lifecycle.ErrCancelEscalated), + } + svc := createTestServiceWithAgent(repo, newMockStepGetter(), taskRepo, agentMgr) + svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{}) + + queuedMsg := &messagequeue.QueuedMessage{ + ID: "q-cancel", + SessionID: "s1", + TaskID: "t1", + Content: "hello after cancel", + QueuedBy: "test", + } + + svc.executeQueuedMessage("s1", queuedMsg) + + status := svc.messageQueue.GetStatus(ctx, "s1") + if status.Count != 1 { + t.Fatalf("expected queued message to be requeued after cancel-release failure, count=%d", status.Count) + } + if status.Entries[0].Content != "hello after cancel" { + t.Fatalf("expected queued content to be preserved, got %q", status.Entries[0].Content) + } +} + // TestExecuteQueuedMessage_SkipsUserMessageWhenAlreadyRecorded pins the // duplicate-prompt fix: when a queued workflow auto-start carries // metadata[user_message_recorded]=true (set by autoStartStepPrompt's diff --git a/apps/backend/internal/orchestrator/event_handlers_test.go b/apps/backend/internal/orchestrator/event_handlers_test.go index 54566a6be8..ebafc4f64a 100644 --- a/apps/backend/internal/orchestrator/event_handlers_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_test.go @@ -165,6 +165,7 @@ type mockAgentManager struct { restartProcessErr error promptErr error promptResult *executor.PromptResult + promptAgentFunc func(context.Context, string, string, []v1.MessageAttachment, bool) (*executor.PromptResult, error) launchAgentFunc func(context.Context, *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) mu sync.Mutex @@ -281,11 +282,12 @@ func (m *mockAgentManager) StopAgentWithReason(ctx context.Context, agentExecuti } return err } -func (m *mockAgentManager) PromptAgent(_ context.Context, executionID string, prompt string, _ []v1.MessageAttachment, dispatchOnly bool) (*executor.PromptResult, error) { +func (m *mockAgentManager) PromptAgent(ctx context.Context, executionID string, prompt string, attachments []v1.MessageAttachment, dispatchOnly bool) (*executor.PromptResult, error) { m.mu.Lock() first := len(m.capturedPrompts) == 0 m.capturedPrompts = append(m.capturedPrompts, prompt) m.capturedPromptCalls = append(m.capturedPromptCalls, promptCall{ExecutionID: executionID, Prompt: prompt, DispatchOnly: dispatchOnly}) + promptAgentFunc := m.promptAgentFunc promptErr := m.promptErr promptResult := m.promptResult doneCh := m.promptDone @@ -293,6 +295,9 @@ func (m *mockAgentManager) PromptAgent(_ context.Context, executionID string, pr if first && doneCh != nil { close(doneCh) } + if promptAgentFunc != nil { + return promptAgentFunc(ctx, executionID, prompt, attachments, dispatchOnly) + } if promptErr != nil { return nil, promptErr } diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index 0f546d8f95..6bce7db59c 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -2635,6 +2635,11 @@ func (s *Service) CancelAgent(ctx context.Context, sessionID string) error { // concurrent agent.complete event having already closed the turn. s.completeTurnForSession(ctx, sessionID) + // Clear the duplicate-cancel guard early so drainQueuedMessageForPromptableSession + // can dispatch the queued prompt without being blocked by the in-flight sentinel. + // The defer at function entry remains as the error-path safety net. + s.cancelInFlight.Delete(sessionID) + // Deliver any message the operator queued while the turn was running // (#1597 pause→resume recovery). On a normal cancel the agent emits // complete(cancelled) → handleAgentReady → on_turn_complete, which drains diff --git a/apps/backend/internal/orchestrator/task_operations_test.go b/apps/backend/internal/orchestrator/task_operations_test.go index 2c2b14f88b..f02e6ec032 100644 --- a/apps/backend/internal/orchestrator/task_operations_test.go +++ b/apps/backend/internal/orchestrator/task_operations_test.go @@ -654,6 +654,48 @@ func TestCancelAgent_DeliversQueuedMessageOnResume(t *testing.T) { } } +func TestCancelAgent_DrainsQueuedMessageAfterCancelGuardClears(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + promptDone := make(chan struct{}) + agentMgr := &mockAgentManager{ + isAgentRunning: true, + repoForExecutionLookup: repo, + promptDone: promptDone, + } + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr) + svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{}) + agentMgr.promptAgentFunc = func(context.Context, string, string, []v1.MessageAttachment, bool) (*executor.PromptResult, error) { + if svc.isCancelInFlight("session1") { + return nil, fmt.Errorf("prompt abandoned after cancel: %w", lifecycle.ErrCancelEscalated) + } + return &executor.PromptResult{}, nil + } + + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateRunning) + seedExecutorRunning(t, repo, "session1", "task1", "exec-1") + if _, err := svc.messageQueue.QueueMessage( + ctx, "session1", "task1", "queued after cancel", "", messagequeue.QueuedByUser, false, nil, + ); err != nil { + t.Fatalf("queue message: %v", err) + } + + if err := svc.CancelAgent(ctx, "session1"); err != nil { + t.Fatalf("cancel agent: %v", err) + } + + select { + case <-promptDone: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for queued prompt dispatch") + } + + status := svc.messageQueue.GetStatus(ctx, "session1") + if status.Count != 0 { + t.Fatalf("expected queued prompt to stay drained after cancel guard clears, count=%d entries=%+v", status.Count, status.Entries) + } +} + // --- StartCreatedSession --- func TestStartCreatedSession_WrongTask(t *testing.T) {