diff --git a/apps/backend/internal/agent/hostutility/instance_notfound_test.go b/apps/backend/internal/agent/hostutility/instance_notfound_test.go deleted file mode 100644 index cec7e3e8ce..0000000000 --- a/apps/backend/internal/agent/hostutility/instance_notfound_test.go +++ /dev/null @@ -1,50 +0,0 @@ -package hostutility - -import ( - "errors" - "fmt" - "testing" -) - -func TestIsInstanceNotFound(t *testing.T) { - tests := []struct { - name string - err error - want bool - }{ - {name: "nil", err: nil, want: false}, - { - // The exact shape ControlClient.DeleteInstance produces on shutdown. - name: "delete instance 404", - err: fmt.Errorf("failed to delete instance: instance not found (status 404)"), - want: true, - }, - { - name: "not found phrase without status", - err: errors.New("instance \"inst-3\" not found"), - want: true, - }, - { - name: "status 404 uppercased", - err: errors.New("DELETE returned STATUS 404"), - want: true, - }, - { - name: "transport failure is not not-found", - err: errors.New("failed to delete instance: connection refused"), - want: false, - }, - { - name: "server error is not not-found", - err: errors.New("failed to delete instance: internal error (status 500)"), - want: false, - }, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - if got := isInstanceNotFound(tt.err); got != tt.want { - t.Fatalf("isInstanceNotFound(%v) = %v, want %v", tt.err, got, tt.want) - } - }) - } -} diff --git a/apps/backend/internal/agent/hostutility/manager.go b/apps/backend/internal/agent/hostutility/manager.go index 05d3c2c1fe..1680646c6f 100644 --- a/apps/backend/internal/agent/hostutility/manager.go +++ b/apps/backend/internal/agent/hostutility/manager.go @@ -249,20 +249,10 @@ func (m *Manager) deleteInstance(ctx context.Context, inst *instance) { } if m.controlClient != nil { if err := m.controlClient.DeleteInstance(ctx, inst.instanceID); err != nil { - // During shutdown agentctl may already be gone, so the delete - // returns a not-found/404. That is benign and idempotent: log DEBUG - // so it does not add teardown noise. Any other failure stays WARN. - if isInstanceNotFound(err) { - m.log.Debug("host utility instance already deleted", - zap.String("agent_type", inst.agentType), - zap.String("instance_id", inst.instanceID), - zap.String("error", err.Error())) - } else { - m.log.Warn("failed to delete host utility instance", - zap.String("agent_type", inst.agentType), - zap.String("instance_id", inst.instanceID), - zap.Error(err)) - } + m.log.Warn("failed to delete host utility instance", + zap.String("agent_type", inst.agentType), + zap.String("instance_id", inst.instanceID), + zap.Error(err)) } } if inst.workDir == "" { @@ -276,19 +266,6 @@ func (m *Manager) deleteInstance(ctx context.Context, inst *instance) { } } -// isInstanceNotFound reports whether a DeleteInstance error means the agentctl -// instance was already gone. ControlClient.DeleteInstance stringifies the -// upstream error as "failed to delete instance: (status 404)" without a -// wrapped sentinel, so this matches on the 404 status and the not-found phrase -// rather than errors.Is. -func isInstanceNotFound(err error) bool { - if err == nil { - return false - } - msg := strings.ToLower(err.Error()) - return strings.Contains(msg, "status 404") || strings.Contains(msg, "not found") -} - // eligibleAgents returns enabled agents that implement InferenceAgent AND whose // runtime protocol is ACP. This is the v1 scope. func (m *Manager) eligibleAgents() []agents.InferenceAgent { diff --git a/apps/backend/internal/agent/runtime/agentctl/control.go b/apps/backend/internal/agent/runtime/agentctl/control.go index 4e8e08b480..73b7e7088b 100644 --- a/apps/backend/internal/agent/runtime/agentctl/control.go +++ b/apps/backend/internal/agent/runtime/agentctl/control.go @@ -301,7 +301,11 @@ func (c *ControlClient) DeleteInstance(ctx context.Context, instanceID string) e _ = resp.Body.Close() }() - if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNoContent { + // A retry may observe 404 after agentctl completed the first delete but its + // response was lost. The desired postcondition is already satisfied. + if resp.StatusCode != http.StatusOK && + resp.StatusCode != http.StatusNoContent && + resp.StatusCode != http.StatusNotFound { var errResp struct { Error string `json:"error"` } diff --git a/apps/backend/internal/agent/runtime/agentctl/control_test.go b/apps/backend/internal/agent/runtime/agentctl/control_test.go index 3c4ba785ef..fba38b1a24 100644 --- a/apps/backend/internal/agent/runtime/agentctl/control_test.go +++ b/apps/backend/internal/agent/runtime/agentctl/control_test.go @@ -3,7 +3,9 @@ package client import ( "context" "encoding/json" + "errors" "fmt" + "io" "net" "net/http" "net/http/httptest" @@ -460,6 +462,39 @@ func TestDeleteInstance_UsesDeleteVerbWithIDInPath(t *testing.T) { } } +func TestDeleteInstance_LostResponseThenNotFoundIsSuccessfulRetry(t *testing.T) { + transport := &lostDeleteResponseTransport{} + client := NewControlClient("agentctl.test", 80, newTestLogger()) + client.httpClient = &http.Client{Transport: transport} + + if err := client.DeleteInstance(context.Background(), "inst-9"); err == nil { + t.Fatal("first DeleteInstance error = nil, want lost-response error") + } + if err := client.DeleteInstance(context.Background(), "inst-9"); err != nil { + t.Fatalf("DeleteInstance retry after completed delete: %v", err) + } + if transport.requests != 2 { + t.Fatalf("requests = %d, want 2", transport.requests) + } +} + +type lostDeleteResponseTransport struct { + requests int +} + +func (t *lostDeleteResponseTransport) RoundTrip(req *http.Request) (*http.Response, error) { + t.requests++ + if t.requests == 1 { + return nil, errors.New("delete response lost") + } + return &http.Response{ + StatusCode: http.StatusNotFound, + Body: io.NopCloser(strings.NewReader(`{"error":"instance not found"}`)), + Header: make(http.Header), + Request: req, + }, nil +} + func TestDeleteInstance_FailureModes(t *testing.T) { tests := []struct { name string @@ -469,9 +504,9 @@ func TestDeleteInstance_FailureModes(t *testing.T) { }{ { name: "structured error", - status: http.StatusNotFound, - body: `{"error":"no such instance"}`, - wantErr: "failed to delete instance: no such instance (status 404)", + status: http.StatusConflict, + body: `{"error":"instance still stopping"}`, + wantErr: "failed to delete instance: instance still stopping (status 409)", }, { name: "unparseable error body", diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go b/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go index 384552d182..7357447475 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go @@ -880,7 +880,10 @@ func (m *Manager) StopAgentWithReason(ctx context.Context, executionID string, r return err } defer activityLease.Release() - m.releaseActivity(executionActivityKey(executionID)) + // Keep the execution's running activity lease until backend teardown + // succeeds. A failed stop remains retryable, so maintenance must not treat + // a potentially live runtime as idle. RemoveExecution releases the lease on + // the successful path. m.logger.Info("stopping agent", zap.String("execution_id", executionID), @@ -910,8 +913,12 @@ func (m *Manager) StopAgentWithReason(ctx context.Context, executionID string, r execution.agentctl.Close() } - // Stop the agent execution via the runtime that created it - _ = m.stopAgentViaBackend(ctx, executionID, execution, reason, force, agentStopFailed) + // Stop the agent execution via the runtime that created it. A failed stop + // must remain tracked: removing it here would turn a retryable cleanup into + // an unobservable orphan process. + if err := m.stopAgentViaBackend(ctx, executionID, execution, reason, force, agentStopFailed); err != nil { + return fmt.Errorf("stop runtime for execution %q: %w", executionID, err) + } // Update execution status and remove from tracking _ = m.executionStore.WithLock(executionID, func(exec *AgentExecution) { diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_interaction_accessors_test.go b/apps/backend/internal/agent/runtime/lifecycle/manager_interaction_accessors_test.go index 22ada0ba6a..16101ab670 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_interaction_accessors_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_interaction_accessors_test.go @@ -2,12 +2,14 @@ package lifecycle import ( "context" + "errors" "testing" "time" "github.com/stretchr/testify/require" "github.com/kandev/kandev/internal/agent/executor" + "github.com/kandev/kandev/internal/agent/runtime/activity" "github.com/kandev/kandev/internal/agentctl/types/streams" "github.com/kandev/kandev/internal/events" v1 "github.com/kandev/kandev/pkg/api/v1" @@ -320,6 +322,60 @@ func TestStopAgentForcePassesForceToBackend(t *testing.T) { require.Equal(t, StopReasonBackendShutdown, stopTracker.stopReason) } +func TestStopAgentWithReason_BackendFailureKeepsExecutionRetryable(t *testing.T) { + log := newTestRegistryLogger() + execRegistry := NewExecutorRegistry(log) + backend := &retryableStopBackend{ + MockExecutor: MockExecutor{name: executor.NameStandalone}, + stopErr: errors.New("runtime stop failed"), + } + execRegistry.Register(backend) + mgr := NewManager(newTestRegistry(), &MockEventBus{}, execRegistry, nil, nil, nil, ExecutorFallbackWarn, "", log) + cleanupManagerStopCh(t, mgr) + coordinator := activity.NewCoordinator(activity.Options{}) + mgr.SetActivityCoordinator(coordinator) + runningLease, err := coordinator.AcquireTask(context.Background(), activity.KindExecutionRunning) + require.NoError(t, err) + mgr.trackActivity(executionActivityKey("exec-retryable-stop"), runningLease) + + require.NoError(t, mgr.executionStore.Add(&AgentExecution{ + ID: "exec-retryable-stop", + TaskID: "task-retryable-stop", + SessionID: "session-retryable-stop", + RuntimeName: executor.NameStandalone, + Status: v1.AgentStatusRunning, + })) + + err = mgr.StopAgentWithReason(context.Background(), "exec-retryable-stop", "idle cleanup", false) + require.ErrorIs(t, err, backend.stopErr) + _, exists := mgr.executionStore.Get("exec-retryable-stop") + require.True(t, exists, "a failed runtime stop must retain the execution for retry") + + bus := mgr.eventBus.(*MockEventBus) + require.Empty(t, bus.PublishedEvents, "a failed runtime stop must not publish agent.stopped") + _, _, err = coordinator.TryAcquireMaintenance(context.Background(), 0) + require.ErrorIs(t, err, activity.ErrBusy, "a failed stop must retain execution activity") + + backend.stopErr = nil + require.NoError(t, mgr.StopAgentWithReason(context.Background(), "exec-retryable-stop", "idle cleanup retry", false)) + _, exists = mgr.executionStore.Get("exec-retryable-stop") + require.False(t, exists) + require.Len(t, bus.PublishedEvents, 1) + require.Equal(t, events.AgentStopped, bus.PublishedEvents[0].Type) + maintenance, _, err := coordinator.TryAcquireMaintenance(context.Background(), 0) + require.NoError(t, err) + maintenance.Release() +} + +type retryableStopBackend struct { + MockExecutor + stopErr error +} + +func (b *retryableStopBackend) StopInstance(context.Context, *ExecutorInstance, bool) error { + return b.stopErr +} + type forceRecordingStopTracker struct { mockStopTracker forced bool diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_stop_retry_agentctl_test.go b/apps/backend/internal/agent/runtime/lifecycle/manager_stop_retry_agentctl_test.go new file mode 100644 index 0000000000..a155ce9731 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_stop_retry_agentctl_test.go @@ -0,0 +1,67 @@ +package lifecycle + +import ( + "context" + "errors" + "net" + "net/http" + "net/http/httptest" + "net/url" + "strconv" + "sync/atomic" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/kandev/kandev/internal/agent/executor" + agentctl "github.com/kandev/kandev/internal/agent/runtime/agentctl" +) + +// This reviewer-requested contract test documents that Client.Close drains +// streams and idle connections without disabling later HTTP stop retries. +func TestStopAgentWithReason_BackendRetryReusesClosedAgentctlClient(t *testing.T) { + var stopRequests atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Path != "/api/v1/stop" { + http.NotFound(w, r) + return + } + stopRequests.Add(1) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"success":true}`)) + })) + t.Cleanup(server.Close) + + log := newTestRegistryLogger() + parsedURL, err := url.Parse(server.URL) + require.NoError(t, err) + host, portString, err := net.SplitHostPort(parsedURL.Host) + require.NoError(t, err) + port, err := strconv.Atoi(portString) + require.NoError(t, err) + + execRegistry := NewExecutorRegistry(log) + backend := &retryableStopBackend{ + MockExecutor: MockExecutor{name: executor.NameStandalone}, + stopErr: errors.New("runtime stop failed"), + } + execRegistry.Register(backend) + mgr := NewManager(newTestRegistry(), &MockEventBus{}, execRegistry, nil, nil, nil, ExecutorFallbackWarn, "", log) + cleanupManagerStopCh(t, mgr) + + require.NoError(t, mgr.executionStore.Add(&AgentExecution{ + ID: "exec-agentctl-retry", + SessionID: "session-agentctl-retry", + RuntimeName: executor.NameStandalone, + agentctl: agentctl.NewClient(host, port, log), + })) + + err = mgr.StopAgentWithReason(context.Background(), "exec-agentctl-retry", "first attempt", false) + require.ErrorIs(t, err, backend.stopErr) + + backend.stopErr = nil + require.NoError(t, mgr.StopAgentWithReason(context.Background(), "exec-agentctl-retry", "retry", false)) + require.Equal(t, int32(2), stopRequests.Load(), "the retry must reach agentctl after the first Close") + _, exists := mgr.executionStore.Get("exec-agentctl-retry") + require.False(t, exists) +} diff --git a/apps/backend/internal/orchestrator/dynamic_launch.go b/apps/backend/internal/orchestrator/dynamic_launch.go index 0a1f2de9e3..0625537121 100644 --- a/apps/backend/internal/orchestrator/dynamic_launch.go +++ b/apps/backend/internal/orchestrator/dynamic_launch.go @@ -8,6 +8,7 @@ import ( "fmt" "strings" + agentruntime "github.com/kandev/kandev/internal/agent/runtime" dynamicruntime "github.com/kandev/kandev/internal/agent/runtime/dynamic" "github.com/kandev/kandev/internal/agent/runtime/routingerr" "github.com/kandev/kandev/internal/orchestrator/executor" @@ -632,7 +633,12 @@ func (s *Service) relaunchDynamicTaskAfterFailure( return false } if err := s.executor.StopExecution(ctx, data.AgentExecutionID, "dynamic route fallback", true); err != nil { - s.logger.Debug("failed to stop dynamic fallback predecessor", zap.Error(err)) + if errors.Is(err, agentruntime.ErrNotFound) { + s.logger.Debug("dynamic fallback predecessor is already absent", zap.Error(err)) + } else { + s.logger.Debug("failed to stop dynamic fallback predecessor", zap.Error(err)) + return false + } } if err := s.repo.UpdateTaskSessionState(ctx, data.SessionID, models.TaskSessionStateCreated, ""); err != nil { return false diff --git a/apps/backend/internal/orchestrator/dynamic_launch_test.go b/apps/backend/internal/orchestrator/dynamic_launch_test.go new file mode 100644 index 0000000000..1f9d77962f --- /dev/null +++ b/apps/backend/internal/orchestrator/dynamic_launch_test.go @@ -0,0 +1,66 @@ +package orchestrator + +import ( + "context" + "errors" + "testing" + + "github.com/kandev/kandev/internal/orchestrator/watcher" + "github.com/kandev/kandev/internal/task/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +func TestRelaunchDynamicTaskAfterFailure_DoesNotLaunchSuccessorWhenStopFails(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-dynamic-stop-failure" + sessionID = "session-dynamic-stop-failure" + executionID = "execution-dynamic-stop-failure" + ) + + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateRunning) + seedExecutorRunning(t, repo, sessionID, taskID, executionID) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress) + stopErr := errors.New("runtime teardown failed") + agentManager := &mockAgentManager{ + stopAgentWithReasonErr: stopErr, + } + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentManager) + svc.lastTurnPrompt.Store(sessionID, capturedPrompt{text: "retry the task"}) + + relaunched := svc.relaunchDynamicTaskAfterFailure( + ctx, + watcher.AgentEventData{ + TaskID: taskID, + SessionID: sessionID, + AgentExecutionID: executionID, + }, + "fallback-profile", + ) + + if relaunched { + t.Fatal("relaunchDynamicTaskAfterFailure returned success after predecessor stop failed") + } + session, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("get session: %v", err) + } + if session.State != models.TaskSessionStateRunning { + t.Fatalf("session state = %q, want RUNNING while predecessor teardown is unresolved", session.State) + } + if len(agentManager.startAgentProcessCalls) != 0 { + t.Fatalf("successor launch started %d processes after stop failure", len(agentManager.startAgentProcessCalls)) + } + if len(agentManager.stopAgentWithReasonArgs) != 1 { + t.Fatalf("stop calls = %d, want 1", len(agentManager.stopAgentWithReasonArgs)) + } + if agentManager.stopAgentWithReasonArgs[0] != (stopAgentCall{ + ExecutionID: executionID, + Reason: "dynamic route fallback", + Force: true, + }) { + t.Fatalf("unexpected stop call: %#v", agentManager.stopAgentWithReasonArgs[0]) + } +} diff --git a/apps/backend/internal/orchestrator/executor/executor_interaction.go b/apps/backend/internal/orchestrator/executor/executor_interaction.go index b566538c70..62b657282d 100644 --- a/apps/backend/internal/orchestrator/executor/executor_interaction.go +++ b/apps/backend/internal/orchestrator/executor/executor_interaction.go @@ -649,9 +649,10 @@ func (e *Executor) prepareModelSwitch(ctx context.Context, taskID, sessionID str e.logger.Info("stopping current agent for model switch", zap.String("agent_execution_id", executionID)) if err := e.agentManager.StopAgent(ctx, executionID, false); err != nil { - e.logger.Warn("failed to stop agent for model switch, continuing anyway", + e.logger.Warn("failed to stop agent for model switch", zap.Error(err), zap.String("agent_execution_id", executionID)) + return nil, nil, "", nil, fmt.Errorf("failed to stop agent for model switch: %w", err) } return session, task, acpSessionID, existingRunning, nil diff --git a/apps/backend/internal/orchestrator/executor/executor_interaction_test.go b/apps/backend/internal/orchestrator/executor/executor_interaction_test.go index a234935d22..242369b5d9 100644 --- a/apps/backend/internal/orchestrator/executor/executor_interaction_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_interaction_test.go @@ -427,6 +427,35 @@ func TestStopExecution_PreservesRuntimeFailureClassification(t *testing.T) { } } +func TestPrepareModelSwitch_StopsBeforeReplacementWhenTeardownFails(t *testing.T) { + stopErr := errors.New("runtime teardown failed") + repo := newMockRepository() + repo.sessions["session-model-switch"] = &models.TaskSession{ + ID: "session-model-switch", TaskID: "task-model-switch", + } + repo.tasks["task-model-switch"] = &models.Task{ID: "task-model-switch"} + manager := &mockAgentManager{ + getExecutionIDForSessionFunc: func(context.Context, string) (string, error) { + return "execution-model-switch", nil + }, + stopAgentFunc: func(context.Context, string, bool) error { + return stopErr + }, + } + exec := newTestExecutor(t, manager, repo) + + session, task, acpSessionID, existingRunning, err := exec.prepareModelSwitch( + context.Background(), "task-model-switch", "session-model-switch", + ) + + if !errors.Is(err, stopErr) { + t.Fatalf("prepareModelSwitch error = %v, want %v", err, stopErr) + } + if session != nil || task != nil || acpSessionID != "" || existingRunning != nil { + t.Fatalf("prepareModelSwitch returned replacement inputs after stop failure: session=%#v task=%#v acp=%q running=%#v", session, task, acpSessionID, existingRunning) + } +} + func TestLaunchModelSwitchAgent_CleansStartedExecutionAfterTerminalRace(t *testing.T) { ctx := context.Background() repo := newMockRepository()