Kanban on_agent_error dispatcher
apps/backend/internal/orchestrator/event_handlers_agent_error.go ↗This is the new kanban fire site. It checks guards, builds PreloadedState after reconciliation, and calls the engine.
Guard sequence and engine call
func (s *Service) dispatchKanbanAgentErrorTrigger(ctx context.Context, data watcher.AgentEventData) {
deps := s.agentErrorDeps.Load()
if deps == nil || deps.engine == nil {
return
}
if data.SessionID == "" {
return
}
if data.UserInitiated {
s.logger.Debug(msgAgentErrorUserInitiated,
zap.String("task_id", data.TaskID),
zap.String("session_id", data.SessionID))
return
}
session, task, ok := s.resolveAgentErrorDispatchTarget(ctx, data)
if !ok {
return
}
state := s.buildMachineState(ctx, task, session)
operationID := agentErrorOperationID(data.SessionID, data.AgentExecutionID)
s.warnUnregisteredAgentErrorActions(ctx, deps, task, data.SessionID)
result, err := deps.engine.HandleTrigger(ctx, engine.HandleInput{
TaskID: data.TaskID,
SessionID: data.SessionID,
Trigger: engine.TriggerOnAgentError,
OperationID: operationID,
EvaluateOnly: true,
PreloadedState: &state,
Payload: engine.OnAgentErrorPayload{
FailedAgentID: agentErrorFailedAgentID(data, session),
FailedSessionID: data.SessionID,
ErrorMessage: agentErrorMessage(data),
},
})
if err != nil {
s.logger.Error(msgAgentErrorDispatchFailed, zap.Error(err))
return
}
if result.Idempotent {
return
}
if result.Transitioned {
s.applyEngineTransition(ctx, data.TaskID, session, result, engine.TriggerOnAgentError, task.Description, true)
}
if result.ActionCount > 0 {
s.logger.Info(msgAgentErrorDispatched, zap.String("operation_id", operationID))
}
}
Panic recovery wrapper
func (s *Service) dispatchKanbanAgentErrorTriggerRecovered(ctx context.Context, data watcher.AgentEventData) {
defer func() {
if r := recover(); r != nil {
s.logger.Error(msgAgentErrorDispatchPanicked,
zap.String("task_id", data.TaskID),
zap.String("session_id", data.SessionID),
zap.Any("panic", r),
zap.String("stack", string(debug.Stack())))
}
}()
s.dispatchKanbanAgentErrorTrigger(ctx, data)
}
Off-guard recovery funnel
apps/backend/internal/orchestrator/event_handlers_agent.go ↗The funnel now returns a closure. The caller releases the guard, then runs the closure, so auto_start_agent can lock the same guard.
Return dispatch closure
func (s *Service) handleRecoverableFailureLockedState(ctx context.Context, data watcher.AgentEventData) func() {
s.logger.Warn("handling recoverable agent failure", zap.String("task_id", data.TaskID))
s.completeTurnForSession(ctx, data.SessionID)
s.persistLastAgentError(ctx, data)
if s.messageCreator != nil && !s.isOfficeSession(ctx, data.SessionID) {
s.createRecoveryStatusMessage(ctx, data)
}
nextState := models.TaskSessionStateWaitingForInput
if s.isOfficeSession(ctx, data.SessionID) {
nextState = models.TaskSessionStateFailed
}
s.updateTaskSessionState(ctx, data.TaskID, data.SessionID, nextState, data.ErrorMessage, false)
s.writeTaskReviewState(ctx, data.TaskID, data.SessionID)
if nextState == models.TaskSessionStateWaitingForInput && data.SessionID != "" {
session, err := s.repo.GetTaskSession(ctx, data.SessionID)
if err == nil {
if signal, ok := models.LoadPendingStepSignal(session.Metadata); ok {
s.reconcileStepCompletionSignalLocked(ctx, data.TaskID, data.SessionID, signal.StepID)
}
}
}
go s.cleanupAgentExecution(data.AgentExecutionID, data.TaskID, data.SessionID)
return func() {
s.dispatchKanbanAgentErrorTriggerRecovered(context.WithoutCancel(ctx), data)
}
}
Caller releases guard then dispatches
func (s *Service) handleRecoverableFailure(ctx context.Context, data watcher.AgentEventData) {
if data.SessionID == "" {
if dispatch := s.handleRecoverableFailureLockedState(ctx, data); dispatch != nil {
dispatch()
}
return
}
lock, release := s.acquireCancelInFlightGuard(data.SessionID)
lock.Lock()
if _, err := s.repo.GetTaskSession(ctx, data.SessionID); err != nil {
if errors.Is(err, models.ErrTaskSessionNotFound) {
lock.Unlock()
release()
return
}
}
dispatch := s.handleRecoverableFailureLockedState(ctx, data)
lock.Unlock()
release()
if dispatch != nil {
dispatch()
}
}
Failure entry points wire the closure
apps/backend/internal/orchestrator/event_handlers_agent.go ↗handleAgentFailedLocked and handleAgentStartFailed now return the dispatch closure and run it after the guard, covering routes R1, R2, and R3.
handleAgentFailedLocked returns dispatch
func (s *Service) handleAgentFailedLocked(ctx context.Context, data watcher.AgentEventData) func() {
data = s.withPromptAttemptEvidence(data)
defer s.clearPromptAttemptEvidence(data.SessionID, data.AgentExecutionID, data.PromptGeneration)
s.markExecutionFailed(data.SessionID, data.AgentExecutionID)
if drop, _ := s.shouldDropSessionFailure(ctx, data, "agent.failed", true); drop {
s.retireExecutionActivityAndPublish(context.WithoutCancel(ctx), data.TaskID, data.SessionID, data.AgentExecutionID)
return nil
}
if data.SessionID != "" && s.handleTransientFailure(ctx, data) {
return nil
}
if data.SessionID != "" && s.routeDynamicAgentFailure(ctx, data, classifyKanbanFailure(data)) {
return nil
}
s.retireExecutionActivityAndPublish(context.WithoutCancel(ctx), data.TaskID, data.SessionID, data.AgentExecutionID)
errMsg := data.ErrorMessage
if errMsg == "" {
errMsg = defaultAgentFailedMessage
}
s.finalizeAutomationRun(ctx, data.TaskID, false, errMsg)
if data.SessionID != "" {
return s.handleRecoverableFailureLockedState(ctx, data)
}
s.scheduler.HandleTaskCompleted(data.TaskID, false)
s.scheduler.RetryTask(data.TaskID)
s.writeTaskReviewState(ctx, data.TaskID, data.SessionID)
go s.cleanupAgentExecution(data.AgentExecutionID, data.TaskID, data.SessionID)
return nil
}
handleAgentStartFailed defers dispatch
func (s *Service) handleAgentStartFailed(ctx context.Context, taskID, sessionID, agentExecutionID string, err error, fromResume bool) bool {
failureData := watcher.AgentEventData{TaskID: taskID, SessionID: sessionID, AgentExecutionID: agentExecutionID, ErrorMessage: err.Error()}
var unlockGuard func()
var releaseGuard func()
var dispatch func()
defer func() {
if unlockGuard != nil {
unlockGuard()
releaseGuard()
}
if dispatch != nil {
dispatch()
}
}()
if sessionID != "" {
lock, release := s.acquireCancelInFlightGuard(sessionID)
unlockGuard = lock.Unlock
releaseGuard = release
if s.isCancelInFlight(sessionID) {
return true
}
}
if failureData.FailureCode == string(routingerr.CodeManagedRuntimeNpmResolution) {
dispatch = s.handleRecoverableFailureLockedState(ctx, failureData)
return true
}
if !isAuthError(err.Error()) {
return false
}
dispatch = s.handleRecoverableFailureLockedState(ctx, failureData)
return true
}
Transient retry terminal paths dispatch
apps/backend/internal/orchestrator/event_handlers_transient.go ↗When the retry loop exhausts or cannot start, it now surfaces recovery through the same funnel, so R4 and R5 dispatch.
No cached prompt path R4
func (s *Service) retryTransientPrompt(ctx context.Context, taskID, sessionID, execID string) {
if ctx.Err() != nil {
return
}
v, ok := s.lastTurnPrompt.Load(sessionID)
if !ok {
if ctx.Err() != nil {
return
}
s.logger.Warn("transient retry has no cached prompt; surfacing recovery banner", zap.String("session_id", sessionID))
s.resetTransientRetry(sessionID)
s.handleRecoverableFailure(context.Background(), watcher.AgentEventData{
TaskID: taskID, SessionID: sessionID, AgentExecutionID: execID,
ErrorMessage: "Automatic provider retry was not possible. Resume or start fresh to continue.",
})
return
}
// ... stop execution, retire activity ...
if _, err := s.PromptTask(ctx, taskID, sessionID, cp.text, cp.model, cp.planMode, cp.attachments, false); err != nil {
if ctx.Err() != nil {
return
}
s.logger.Error("transient retry prompt failed synchronously; surfacing recovery banner", zap.Error(err))
s.resetTransientRetry(sessionID)
s.handleRecoverableFailure(context.Background(), watcher.AgentEventData{
TaskID: taskID, SessionID: sessionID, AgentExecutionID: execID,
ErrorMessage: "Automatic provider retry could not be started. Resume or start fresh to continue.",
})
}
}
User cancel marks UserInitiated
func (s *Service) CancelTransientRetry(ctx context.Context, taskID, sessionID string) bool {
if err := s.authorizeTaskSessionPair(ctx, taskID, sessionID); err != nil {
return false
}
_, active := s.transientRetries.Load(sessionID)
s.resetTransientRetryWithContext(ctx, sessionID, true)
if !active {
return false
}
execID, _ := s.agentManager.GetExecutionIDForSession(ctx, sessionID)
s.handleRecoverableFailure(ctx, watcher.AgentEventData{
TaskID: taskID, SessionID: sessionID, AgentExecutionID: execID,
ErrorMessage: "Automatic provider retries cancelled. Resume or start fresh to continue.",
UserInitiated: true,
})
return true
}
Payload, operation ID, and trigger label
apps/backend/internal/workflow/models/models.go ↗The payload maps failure identity to engine fields, the operation ID makes retries idempotent, and the new trigger label keeps recovery moves distinct in history.
Watcher event field
type AgentEventData struct {
TaskID string `json:"task_id"`
SessionID string `json:"session_id"`
AgentExecutionID string `json:"agent_execution_id"`
AgentProfileID string `json:"agent_profile_id"`
ErrorMessage string `json:"error_message,omitempty"`
UserInitiated bool `json:"user_initiated,omitempty"`
}
Operation ID and payload helpers
func agentErrorOperationID(sessionID, agentExecutionID string) string {
if agentExecutionID == "" {
return fmt.Sprintf("agent_error:session:%s", sessionID)
}
return fmt.Sprintf("agent_error:session:%s:%s", sessionID, agentExecutionID)
}
func agentErrorFailedAgentID(data watcher.AgentEventData, session *models.TaskSession) string {
if data.AgentProfileID != "" {
return data.AgentProfileID
}
if session != nil {
return session.AgentProfileID
}
return ""
}
func agentErrorMessage(data watcher.AgentEventData) string {
if data.ErrorMessage != "" {
return data.ErrorMessage
}
return defaultAgentFailedMessage
}
New history trigger
const (
StepTransitionTriggerManual StepTransitionTrigger = "manual"
StepTransitionTriggerAutoComplete StepTransitionTrigger = "auto_complete"
StepTransitionTriggerAgentError StepTransitionTrigger = "on_agent_error"
)
Pinned fire-site guard
apps/backend/internal/orchestrator/agent_error_fire_site_pin_test.go ↗The test walks the whole backend AST and asserts exactly two functions fire TriggerOnAgentError, so a third dispatcher or a removed one fails the build.
Registered sites
var registeredAgentErrorFireSites = []string{
"internal/office/service/Service.dispatchAgentErrorTrigger",
"internal/orchestrator/Service.dispatchKanbanAgentErrorTrigger",
}
func TestOnAgentErrorFireSitesArePinned(t *testing.T) {
root, err := findAgentErrorBackendSourceRoot(".")
found, err := findAgentErrorFireSites(root)
registered := make(map[string]bool)
for _, name := range registeredAgentErrorFireSites {
registered[name] = true
}
var unregistered []string
for name := range found {
if !registered[name] {
unregistered = append(unregistered, name)
}
}
if len(unregistered) > 0 {
t.Fatalf("function(s) %v fire engine.TriggerOnAgentError but are not in registeredAgentErrorFireSites", unregistered)
}
}
Syntactic fire-site definition
func functionFiresTriggerOnAgentError(body *ast.BlockStmt) bool {
fires := false
ast.Inspect(body, func(n ast.Node) bool {
switch v := n.(type) {
case *ast.CompositeLit:
sel, ok := v.Type.(*ast.SelectorExpr)
if !ok || sel.Sel.Name != "HandleInput" {
return true
}
for _, elt := range v.Elts {
kv, ok := elt.(*ast.KeyValueExpr)
if !ok { continue }
key, ok := kv.Key.(*ast.Ident)
if !ok || key.Name != "Trigger" { continue }
if identIsTriggerOnAgentError(kv.Value) {
fires = true
}
}
case *ast.CallExpr:
sel, ok := v.Fun.(*ast.SelectorExpr)
if !ok || sel.Sel.Name != "dispatchEngineTrigger" { return true }
for _, arg := range v.Args {
if identIsTriggerOnAgentError(arg) { fires = true }
}
}
return true
})
return fires
}
