fix(edge): separate attempt-cleanup from terminal finish on fault-escalation (#867)

edge-server/internal/lifecycle/process_executor.go

@@ -335,7 +335,15 @@ func (e *ProcessExecutor) Cancel(runID string) CancelResult {
 }
 
 func (e *ProcessExecutor) run(ctx context.Context, run store.Run, runCtx RunProcessContext) {
-	defer e.finish(run.ID)
+	// terminalFinish is true unless this attempt hands off to a fault-escalation
+	// successor. The deferred finish must not tear down the concurrency slot the
+	// successor just re-registered (see #867).
+	terminalFinish := true
+	defer func() {
+		if terminalFinish {
+			e.finish(run.ID)
+		}
+	}()
 
 	// Store Hub task ID for Edge→Hub direct callback reporting
 	if runCtx.HubTaskID != "" {
@@ -411,6 +419,7 @@ func (e *ProcessExecutor) run(ctx context.Context, run store.Run, runCtx RunProc
 	// generate a fresh random session ID and retry once. This handles the case where
 	// a stale CC process from a previous Edge instance still holds the session lock.
 	var lastWaitErr error
+	var lastOutStore *runnerctx.RunOutputStore
 	for attempt := 0; attempt < maxSessionRetries; attempt++ {
 		var cmdPath string
 		var args, env []string
@@ -600,6 +609,7 @@ func (e *ProcessExecutor) run(ctx context.Context, run store.Run, runCtx RunProc
 		// transient "file already closed" read errors.
 		wg.Wait()
 		lastWaitErr = cmd.Wait()
+		lastOutStore = outStore
 		slog.Debug("executor.subprocess.exited", "runId", run.ID, "exitCode", ExitCodeFromErr(lastWaitErr), "attempt", attempt)
 
 		// Context budget compaction check: after the stream completes, evaluate
@@ -660,11 +670,11 @@ func (e *ProcessExecutor) run(ctx context.Context, run store.Run, runCtx RunProc
 			continue
 		}
 
-		// Not retrying — process the final result.
+		// Wait error is not a session-conflict retry. Leave the session-retry
+		// loop so fault-escalation can re-launch the run when configured.
+		// Terminal publishFailed happens after the escalation check below.
 		if lastWaitErr != nil {
-			e.publishFailed(run, errorWithRunOutput(lastWaitErr, outStore))
-			e.sendSubAgentResult(run.ID, "failed", subAgentErrorPayload(lastWaitErr))
-			return
+			break
 		}
 		// #179: handle structured output parse errors with recoverability distinction.
 		// Non-recoverable errors (pipe broken, context cancelled) fail the run.
@@ -716,23 +726,32 @@ func (e *ProcessExecutor) run(ctx context.Context, run store.Run, runCtx RunProc
 		return
 	}
 
-	// Exhausted retries — attempt fault escalation if configured.
+	// Exhausted session retries — attempt fault escalation if configured.
 	if lastWaitErr != nil && e.faultEscalationCfg.Enabled && e.faultEscalationCfg.MaxRetries > 0 {
 		r, ok := e.store.GetRun(run.ID)
 		if ok && r.RetryCount < e.faultEscalationCfg.MaxRetries {
 			newCount := r.RetryCount + 1
 			e.store.SetRunRetryCount(run.ID, newCount)
 			run.RetryCount = newCount
-			if _, ok2 := e.store.SetRunStatusIf(run.ID, "queued", "failed"); ok2 {
+			// Re-queue from started (normal wait failure) or failed (defensive).
+			if _, ok2 := e.store.SetRunStatusIf(run.ID, "queued", "started", "failed"); ok2 {
 				run.Status = "queued"
 			}
-			// Clean up process tracking.
+			// Attempt-local cleanup only — leave concurrency slot / hub bookkeeping
+			// for the successor attempt. Terminal finish is owned by the successor.
 			e.mu.Lock()
 			delete(e.processes, run.ID)
 			if s, ok3 := e.runOutputs[run.ID]; ok3 {
 				_ = s.Close()
 				delete(e.runOutputs, run.ID)
 			}
+			// Re-register the cancel func before releasing the slot ownership so
+			// Cancel() and max-concurrent accounting stay consistent across handoff.
+			newCtx, cancel := context.WithTimeout(context.Background(), e.runTimeout)
+			if oldCancel, ok4 := e.running[run.ID]; ok4 {
+				oldCancel()
+			}
+			e.running[run.ID] = cancel
 			e.mu.Unlock()
 			e.bus.Publish("run.fault_escalation.retry", runScope(run),
 				faultEscalationRetryPayload(run.ID, newCount, e.faultEscalationCfg.MaxRetries))
@@ -741,14 +760,7 @@ func (e *ProcessExecutor) run(ctx context.Context, run store.Run, runCtx RunProc
 				"retryCount", newCount,
 				"maxRetries", e.faultEscalationCfg.MaxRetries,
 			)
-			// Create fresh context and retry from Start().
-			newCtx, cancel := context.WithTimeout(context.Background(), e.runTimeout)
-			e.mu.Lock()
-			if oldCancel, ok4 := e.running[run.ID]; ok4 {
-				oldCancel()
-			}
-			e.running[run.ID] = cancel
-			e.mu.Unlock()
+			terminalFinish = false
 			go e.run(newCtx, run, runCtx)
 			return
 		}
@@ -757,9 +769,9 @@ func (e *ProcessExecutor) run(ctx context.Context, run store.Run, runCtx RunProc
 			faultEscalationExhaustedPayload(run.ID, e.faultEscalationCfg.MaxRetries))
 		slog.Warn("process: fault escalation exhausted", "runId", run.ID)
 	}
-	// Report the last error.
+	// Report the last error (terminal failure after retries exhausted or disabled).
 	if lastWaitErr != nil {
-		e.publishFailed(run, errorWithRunOutput(lastWaitErr, nil))
+		e.publishFailed(run, errorWithRunOutput(lastWaitErr, lastOutStore))
 		e.sendSubAgentResult(run.ID, "failed", subAgentErrorPayload(lastWaitErr))
 	}
 }

edge-server/internal/lifecycle/process_executor_test.go

@@ -1467,6 +1467,164 @@ func TestProcessExecutorStartCancelRace(t *testing.T) {
 	wg.Wait()
 }
 
+// TestProcessExecutorFaultEscalationRetryKeepsRunRegistered verifies #867:
+// on fault-escalation auto-retry the concurrency slot remains registered for the
+// successor attempt (deferred finish must not tear it down between attempts).
+func TestProcessExecutorFaultEscalationRetryKeepsRunRegistered(t *testing.T) {
+	bus := events.NewBus(100)
+	s := store.New()
+	run := newExecutorTestRun(t, s)
+	_, ch, _ := bus.Subscribe(0)
+	executor := newTestProcessExecutor(t, bus, s, "fail")
+	executor.faultEscalationCfg = FaultEscalationConfig{
+		Enabled:    true,
+		MaxRetries: 1,
+	}
+
+	if err := executor.Start(run, RunProcessContext{}); err != nil {
+		t.Fatalf("Start returned error: %v", err)
+	}
+
+	var sawRetry bool
+	var sawFailed bool
+	deadline := time.After(15 * time.Second)
+	for !sawFailed {
+		select {
+		case evt := <-ch:
+			switch evt.Type {
+			case "run.started", "run.output.batch", "message.created", "item.created":
+			case "run.fault_escalation.retry":
+				sawRetry = true
+				// Immediately after the handoff event the run must still occupy
+				// the concurrency slot so Cancel / max-concurrent accounting work.
+				executor.mu.Lock()
+				_, registered := executor.running[run.ID]
+				executor.mu.Unlock()
+				if !registered {
+					t.Fatal("run not registered in executor.running after fault-escalation retry handoff")
+				}
+				stored, ok := s.GetRun(run.ID)
+				if !ok {
+					t.Fatal("run missing from store after retry handoff")
+				}
+				if stored.RetryCount != 1 {
+					t.Fatalf("RetryCount = %d, want 1 after first escalation retry", stored.RetryCount)
+				}
+				if stored.Status != "queued" && stored.Status != "started" {
+					t.Fatalf("status after retry handoff = %q, want queued or started", stored.Status)
+				}
+			case "run.fault_escalation.exhausted":
+			case "run.failed":
+				sawFailed = true
+			default:
+				t.Fatalf("unexpected event type %q", evt.Type)
+			}
+		case <-deadline:
+			t.Fatal("timed out waiting for terminal run.failed after fault escalation")
+		}
+	}
+	if !sawRetry {
+		t.Fatal("expected run.fault_escalation.retry before terminal failure")
+	}
+
+	// Terminal finish must clear the slot exactly once.
+	deadline2 := time.After(2 * time.Second)
+	for {
+		executor.mu.Lock()
+		_, registered := executor.running[run.ID]
+		executor.mu.Unlock()
+		if !registered {
+			break
+		}
+		select {
+		case <-deadline2:
+			t.Fatal("run still registered after terminal failure finish")
+		case <-time.After(10 * time.Millisecond):
+		}
+	}
+
+	// finish is idempotent: a second call must not panic or re-introduce state.
+	executor.finish(run.ID)
+	executor.mu.Lock()
+	_, registered := executor.running[run.ID]
+	executor.mu.Unlock()
+	if registered {
+		t.Fatal("run reappeared in running map after idempotent finish")
+	}
+}
+
+// TestProcessExecutorFaultEscalationExhaustedSingleFinish verifies that when
+// retries are exhausted (or disabled), terminal finish runs once and the slot
+// is released.
+func TestProcessExecutorFaultEscalationExhaustedSingleFinish(t *testing.T) {
+	bus := events.NewBus(100)
+	s := store.New()
+	run := newExecutorTestRun(t, s)
+	_, ch, _ := bus.Subscribe(0)
+	executor := newTestProcessExecutor(t, bus, s, "fail")
+	// MaxRetries=0: escalation enabled but no auto-retry budget → single finish.
+	executor.faultEscalationCfg = FaultEscalationConfig{
+		Enabled:    true,
+		MaxRetries: 0,
+	}
+
+	if err := executor.Start(run, RunProcessContext{}); err != nil {
+		t.Fatalf("Start returned error: %v", err)
+	}
+
+	var sawRetry bool
+	var sawFailed bool
+	deadline := time.After(10 * time.Second)
+	for !sawFailed {
+		select {
+		case evt := <-ch:
+			switch evt.Type {
+			case "run.started", "run.output.batch", "message.created", "item.created":
+			case "run.fault_escalation.retry":
+				sawRetry = true
+			case "run.fault_escalation.exhausted":
+			case "run.failed":
+				sawFailed = true
+			default:
+				t.Fatalf("unexpected event type %q", evt.Type)
+			}
+		case <-deadline:
+			t.Fatal("timed out waiting for run.failed")
+		}
+	}
+	if sawRetry {
+		t.Fatal("did not expect fault-escalation retry when MaxRetries=0")
+	}
+
+	deadline2 := time.After(2 * time.Second)
+	for {
+		executor.mu.Lock()
+		_, registered := executor.running[run.ID]
+		n := len(executor.running)
+		executor.mu.Unlock()
+		if !registered {
+			if n != 0 {
+				t.Fatalf("running map size = %d after terminal finish, want 0", n)
+			}
+			break
+		}
+		select {
+		case <-deadline2:
+			t.Fatal("run still registered after exhausted terminal finish")
+		case <-time.After(10 * time.Millisecond):
+		}
+	}
+
+	// Second finish is a no-op (idempotent teardown).
+	executor.finish(run.ID)
+	executor.mu.Lock()
+	n := len(executor.running)
+	executor.mu.Unlock()
+	if n != 0 {
+		t.Fatalf("running map size = %d after second finish, want 0", n)
+	}
+}
+
 func newTestProcessExecutor(t *testing.T, bus *events.Bus, s store.RunLifecycleStore, mode string) *ProcessExecutor {
 	t.Helper()
 
@@ -1478,6 +1636,9 @@ func newTestProcessExecutor(t *testing.T, bus *events.Bus, s store.RunLifecycleS
 	if err != nil {
 		t.Fatalf("NewProcessExecutor returned error: %v", err)
 	}
+	// Disable fault-escalation by default so existing unit tests observe a single
+	// terminal attempt. Escalation-specific tests re-enable it explicitly.
+	executor.faultEscalationCfg = FaultEscalationConfig{Enabled: false}
 	return executor
 }