AttemptID on every lifecycle event
apps/backend/internal/agent/runtime/lifecycle/event_types.go ↗Every lifecycle payload now carries the immutable recovery attempt that owns the callback.
Payloads
type AgentEventPayload struct {
AgentExecutionID string `json:"agent_execution_id"`
AttemptID string `json:"attempt_id,omitempty"`
RunID string `json:"run_id,omitempty"`
TaskID string `json:"task_id"`
SessionID string `json:"session_id,omitempty"`
}
type AgentctlEventPayload struct {
TaskID string `json:"task_id"`
SessionID string `json:"session_id"`
AgentExecutionID string `json:"agent_execution_id"`
AttemptID string `json:"attempt_id,omitempty"`
}
type ACPSessionCreatedPayload struct {
TaskID string `json:"task_id"`
SessionID string `json:"session_id"`
AgentExecutionID string `json:"agent_execution_id"`
AttemptID string `json:"attempt_id,omitempty"`
ACPSessionID string `json:"acp_session_id"`
}
type AgentStreamEventPayload struct {
Type string `json:"type"`
AgentID string `json:"agent_id"`
ExecutionID string `json:"execution_id"`
AttemptID string `json:"attempt_id,omitempty"`
AgentProfileID string `json:"agent_profile_id,omitempty"`
TaskID string `json:"task_id"`
SessionID string `json:"session_id"`
Data *AgentStreamEventData `json:"data"`
}
Process-local resume attempt registry
apps/backend/internal/orchestrator/resume_attempt.go ↗The registry owns one cancellable attempt per session and retains tombstones so late callbacks fail closed.
Registry
type resumeAttempt struct {
id uint64
taskID string
sessionID string
ctx context.Context
cancel context.CancelFunc
done chan struct{}
executionID string
retained bool
}
type resumeAttemptRegistry struct {
mu sync.Mutex
nextID uint64
attempts map[string]*resumeAttempt
tombstones map[string][]resumeAttemptTombstone
recoveryHistory map[string]struct{}
latestExecution map[string]map[string]uint64
}
func (r *resumeAttemptRegistry) begin(parent context.Context, taskID, sessionID string) (*resumeAttempt, bool) {
current := r.attempts[sessionID]
if current != nil && current.ctx.Err() == nil {
return current, false
}
if current != nil {
r.retainLocked(current)
delete(r.attempts, sessionID)
}
r.nextID++
attemptCtx, cancel := context.WithCancel(parent)
attempt := &resumeAttempt{id: r.nextID, taskID: taskID, sessionID: sessionID, ctx: attemptCtx, cancel: cancel, done: make(chan struct{})}
r.attempts[sessionID] = attempt
return attempt, true
}
func (s *Service) resumeAttemptAllowsExecution(sessionID, executionID string, origin ...string) bool {
originID := resumeAttemptOrigin(origin)
registry := s.resumeAttemptStore()
registry.mu.Lock()
defer registry.mu.Unlock()
if current := registry.attempts[sessionID]; current != nil {
return activeResumeAttemptAllowsExecutionLocked(registry, current, executionID, originID)
}
return resumeAttemptTombstoneAllowsExecution(registry, sessionID, executionID, originID)
}
Lifecycle binds attempt to execution generation
apps/backend/internal/agent/runtime/lifecycle/types.go ↗The execution keeps a per-generation attempt map and leases callbacks so a replacement cannot be mutated by an old attempt.
Execution state
type AgentExecution struct {
ID string
ResumeAttemptID string
startupAttemptGeneration uint64
startupRecoveryStarted bool
startupAttemptIDs map[uint64]string
startupLifecycleMu sync.Mutex
startupCallbackMu sync.RWMutex
}
func (e *AgentExecution) beginStartupAttemptWithID(attemptID string) uint64 {
e.startupCallbackMu.Lock()
defer e.startupCallbackMu.Unlock()
e.startupLifecycleMu.Lock()
defer e.startupLifecycleMu.Unlock()
e.startupAttemptGeneration++
e.startupRecoveryStarted = false
e.recordStartupAttemptIDLocked(e.startupAttemptGeneration, attemptID)
return e.startupAttemptGeneration
}
func (e *AgentExecution) withStartupAttempt(generation uint64, callback func(attemptID string)) bool {
e.startupCallbackMu.RLock()
defer e.startupCallbackMu.RUnlock()
e.startupLifecycleMu.Lock()
if e.startupAttemptGeneration != generation {
e.startupLifecycleMu.Unlock()
return false
}
attemptID := e.startupAttemptIDs[generation]
if attemptID == "" {
attemptID = e.ResumeAttemptID
}
e.startupLifecycleMu.Unlock()
callback(attemptID)
return true
}
Inconclusive load failures preserve identity
apps/backend/internal/agent/runtime/lifecycle/session.go ↗Only confirmed unsupported or unknown session errors may fall back to session/new; timeouts and internal errors return without replacing the token.
Load decision
if rt.SessionConfig.NativeSessionResume && existingSessionID != "" {
sessionID, err := sm.loadSession(ctx, client, agentConfig, existingSessionID, mcpServers)
if err == nil {
return sessionID, nil
}
if isTransportDeadErr(err) {
return "", err
}
if !isSessionLoadFallbackErr(err) {
sm.logger.Warn("session/load failed with an inconclusive error, preserving session identity",
zap.String("existing_session_id", existingSessionID),
zap.String("reason", err.Error()))
return "", err
}
sm.logger.Warn("session/load failed, falling back to session/new",
zap.String("existing_session_id", existingSessionID),
zap.String("reason", err.Error()))
return sm.createNewSession(ctx, client, agentConfig, workspacePath, mcpServers)
}
return sm.createNewSession(ctx, client, agentConfig, workspacePath, mcpServers)
Fenced event handlers and exact cleanup
apps/backend/internal/orchestrator/event_handlers_agent.go ↗Boot-ready and ready handlers reject stale attempts and cleanup removes only the exact cancelled execution.
Handler fence
func (s *Service) handleAgentBootReady(ctx context.Context, data watcher.AgentEventData) {
if !s.resumeAttemptAllowsExecution(data.SessionID, data.AgentExecutionID, data.AttemptID) {
s.logger.Debug("ignoring agent.boot_ready from a stale resume attempt",
zap.String("session_id", data.SessionID),
zap.String("attempt_id", data.AttemptID))
return
}
// ... session state checks and drain
}
func (s *Service) cleanupCancelledResumeAttempt(attempt *resumeAttempt) {
executionID := attempt.execution()
if executionID == "" || !s.resumeAttemptStore().canCleanup(attempt) ||
!s.claimForcedExecutionCleanup(attempt.sessionID, executionID) {
return
}
cleanupCtx, cancel := context.WithTimeout(context.Background(), cancellationOperationTTL)
defer cancel()
s.executor.StopExecution(cleanupCtx, executionID, "cancelled resume startup", true)
}
func (s *Service) cleanupStaleResumeExecution(executionID, taskID, sessionID, attemptID string) {
if !s.resumeAttemptStore().canCleanupIdentity(sessionID, executionID, attemptID) {
return
}
go s.cleanupAgentExecution(executionID, taskID, sessionID)
}
Correlated recovery failure presentation
apps/backend/internal/orchestrator/session_recovery_feedback.go ↗Pre-dispatch resume errors carry attempt, execution, and stamp identity so the UI shows one recovery card and suppresses duplicate banners.
Failure identity
type SessionRecoveryIdentity struct {
AttemptID string
ExecutionID string
ErrorStamp string
}
type SessionRecoveryFailure struct {
Err error
Identity SessionRecoveryIdentity
}
func (s *Service) withSessionRecoveryFailureIdentity(ctx context.Context, taskID, sessionID string, attempt *resumeAttempt, executionID string, err error) error {
if err == nil || errors.Is(err, ErrResumeAttemptCancelled) {
return err
}
identity := sessionRecoveryIdentity(attempt, executionID)
identity.ErrorStamp = s.matchingSessionRecoveryErrorStamp(ctx, taskID, sessionID, identity)
return &SessionRecoveryFailure{Err: err, Identity: identity}
}
func (s *Service) HasActiveSessionRecoveryForFailure(ctx context.Context, taskID, sessionID string, failure error) bool {
identity, ok := RecoveryFailureIdentity(failure)
if !ok {
return false
}
lastError, ok := models.LoadLastAgentError(session.Metadata)
return ok && !lastError.IsDismissed() && recoveryIdentityMatchesLastError(identity, lastError)
}
