fix(passthrough): stop resume loop on first launch of an unstarted session (#2331)
This commit is contained in:
@@ -418,7 +418,10 @@ func (m *Manager) resumeExistingExecution(ctx context.Context, sessionID string,
|
||||
}
|
||||
|
||||
// createExecutionFromSessionInfo creates a new execution for a passthrough session
|
||||
// when no execution exists (e.g., backend restarted and execution store was cleared).
|
||||
// when no execution exists — either because the session has never run (a task
|
||||
// created with start_agent:false, started later) or because a backend restart
|
||||
// cleared the execution store. The two are distinguished by applyResumeIntent,
|
||||
// which decides fresh launch vs resume.
|
||||
//
|
||||
// Terminal sessions are rejected by the guard at the top of createExecution, the
|
||||
// same one every other creation path goes through, so this recovery path never
|
||||
@@ -443,18 +446,24 @@ func (m *Manager) createExecutionFromSessionInfo(ctx context.Context, sessionID
|
||||
}
|
||||
|
||||
// Verify this session should use passthrough mode
|
||||
if err := m.verifyPassthroughEnabled(ctx, sessionID, workspaceExecutionProfileID(info)); err != nil {
|
||||
profileInfo, err := m.verifyPassthroughEnabled(ctx, sessionID, workspaceExecutionProfileID(info))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// If agent ID not in workspace info (snapshot missing/empty), resolve from profile
|
||||
executionProfileID := workspaceExecutionProfileID(info)
|
||||
if info.AgentID == "" && executionProfileID != "" && m.profileResolver != nil {
|
||||
profileInfo, err := m.profileResolver.ResolveProfile(ctx, executionProfileID)
|
||||
// Resolve only to backfill info.AgentID — keep the name distinct from the
|
||||
// outer profileInfo so it's clear this one is not what reaches
|
||||
// startPassthroughExecution below. Both resolve from the same profile ID,
|
||||
// so the content is identical, but avoiding the shadow keeps ownership
|
||||
// unambiguous for future readers.
|
||||
agentProfile, err := m.profileResolver.ResolveProfile(ctx, executionProfileID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("resolve agent for session %s: %w", sessionID, err)
|
||||
}
|
||||
info.AgentID = profileInfo.AgentName
|
||||
info.AgentID = agentProfile.AgentName
|
||||
}
|
||||
|
||||
// Create the execution
|
||||
@@ -468,12 +477,15 @@ func (m *Manager) createExecutionFromSessionInfo(ctx context.Context, sessionID
|
||||
return nil, fmt.Errorf("create execution for session %s: %w", sessionID, err)
|
||||
}
|
||||
|
||||
// Start the passthrough process using resume command (recovery after restart)
|
||||
// createExecution derived the resume intent (applyResumeIntent): a session
|
||||
// with no prior agent execution has never run, so there is nothing for the
|
||||
// CLI's resume flag to attach to and it launches fresh (issue #2330).
|
||||
m.logger.Info("starting passthrough process for session",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("execution_id", execution.ID))
|
||||
zap.String("execution_id", execution.ID),
|
||||
zap.Bool("resumed_session", execution.isResumedSession))
|
||||
|
||||
if err := m.ResumePassthroughSession(ctx, sessionID); err != nil {
|
||||
if err := m.startPassthroughExecution(ctx, execution, profileInfo); err != nil {
|
||||
return nil, fmt.Errorf("start passthrough process for session %s: %w", sessionID, err)
|
||||
}
|
||||
|
||||
@@ -486,10 +498,27 @@ func (m *Manager) createExecutionFromSessionInfo(ctx context.Context, sessionID
|
||||
return execution, nil
|
||||
}
|
||||
|
||||
// verifyPassthroughEnabled checks if the session's profile has CLI passthrough enabled.
|
||||
func (m *Manager) verifyPassthroughEnabled(ctx context.Context, sessionID, profileID string) error {
|
||||
// applyResumeIntent marks whether a freshly built execution should launch as a
|
||||
// resume, from the same PreviousExecutionID buildExecutionFromInstance uses on
|
||||
// the ACP launch path (populated here from info.AgentExecutionID).
|
||||
//
|
||||
// An empty PreviousExecutionID means no agent execution has ever been recorded
|
||||
// for this session — the state a task created with start_agent:false is in.
|
||||
// startPassthroughExecution reads the flag: such a session has no CLI-side
|
||||
// conversation for `-c` / `--resume` to attach to, and its stored prompt has
|
||||
// never been delivered, so it must take the fresh-launch path. Only a session
|
||||
// that previously ran — one whose execution was lost from the in-memory store
|
||||
// by a backend restart — is a genuine recovery.
|
||||
func applyResumeIntent(execution *AgentExecution, req *ExecutorCreateRequest) {
|
||||
execution.isResumedSession = req.PreviousExecutionID != ""
|
||||
}
|
||||
|
||||
// verifyPassthroughEnabled checks if the session's profile has CLI passthrough
|
||||
// enabled, returning the resolved profile so callers can reuse it for command
|
||||
// building instead of resolving twice.
|
||||
func (m *Manager) verifyPassthroughEnabled(ctx context.Context, sessionID, profileID string) (*AgentProfileInfo, error) {
|
||||
if m.profileResolver == nil || profileID == "" {
|
||||
return fmt.Errorf("session %s has no profile configured for passthrough mode", sessionID)
|
||||
return nil, fmt.Errorf("session %s has no profile configured for passthrough mode", sessionID)
|
||||
}
|
||||
|
||||
profileInfo, err := m.profileResolver.ResolveProfile(ctx, profileID)
|
||||
@@ -498,14 +527,14 @@ func (m *Manager) verifyPassthroughEnabled(ctx context.Context, sessionID, profi
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("profile_id", profileID),
|
||||
zap.Error(err))
|
||||
return fmt.Errorf("session %s: failed to resolve profile %s: %w", sessionID, profileID, err)
|
||||
return nil, fmt.Errorf("session %s: failed to resolve profile %s: %w", sessionID, profileID, err)
|
||||
}
|
||||
|
||||
if profileInfo == nil || !profileInfo.CLIPassthrough {
|
||||
return fmt.Errorf("session %s is not configured for CLI passthrough mode", sessionID)
|
||||
return nil, fmt.Errorf("session %s is not configured for CLI passthrough mode", sessionID)
|
||||
}
|
||||
|
||||
return nil
|
||||
return profileInfo, nil
|
||||
}
|
||||
|
||||
// createExecution creates an agentctl execution.
|
||||
@@ -655,6 +684,10 @@ func (m *Manager) createExecution(ctx context.Context, taskID string, info *Work
|
||||
|
||||
execution := runtimeInstance.ToAgentExecution(req)
|
||||
execution.RuntimeName = rt.Name()
|
||||
// Set before executionStore.Add: once the execution is registered, a
|
||||
// concurrent EnsurePassthroughExecution can reach it, and it must never
|
||||
// observe a half-initialised resume intent.
|
||||
applyResumeIntent(execution, req)
|
||||
|
||||
// Cache only agent-profile values for the best-effort configure fallback.
|
||||
// The effective runtime snapshot (including repository secrets) is already
|
||||
|
||||
@@ -982,6 +982,10 @@ func (m *Manager) ResumePassthroughSession(ctx context.Context, sessionID string
|
||||
}
|
||||
|
||||
execution.PassthroughStartedAt = time.Now()
|
||||
// Must precede PassthroughProcessID: handlePassthroughStatus reads
|
||||
// passthroughLaunchUsedResume synchronously once the process ID is visible
|
||||
// (the ID is what routes a status update to this execution), so the flag has
|
||||
// to be set before the process can exit and publish a status.
|
||||
execution.passthroughLaunchUsedResume = useResume
|
||||
execution.PassthroughProcessID = processInfo.ID
|
||||
|
||||
@@ -1107,15 +1111,22 @@ func (m *Manager) handlePassthroughStatus(status *agentctltypes.ProcessStatusUpd
|
||||
// goroutine doesn't race the next launch's writes to those fields.
|
||||
startedAt := execution.PassthroughStartedAt
|
||||
usedResume := execution.passthroughLaunchUsedResume
|
||||
// Detect fast-fail synchronously so we can flip the resume-failed
|
||||
// flag before the next WS reconnect arrives. The goroutine below
|
||||
// would otherwise miss this race: when the PTY exits the WS bridge
|
||||
// closes too, and by the time the goroutine runs (after a 100ms
|
||||
// cleanup delay) HasActiveWebSocketBySession returns false and it
|
||||
// bails without setting any flags. Scoped to fast-fail+usedResume
|
||||
// so a healthy resumed session that exits cleanly or crashes long
|
||||
// after launch keeps its resume intent for auto-restart.
|
||||
if usedResume && passthroughExitIsFastFail(startedAt, status) {
|
||||
// Flip the resume-failed flag synchronously so we beat the next WS
|
||||
// reconnect. The goroutine below would otherwise miss this race:
|
||||
// when the PTY exits the WS bridge closes too, and by the time the
|
||||
// goroutine runs (after a 100ms cleanup delay)
|
||||
// HasActiveWebSocketBySession returns false and it bails without
|
||||
// setting any flags.
|
||||
//
|
||||
// Any non-zero exit of a resume launch counts, not just a fast one:
|
||||
// a real CLI takes seconds to boot before it can report "No
|
||||
// conversation found to continue", which lands outside the
|
||||
// fast-fail window and previously left the flag unset — so every
|
||||
// restart re-attached the same broken resume flag and the session
|
||||
// looped forever (issue #2330). A clean (zero) exit still keeps the
|
||||
// resume intent, so a healthy resumed session that the user quits
|
||||
// is auto-restarted as a resume.
|
||||
if usedResume && passthroughExitCode(status) != 0 {
|
||||
execution.passthroughResumeFailed = true
|
||||
execution.isResumedSession = false
|
||||
execution.passthroughLaunchUsedResume = false
|
||||
@@ -1177,10 +1188,7 @@ func (m *Manager) handlePassthroughExit(execution *AgentExecution, status *agent
|
||||
return
|
||||
}
|
||||
|
||||
exitCode := 0
|
||||
if status.ExitCode != nil {
|
||||
exitCode = *status.ExitCode
|
||||
}
|
||||
exitCode := passthroughExitCode(status)
|
||||
|
||||
// Use the exit timestamp from the status event (set when the child
|
||||
// actually exited), not time.Now() — the cleanupDelay sleep and goroutine
|
||||
@@ -1202,9 +1210,8 @@ func (m *Manager) handlePassthroughExit(execution *AgentExecution, status *agent
|
||||
// fresh command (no resume flag) before giving up.
|
||||
if usedResume {
|
||||
// Resume-failed flags have already been flipped synchronously in
|
||||
// handlePassthroughStatus (see passthroughExitIsFastFail) so any
|
||||
// concurrent WS reconnect that races this goroutine sees the new
|
||||
// values immediately.
|
||||
// handlePassthroughStatus so any concurrent WS reconnect that
|
||||
// races this goroutine sees the new values immediately.
|
||||
m.attemptResumeFallback(execution, interactiveRunner, sessionID, exitCode, uptime)
|
||||
return
|
||||
}
|
||||
@@ -1387,20 +1394,13 @@ func (m *Manager) notifyFastFailExit(runner *process.InteractiveRunner, sessionI
|
||||
}
|
||||
}
|
||||
|
||||
// passthroughExitIsFastFail wraps isFastFailExit for the synchronous
|
||||
// status-callback path that doesn't yet have the unpacked exit code or
|
||||
// timestamp. Same window/semantics as the goroutine path.
|
||||
func passthroughExitIsFastFail(startedAt time.Time, status *agentctltypes.ProcessStatusUpdate) bool {
|
||||
const fastFailWindow = 2 * time.Second
|
||||
exitCode := 0
|
||||
if status.ExitCode != nil {
|
||||
exitCode = *status.ExitCode
|
||||
// passthroughExitCode unpacks the exit code from a process status update,
|
||||
// treating an absent code as a clean exit.
|
||||
func passthroughExitCode(status *agentctltypes.ProcessStatusUpdate) int {
|
||||
if status.ExitCode == nil {
|
||||
return 0
|
||||
}
|
||||
exitedAt := status.Timestamp
|
||||
if exitedAt.IsZero() {
|
||||
exitedAt = time.Now()
|
||||
}
|
||||
return isFastFailExit(startedAt, exitedAt, exitCode, fastFailWindow)
|
||||
return *status.ExitCode
|
||||
}
|
||||
|
||||
// isFastFailExit reports whether a passthrough process exit looks like a
|
||||
|
||||
+205
@@ -0,0 +1,205 @@
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"testing/synctest"
|
||||
"time"
|
||||
|
||||
agentctltypes "github.com/kandev/kandev/internal/agentctl/types"
|
||||
)
|
||||
|
||||
// TestApplyResumeIntent pins the first half of issue #2330: a passthrough
|
||||
// session with no prior agent execution — the state a task created with
|
||||
// start_agent:false is in — has never run, so its first launch must not carry
|
||||
// the CLI resume flag. A session that did run before (its execution lost from
|
||||
// the in-memory store by a backend restart) must still resume.
|
||||
func TestApplyResumeIntent(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
previousExecutionID string
|
||||
wantResumed bool
|
||||
}{
|
||||
{
|
||||
name: "session that has never run is a fresh launch",
|
||||
previousExecutionID: "",
|
||||
wantResumed: false,
|
||||
},
|
||||
{
|
||||
name: "session with a prior execution is a recovery",
|
||||
previousExecutionID: "exec-previous",
|
||||
wantResumed: true,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
execution := &AgentExecution{ID: "exec-1", SessionID: "sess-1"}
|
||||
applyResumeIntent(execution, &ExecutorCreateRequest{
|
||||
SessionID: "sess-1",
|
||||
PreviousExecutionID: tt.previousExecutionID,
|
||||
})
|
||||
if execution.isResumedSession != tt.wantResumed {
|
||||
t.Errorf("isResumedSession = %v, want %v", execution.isResumedSession, tt.wantResumed)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestApplyResumeIntentRoutesPassthroughLaunch pins the mapping from that
|
||||
// intent to the launch path actually taken. The distinction matters beyond the
|
||||
// resume flag: only startPassthroughSession delivers the session's stored
|
||||
// prompt, while ResumePassthroughSession deliberately does not (it would
|
||||
// duplicate the prompt in agent history).
|
||||
func TestApplyResumeIntentRoutesPassthroughLaunch(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
previousExecutionID string
|
||||
wantFreshLaunch bool
|
||||
}{
|
||||
{
|
||||
name: "never ran routes to the fresh-launch path",
|
||||
previousExecutionID: "",
|
||||
wantFreshLaunch: true,
|
||||
},
|
||||
{
|
||||
name: "prior execution routes to the resume path",
|
||||
previousExecutionID: "exec-previous",
|
||||
wantFreshLaunch: false,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
mgr := newTestManager(t)
|
||||
mgr.profileResolver = &mockAgentProfileResolver{cliPassthrough: true}
|
||||
|
||||
execution := &AgentExecution{
|
||||
ID: "exec-1",
|
||||
SessionID: "sess-1",
|
||||
AgentProfileID: "profile-1",
|
||||
IsPassthrough: true,
|
||||
}
|
||||
if err := mgr.executionStore.Add(execution); err != nil {
|
||||
t.Fatalf("seed execution: %v", err)
|
||||
}
|
||||
applyResumeIntent(execution, &ExecutorCreateRequest{
|
||||
SessionID: "sess-1",
|
||||
PreviousExecutionID: tt.previousExecutionID,
|
||||
})
|
||||
|
||||
err := mgr.startPassthroughExecution(context.Background(), execution, nil)
|
||||
if err == nil {
|
||||
t.Fatal("expected error (no interactive runner)")
|
||||
}
|
||||
// Both paths bail on the missing interactive runner, but with
|
||||
// distinct messages: startPassthroughSession's carries the
|
||||
// "for passthrough mode" suffix, ResumePassthroughSession's does not.
|
||||
const (
|
||||
freshLaunchError = "interactive runner not available for passthrough mode"
|
||||
resumeError = "interactive runner not available"
|
||||
)
|
||||
errText := err.Error()
|
||||
gotFreshLaunch := errText == freshLaunchError
|
||||
if !gotFreshLaunch && errText != resumeError {
|
||||
t.Fatalf("unexpected runner-missing error: %v", err)
|
||||
}
|
||||
if gotFreshLaunch != tt.wantFreshLaunch {
|
||||
t.Errorf("fresh-launch path taken = %v, want %v (error: %v)", gotFreshLaunch, tt.wantFreshLaunch, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestManager_HandlePassthroughStatus_ResumeFailureIsSticky pins the second
|
||||
// half of issue #2330: any non-zero exit of a resume launch must set the sticky
|
||||
// resume-failed guard, not only one landing inside the 2s fast-fail window. A
|
||||
// real CLI spends seconds on PTY setup, shell startup and its own boot before
|
||||
// it can report "No conversation found to continue", so gating the guard on
|
||||
// fast-fail left it unset and every auto-restart re-attached the same broken
|
||||
// resume flag — an endless loop.
|
||||
func TestManager_HandlePassthroughStatus_ResumeFailureIsSticky(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
usedResume bool
|
||||
exitCode int
|
||||
uptime time.Duration
|
||||
wantFailed bool
|
||||
}{
|
||||
{
|
||||
name: "slow resume failure (real CLI startup cost)",
|
||||
usedResume: true,
|
||||
exitCode: 1,
|
||||
uptime: 3400 * time.Millisecond,
|
||||
wantFailed: true,
|
||||
},
|
||||
{
|
||||
name: "fast resume failure (bad flag, missing binary)",
|
||||
usedResume: true,
|
||||
exitCode: 127,
|
||||
uptime: 100 * time.Millisecond,
|
||||
wantFailed: true,
|
||||
},
|
||||
{
|
||||
name: "clean exit keeps the resume intent",
|
||||
usedResume: true,
|
||||
exitCode: 0,
|
||||
uptime: 3 * time.Second,
|
||||
wantFailed: false,
|
||||
},
|
||||
{
|
||||
name: "failure of a fresh launch is not a resume failure",
|
||||
usedResume: false,
|
||||
exitCode: 1,
|
||||
uptime: 3 * time.Second,
|
||||
wantFailed: false,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
mgr := newTestManager(t)
|
||||
// Shut the manager down so the auto-restart goroutine
|
||||
// short-circuits: this test covers the synchronous guard flip,
|
||||
// not the relaunch it enables.
|
||||
if err := mgr.StopAllAgents(context.Background()); err != nil {
|
||||
t.Fatalf("StopAllAgents: %v", err)
|
||||
}
|
||||
|
||||
startedAt := time.Now()
|
||||
execution := &AgentExecution{
|
||||
ID: "exec-1",
|
||||
SessionID: "sess-1",
|
||||
PassthroughProcessID: "proc-1",
|
||||
PassthroughStartedAt: startedAt,
|
||||
passthroughLaunchUsedResume: tt.usedResume,
|
||||
isResumedSession: tt.usedResume,
|
||||
}
|
||||
if err := mgr.executionStore.Add(execution); err != nil {
|
||||
t.Fatalf("seed execution: %v", err)
|
||||
}
|
||||
|
||||
exitCode := tt.exitCode
|
||||
mgr.handlePassthroughStatus(&agentctltypes.ProcessStatusUpdate{
|
||||
SessionID: "sess-1",
|
||||
ProcessID: "proc-1",
|
||||
Status: agentctltypes.ProcessStatusExited,
|
||||
ExitCode: &exitCode,
|
||||
Timestamp: startedAt.Add(tt.uptime),
|
||||
})
|
||||
|
||||
if execution.passthroughResumeFailed != tt.wantFailed {
|
||||
t.Errorf("passthroughResumeFailed = %v, want %v",
|
||||
execution.passthroughResumeFailed, tt.wantFailed)
|
||||
}
|
||||
// A tripped guard must also clear the resume intent, otherwise
|
||||
// the next launch would still be routed as a recovery.
|
||||
if tt.wantFailed && (execution.isResumedSession || execution.passthroughLaunchUsedResume) {
|
||||
t.Errorf("resume intent not cleared: isResumedSession=%v passthroughLaunchUsedResume=%v",
|
||||
execution.isResumedSession, execution.passthroughLaunchUsedResume)
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -1196,10 +1196,26 @@ func TestManager_VerifyPassthroughEnabled(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
err := mgr.verifyPassthroughEnabled(context.Background(), "test-session", tt.profileID)
|
||||
profile, err := mgr.verifyPassthroughEnabled(context.Background(), "test-session", tt.profileID)
|
||||
if (err != nil) != tt.wantErr {
|
||||
t.Errorf("verifyPassthroughEnabled() error = %v, wantErr %v", err, tt.wantErr)
|
||||
}
|
||||
if tt.wantErr {
|
||||
return
|
||||
}
|
||||
// The resolved profile is returned so createExecutionFromSessionInfo
|
||||
// can hand it straight to the launch instead of resolving twice —
|
||||
// a nil profile here would silently drop the model, permissions and
|
||||
// cli_flags from the passthrough command.
|
||||
if profile == nil {
|
||||
t.Fatal("verifyPassthroughEnabled() returned a nil profile on success")
|
||||
}
|
||||
if profile.ProfileID != tt.profileID {
|
||||
t.Errorf("profile.ProfileID = %q, want %q", profile.ProfileID, tt.profileID)
|
||||
}
|
||||
if !profile.CLIPassthrough {
|
||||
t.Error("profile.CLIPassthrough = false, want true")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -112,7 +112,7 @@ type AgentExecution struct {
|
||||
// case where the CLI's local conversation history is gone after a backend
|
||||
// restart.
|
||||
passthroughLaunchUsedResume bool
|
||||
// passthroughResumeFailed sticks once a resume launch fast-fails, so that
|
||||
// passthroughResumeFailed sticks once a resume launch exits non-zero, so that
|
||||
// subsequent ResumePassthroughSession calls (e.g. from EnsurePassthroughExecution
|
||||
// when the frontend reconnects its terminal WS) build a fresh command
|
||||
// instead of thrashing on the same broken resume flag.
|
||||
|
||||
Reference in New Issue
Block a user