fix(agent-runtime): retain executions when runtime stop fails (#2998)

Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>

apps/backend/internal/agent/hostutility/instance_notfound_test.go

@@ -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)
-			}
-		})
-	}
-}

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: <msg> (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 {

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"`
 		}

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",

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) {

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

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)
+}

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

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])
+	}
+}

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

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()