diff --git a/apps/backend/config/workflows/builtin_engine_test.go b/apps/backend/config/workflows/builtin_engine_test.go index 9e27f03ee4..0bd107c22d 100644 --- a/apps/backend/config/workflows/builtin_engine_test.go +++ b/apps/backend/config/workflows/builtin_engine_test.go @@ -116,6 +116,43 @@ func TestOfficeDefault_TriggersCompileThroughEngine(t *testing.T) { } } +// TestOfficeDefault_OnAgentErrorEscalatesFromEveryAgentLaunchingStep verifies +// that every agent-launching step in office-default (Work, Review, Approval, +// Done) compiles on_agent_error to a single queue_run action targeting the +// CEO agent. Before this fix only Work declared the block, so a reviewer or +// approver session failure left Engine.HandleTrigger with zero actions +// (ActionCount == 0) and no escalation ever reached the coordinator. +func TestOfficeDefault_OnAgentErrorEscalatesFromEveryAgentLaunchingStep(t *testing.T) { + tmpl := loadTemplateForTest(t, "office-default") + + for _, stepName := range []string{"Work", "Review", "Approval", "Done"} { + t.Run(stepName, func(t *testing.T) { + def := findStep(tmpl.Steps, stepName) + if def == nil { + t.Fatalf("step %q not present in template", stepName) + } + spec := engine.CompileStep(stepFromDefinition(*def)) + actions := spec.Events[engine.TriggerOnAgentError] + if len(actions) != 1 { + t.Fatalf("%s.on_agent_error compiled to %d actions, want 1", stepName, len(actions)) + } + action := actions[0] + if action.Kind != engine.ActionQueueRun { + t.Errorf("%s.on_agent_error[0].Kind = %q, want queue_run", stepName, action.Kind) + } + if action.QueueRun == nil { + t.Fatalf("%s.on_agent_error[0].QueueRun is nil", stepName) + } + if action.QueueRun.Target != engine.TargetWorkspaceCEO { + t.Errorf("%s.on_agent_error[0] target = %q, want %q", stepName, action.QueueRun.Target, engine.TargetWorkspaceCEO) + } + if action.QueueRun.Reason != "agent_error" { + t.Errorf("%s.on_agent_error[0] reason = %q, want %q", stepName, action.QueueRun.Reason, "agent_error") + } + }) + } +} + func loadTemplateForTest(t *testing.T, id string) *wfmodels.WorkflowTemplate { t.Helper() templates, err := LoadTemplates() diff --git a/apps/backend/config/workflows/office-default.yml b/apps/backend/config/workflows/office-default.yml index e23fa5fbfc..fe3ba43876 100644 --- a/apps/backend/config/workflows/office-default.yml +++ b/apps/backend/config/workflows/office-default.yml @@ -80,6 +80,12 @@ steps: reason: task_assigned payload: stage_type: review + on_agent_error: + - type: queue_run + config: + target: workspace.ceo_agent + task_id: this + reason: agent_error on_turn_complete: - type: move_to_step config: @@ -113,6 +119,12 @@ steps: reason: task_assigned payload: stage_type: approval + on_agent_error: + - type: queue_run + config: + target: workspace.ceo_agent + task_id: this + reason: agent_error on_turn_complete: - type: move_to_step config: @@ -141,3 +153,9 @@ steps: target: primary task_id: this reason: task_comment + on_agent_error: + - type: queue_run + config: + target: workspace.ceo_agent + task_id: this + reason: agent_error diff --git a/apps/backend/internal/task/repository/sqlite/base.go b/apps/backend/internal/task/repository/sqlite/base.go index bb7433cd50..acb988fe71 100644 --- a/apps/backend/internal/task/repository/sqlite/base.go +++ b/apps/backend/internal/task/repository/sqlite/base.go @@ -69,8 +69,14 @@ type Repository struct { // single-connection test repositories (SetMaxOpenConns(1) serializes // all writes), so this stands in for one. failParticipantSeatReconcileAttempts int - failUsageEventRollupAttempts int - failUsageEventRollupErr error + // failAgentErrorReconcileAttempts is a test-only failpoint for the + // on_agent_error reconciler's bounded retry loop (WO-05-2): while > 0, + // tryHealAgentErrorRow reports a synthetic concurrent-modification retry + // without touching the database, and decrements the counter. Same + // rationale as failParticipantSeatReconcileAttempts above. + failAgentErrorReconcileAttempts int + failUsageEventRollupAttempts int + failUsageEventRollupErr error // usageEventPreRollupHook is a test-only synchronization seam, called (if // set) inside insertUsageEventAndRollup's transaction at the same point as // the failUsageEventRollup* failpoint - after the ledger row insert diff --git a/apps/backend/internal/task/repository/sqlite/base_schema.go b/apps/backend/internal/task/repository/sqlite/base_schema.go index 73ce7d61ca..ae3a4ed68f 100644 --- a/apps/backend/internal/task/repository/sqlite/base_schema.go +++ b/apps/backend/internal/task/repository/sqlite/base_schema.go @@ -39,6 +39,7 @@ func (r *Repository) initSchema() error { r.hideBuiltinWorkflows, r.healBuiltinWorkflowStepFlags, r.healBuiltinWorkflowStepParticipantSeats, + r.healBuiltinWorkflowStepOnAgentError, r.normalizeTaskWorktreeOwnership, r.healDuplicateTaskEnvironments, r.ensureTaskEnvironmentTaskUniqueIndex, diff --git a/apps/backend/internal/task/repository/sqlite/builtin_workflow_agent_error_reconcile.go b/apps/backend/internal/task/repository/sqlite/builtin_workflow_agent_error_reconcile.go new file mode 100644 index 0000000000..bcf07739d7 --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/builtin_workflow_agent_error_reconcile.go @@ -0,0 +1,228 @@ +// Package sqlite provides SQLite-based repository implementations. +package sqlite + +import ( + "encoding/json" + "fmt" + "strings" + + "go.uber.org/zap" + + workflowcfg "github.com/kandev/kandev/config/workflows" + wfmodels "github.com/kandev/kandev/internal/workflow/models" +) + +// maxAgentErrorReconcileAttempts bounds the read-modify-write retry loop in +// tryHealAgentErrorRow. Exhausting it leaves the step unchanged and logs a +// warning rather than propagating an error — see +// healAgentErrorRowWithRetry. +const maxAgentErrorReconcileAttempts = 5 + +// healBuiltinWorkflowStepOnAgentError reconciles workflow_steps.events on +// system-workflow rows whose steps were materialized before their embedded +// template declared an on_agent_error escalation action (WO-05-2). It is +// modelled on healBuiltinWorkflowStepParticipantSeats: for every embedded +// template, for every step, for every on_agent_error action that step's +// template declares, it finds every system-owned workflow_steps row +// materialized from that (template, step name) and appends the action if +// and only if the same normalized action isn't already present — preserving +// every other declared action and leaving +// user-created or user-customised workflows (is_system = 0) alone. +func (r *Repository) healBuiltinWorkflowStepOnAgentError() error { + templates, err := workflowcfg.LoadTemplates() + if err != nil { + return fmt.Errorf("load embedded templates for on_agent_error healing: %w", err) + } + for _, tmpl := range templates { + for _, step := range tmpl.Steps { + r.warnNonQueueRunAgentErrorActions(tmpl.ID, step) + for _, action := range templateAgentErrorActions(step) { + if err := r.healStepOnAgentErrorAction(tmpl.ID, step.Name, action); err != nil { + return fmt.Errorf("heal on_agent_error action for template %s step %s: %w", tmpl.ID, step.Name, err) + } + } + } + } + return nil +} + +// warnNonQueueRunAgentErrorActions logs any on_agent_error action this +// healer does not reconcile (everything but queue_run, per +// templateAgentErrorActions) so a future template adding a different action +// type doesn't leave materialized rows silently un-reconciled. +func (r *Repository) warnNonQueueRunAgentErrorActions(templateID string, step wfmodels.StepDefinition) { + if r.log == nil { + return + } + for _, action := range step.Events.OnAgentError { + if action.Type == wfmodels.GenericActionQueueRun { + continue + } + r.log.Warn("on_agent_error healer only reconciles queue_run actions; skipping", + zap.String("template_id", templateID), + zap.String("step_name", step.Name), + zap.String("action_type", string(action.Type)), + ) + } +} + +// templateAgentErrorActions returns the distinct queue_run on_agent_error +// actions the step's template declares. The key includes the full normalized +// action config so actions that target different tasks or carry different +// payloads are not collapsed. Empty target and task_id values are normalized +// to the same defaults used by the workflow engine. +func templateAgentErrorActions(step wfmodels.StepDefinition) []wfmodels.GenericAction { + var actions []wfmodels.GenericAction + seen := map[string]struct{}{} + for _, action := range step.Events.OnAgentError { + if action.Type != wfmodels.GenericActionQueueRun { + continue + } + key := agentErrorActionKey(action) + if _, ok := seen[key]; ok { + continue + } + seen[key] = struct{}{} + actions = append(actions, action) + } + return actions +} + +// healStepOnAgentErrorAction finds every system-owned workflow_steps row +// materialized from (templateID, stepName) across every workspace and +// reconciles each independently. Rows are collected up front so the +// transactional retry loop below never runs with an open read cursor over +// the same table. +func (r *Repository) healStepOnAgentErrorAction(templateID, stepName string, action wfmodels.GenericAction) error { + targets, err := r.findSystemOwnedWorkflowSteps(templateID, stepName) + if err != nil { + return err + } + + for _, target := range targets { + if err := r.healAgentErrorRowWithRetry(target.stepID, target.workflowID, templateID, stepName, action); err != nil { + return err + } + } + return nil +} + +// healAgentErrorRowWithRetry drives the CAS retry loop for a single +// workflow_steps row: a concurrent writer changing the row between read and +// write is retried a bounded number of times; exhausting the budget leaves +// the step unchanged, logs a warning naming the workflow and step, and +// returns nil so a single un-reconciled step never blocks backend startup. +func (r *Repository) healAgentErrorRowWithRetry(stepID, workflowID, templateID, stepName string, action wfmodels.GenericAction) error { + target, _ := action.Config["target"].(string) + reason, _ := action.Config["reason"].(string) + for attempt := 0; attempt < maxAgentErrorReconcileAttempts; attempt++ { + applied, retry, err := r.tryHealAgentErrorRow(stepID, action) + if err != nil { + return err + } + if applied || !retry { + return nil + } + } + if r.log != nil { + r.log.Warn("exhausted retries reconciling on_agent_error action; leaving step unchanged", + zap.String("template_id", templateID), + zap.String("step_name", stepName), + zap.String("workflow_id", workflowID), + zap.String("step_id", stepID), + zap.String("target", target), + zap.String("reason", reason), + ) + } + return nil +} + +// tryHealAgentErrorRow makes one read-modify-write attempt for stepID. It +// reads the stored events blob, appends the on_agent_error action if no +// equivalent normalized action already exists, and writes back only if the +// row's events column still matches what was read — the +// read value doubles as the optimistic-concurrency guard. applied is true +// when the row already had the action or the write succeeded (both are +// "nothing left to do" outcomes); retry is true only when a concurrent +// writer changed the row between the read and the write. +func (r *Repository) tryHealAgentErrorRow(stepID string, action wfmodels.GenericAction) (applied, retry bool, err error) { + if r.failAgentErrorReconcileAttempts > 0 { + r.failAgentErrorReconcileAttempts-- + return false, true, nil + } + + return r.tryHealWorkflowStepEvents(stepID, "agent-error", func(events *wfmodels.StepEvents) bool { + if hasAgentErrorEscalation(events.OnAgentError, action) { + return true + } + events.OnAgentError = append(events.OnAgentError, action) + return false + }) +} + +// hasAgentErrorEscalation reports whether actions already declares the same +// normalized queue_run on_agent_error action as wanted. +func hasAgentErrorEscalation(actions []wfmodels.GenericAction, wanted wfmodels.GenericAction) bool { + wantedKey := agentErrorActionKey(wanted) + for _, action := range actions { + if action.Type != wfmodels.GenericActionQueueRun { + continue + } + if agentErrorActionKey(action) == wantedKey { + return true + } + } + return false +} + +const ( + // These defaults must stay aligned with engine.readQueueRunConfig. The + // reconciler does not import the engine package because the repository + // layer must not depend on its runtime implementation. + defaultAgentErrorTarget = "primary" + defaultAgentErrorTaskID = "this" +) + +// agentErrorActionKey returns a stable representation of the queue_run action +// as the engine evaluates it. JSON gives map keys a deterministic order and +// normalizes YAML/JSON numeric values across template materialization. +func agentErrorActionKey(action wfmodels.GenericAction) string { + normalized := action + config := make(map[string]interface{}, len(action.Config)+4) + for key, value := range action.Config { + config[key] = value + } + + target, _ := config["target"].(string) + target = strings.TrimSpace(target) + if target == "" { + target = defaultAgentErrorTarget + } + config["target"] = target + + taskID, _ := config["task_id"].(string) + taskID = strings.TrimSpace(taskID) + if taskID == "" { + taskID = defaultAgentErrorTaskID + } + config["task_id"] = taskID + + reason, _ := config["reason"].(string) + config["reason"] = reason + + if payload, ok := config["payload"].(map[string]interface{}); ok { + config["payload"] = payload + } else { + config["payload"] = nil + } + normalized.Config = config + + encoded, err := json.Marshal(normalized) + if err == nil { + return string(encoded) + } + // Template and persisted configs are JSON-compatible. Keep a deterministic + // fallback for malformed in-memory values so one bad action cannot stop + // startup reconciliation. + return fmt.Sprintf("%s:%#v", normalized.Type, normalized.Config) +} diff --git a/apps/backend/internal/task/repository/sqlite/builtin_workflow_agent_error_reconcile_test.go b/apps/backend/internal/task/repository/sqlite/builtin_workflow_agent_error_reconcile_test.go new file mode 100644 index 0000000000..0321d3f147 --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/builtin_workflow_agent_error_reconcile_test.go @@ -0,0 +1,318 @@ +package sqlite + +import ( + "context" + "testing" + "time" + + "go.uber.org/zap" + "go.uber.org/zap/zaptest/observer" + + "github.com/kandev/kandev/internal/common/logger" + wfmodels "github.com/kandev/kandev/internal/workflow/models" +) + +// seedStaleOfficeStepWithoutAgentError inserts a system-owned office-default +// workflow with a single named step whose events have no on_agent_error +// action, simulating a workspace materialized before WO-05-2 shipped. +func seedStaleOfficeStepWithoutAgentError(t *testing.T, repo *Repository, workflowID, stepID, stepName, stageType string) { + t.Helper() + ctx := context.Background() + legacyTime := time.Date(2000, time.January, 1, 0, 0, 0, 0, time.UTC) + + _, err := repo.db.ExecContext(ctx, repo.db.Rebind(` + INSERT INTO workflows ( + id, workspace_id, name, workflow_template_id, is_system, hidden, created_at, updated_at + ) VALUES (?, 'ws-1', 'Office', 'office-default', 1, 1, ?, ?) + `), workflowID, legacyTime, legacyTime) + if err != nil { + t.Fatalf("insert stale office workflow: %v", err) + } + staleEvents := `{"on_turn_complete":[{"type":"move_to_step","config":{"step_id":"done"}}]}` + _, err = repo.db.ExecContext(ctx, repo.db.Rebind(` + INSERT INTO workflow_steps ( + id, workflow_id, name, position, stage_type, events, auto_advance_requires_signal, created_at, updated_at + ) VALUES (?, ?, ?, 2, ?, ?, 0, ?, ?) + `), stepID, workflowID, stepName, stageType, staleEvents, legacyTime, legacyTime) + if err != nil { + t.Fatalf("insert stale %s step: %v", stepName, err) + } +} + +func loadStepEvents(t *testing.T, repo *Repository, stepID string) *wfmodels.WorkflowStep { + t.Helper() + steps := loadStepsForWorkflowByID(t, repo, stepID) + if len(steps) != 1 { + t.Fatalf("expected exactly one step row for id %q, got %d", stepID, len(steps)) + } + return steps[0] +} + +func TestHealBuiltinWorkflowStepOnAgentError_InsertsEscalation(t *testing.T) { + repo := newRepoForBuiltinWorkflowTests(t) + + seedStaleOfficeStepWithoutAgentError(t, repo, "stale-office-agent-err", "stale-office-agent-err-review", "Review", "review") + + if err := repo.healBuiltinWorkflowStepOnAgentError(); err != nil { + t.Fatalf("healBuiltinWorkflowStepOnAgentError: %v", err) + } + + step := loadStepEvents(t, repo, "stale-office-agent-err-review") + if len(step.Events.OnAgentError) != 1 { + t.Fatalf("Review.on_agent_error len = %d, want 1: %+v", len(step.Events.OnAgentError), step.Events.OnAgentError) + } + action := step.Events.OnAgentError[0] + if action.Type != wfmodels.GenericActionQueueRun { + t.Errorf("on_agent_error[0].Type = %q, want queue_run", action.Type) + } + if target, _ := action.Config["target"].(string); target != "workspace.ceo_agent" { + t.Errorf("on_agent_error[0] target = %q, want workspace.ceo_agent", target) + } + if reason, _ := action.Config["reason"].(string); reason != "agent_error" { + t.Errorf("on_agent_error[0] reason = %q, want agent_error", reason) + } + // Existing actions on other triggers must survive reconciliation. + if len(step.Events.OnTurnComplete) != 1 { + t.Errorf("on_turn_complete was modified by the reconciler: %+v", step.Events.OnTurnComplete) + } +} + +func TestHealBuiltinWorkflowStepOnAgentError_Idempotent(t *testing.T) { + repo := newRepoForBuiltinWorkflowTests(t) + + seedStaleOfficeStepWithoutAgentError(t, repo, "stale-office-agent-err-idem", "stale-office-agent-err-idem-review", "Review", "review") + + if err := repo.healBuiltinWorkflowStepOnAgentError(); err != nil { + t.Fatalf("first heal: %v", err) + } + first := loadStepEvents(t, repo, "stale-office-agent-err-idem-review") + + if err := repo.healBuiltinWorkflowStepOnAgentError(); err != nil { + t.Fatalf("second heal: %v", err) + } + second := loadStepEvents(t, repo, "stale-office-agent-err-idem-review") + + if len(second.Events.OnAgentError) != len(first.Events.OnAgentError) { + t.Fatalf("second heal changed on_agent_error length: %d -> %d", len(first.Events.OnAgentError), len(second.Events.OnAgentError)) + } + if len(second.Events.OnAgentError) != 1 { + t.Errorf("running the reconciler twice produced %d escalation actions, want exactly 1", len(second.Events.OnAgentError)) + } +} + +func TestHealBuiltinWorkflowStepOnAgentError_KeepsUserWorkflowUntouched(t *testing.T) { + repo := newRepoForBuiltinWorkflowTests(t) + ctx := context.Background() + legacyTime := time.Date(2000, time.January, 1, 0, 0, 0, 0, time.UTC) + + _, err := repo.db.ExecContext(ctx, repo.db.Rebind(` + INSERT INTO workflows ( + id, workspace_id, name, workflow_template_id, is_system, hidden, created_at, updated_at + ) VALUES ('user-office-agent-err', 'ws-1', 'My Office', 'office-default', 0, 0, ?, ?) + `), legacyTime, legacyTime) + if err != nil { + t.Fatalf("insert user office workflow: %v", err) + } + _, err = repo.db.ExecContext(ctx, repo.db.Rebind(` + INSERT INTO workflow_steps ( + id, workflow_id, name, position, stage_type, events, auto_advance_requires_signal, created_at, updated_at + ) VALUES ('user-office-agent-err-review', 'user-office-agent-err', 'Review', 2, 'review', '{}', 0, ?, ?) + `), legacyTime, legacyTime) + if err != nil { + t.Fatalf("insert user review step: %v", err) + } + + if err := repo.healBuiltinWorkflowStepOnAgentError(); err != nil { + t.Fatalf("healBuiltinWorkflowStepOnAgentError: %v", err) + } + + step := loadStepEvents(t, repo, "user-office-agent-err-review") + if len(step.Events.OnAgentError) != 0 { + t.Errorf("user-customised (is_system=0) workflow step was modified: on_agent_error = %+v", step.Events.OnAgentError) + } +} + +func TestRepositoryInitialization_HealsBuiltinWorkflowStepOnAgentError(t *testing.T) { + repo := newRepoForBuiltinWorkflowTests(t) + + seedStaleOfficeStepWithoutAgentError(t, repo, "stale-office-boot-agent-err", "stale-office-boot-agent-err-review", "Review", "review") + + if err := repo.initSchema(); err != nil { + t.Fatalf("reinitialize repository: %v", err) + } + + step := loadStepEvents(t, repo, "stale-office-boot-agent-err-review") + if len(step.Events.OnAgentError) != 1 { + t.Error("initSchema did not reconcile the on_agent_error action onto a stale Review step") + } +} + +func TestHealAgentErrorRowWithRetry_ExhaustsRetriesWithoutBlockingStartup(t *testing.T) { + repo := newRepoForBuiltinWorkflowTests(t) + core, logs := observer.New(zap.WarnLevel) + log, err := logger.NewFromZap(zap.New(core)) + if err != nil { + t.Fatalf("logger.NewFromZap: %v", err) + } + repo.log = log + + seedStaleOfficeStepWithoutAgentError(t, repo, "stale-office-agent-err-retry-exhaust", "stale-office-agent-err-retry-exhaust-review", "Review", "review") + repo.failAgentErrorReconcileAttempts = maxAgentErrorReconcileAttempts + 5 + + action := wfmodels.GenericAction{ + Type: wfmodels.GenericActionQueueRun, + Config: map[string]interface{}{"target": "workspace.ceo_agent", "task_id": "this", "reason": "agent_error"}, + } + err = repo.healAgentErrorRowWithRetry("stale-office-agent-err-retry-exhaust-review", "stale-office-agent-err-retry-exhaust", "office-default", "Review", action) + if err != nil { + t.Fatalf("healAgentErrorRowWithRetry returned an error; retry exhaustion must never block startup: %v", err) + } + + step := loadStepEvents(t, repo, "stale-office-agent-err-retry-exhaust-review") + if len(step.Events.OnAgentError) != 0 { + t.Error("step was modified despite every attempt reporting a concurrent-modification retry") + } + if logs.Len() != 1 { + t.Fatalf("expected exactly one warning record after retry exhaustion, got %d", logs.Len()) + } + fields := logs.All()[0].ContextMap() + if fields["step_name"] != "Review" || fields["target"] != "workspace.ceo_agent" { + t.Errorf("warning fields = %+v, want step_name=Review target=workspace.ceo_agent", fields) + } +} + +func TestHealAgentErrorRowWithRetry_SucceedsAfterTransientRetries(t *testing.T) { + repo := newRepoForBuiltinWorkflowTests(t) + + seedStaleOfficeStepWithoutAgentError(t, repo, "stale-office-agent-err-retry-ok", "stale-office-agent-err-retry-ok-review", "Review", "review") + repo.failAgentErrorReconcileAttempts = maxAgentErrorReconcileAttempts - 1 + + action := wfmodels.GenericAction{ + Type: wfmodels.GenericActionQueueRun, + Config: map[string]interface{}{"target": "workspace.ceo_agent", "task_id": "this", "reason": "agent_error"}, + } + if err := repo.healAgentErrorRowWithRetry("stale-office-agent-err-retry-ok-review", "stale-office-agent-err-retry-ok", "office-default", "Review", action); err != nil { + t.Fatalf("healAgentErrorRowWithRetry: %v", err) + } + + step := loadStepEvents(t, repo, "stale-office-agent-err-retry-ok-review") + if len(step.Events.OnAgentError) != 1 { + t.Error("expected the escalation action to be inserted once transient retries were exhausted before the attempt budget") + } +} + +func TestHasAgentErrorEscalation(t *testing.T) { + actions := []wfmodels.GenericAction{ + {Type: wfmodels.GenericActionQueueRun, Config: map[string]interface{}{"target": "workspace.ceo_agent", "reason": "agent_error"}}, + } + if !hasAgentErrorEscalation(actions, wfmodels.GenericAction{ + Type: wfmodels.GenericActionQueueRun, + Config: map[string]interface{}{"target": "workspace.ceo_agent", "reason": "agent_error"}, + }) { + t.Error("hasAgentErrorEscalation did not find an equivalent queue_run action") + } + if hasAgentErrorEscalation(actions, wfmodels.GenericAction{ + Type: wfmodels.GenericActionQueueRun, + Config: map[string]interface{}{"target": "primary", "reason": "agent_error"}, + }) { + t.Error("hasAgentErrorEscalation treated a different target as equivalent") + } +} + +func TestHasAgentErrorEscalation_NormalizesDefaultsAndDistinguishesActionConfig(t *testing.T) { + actions := []wfmodels.GenericAction{ + { + Type: wfmodels.GenericActionQueueRun, + Config: map[string]interface{}{ + "target": "primary", + "task_id": "this", + "reason": "agent_error", + "payload": map[string]interface{}{"source": "template"}, + }, + }, + } + + if !hasAgentErrorEscalation(actions, wfmodels.GenericAction{ + Type: wfmodels.GenericActionQueueRun, + Config: map[string]interface{}{"reason": "agent_error", "payload": map[string]interface{}{"source": "template"}}, + }) { + t.Error("hasAgentErrorEscalation did not normalize omitted target and task_id defaults") + } + if hasAgentErrorEscalation(actions, wfmodels.GenericAction{ + Type: wfmodels.GenericActionQueueRun, + Config: map[string]interface{}{"target": "primary", "task_id": "child", "reason": "agent_error", "payload": map[string]interface{}{"source": "template"}}, + }) { + t.Error("hasAgentErrorEscalation ignored a distinct task_id") + } + if hasAgentErrorEscalation(actions, wfmodels.GenericAction{ + Type: wfmodels.GenericActionQueueRun, + Config: map[string]interface{}{"target": "primary", "task_id": "this", "reason": "agent_error", "payload": map[string]interface{}{"source": "other"}}, + }) { + t.Error("hasAgentErrorEscalation ignored a distinct payload") + } +} + +func TestTemplateAgentErrorActions_DedupesByNormalizedFullAction(t *testing.T) { + step := wfmodels.StepDefinition{ + Events: wfmodels.StepEvents{ + OnAgentError: []wfmodels.GenericAction{ + {Type: wfmodels.GenericActionQueueRun, Config: map[string]interface{}{"target": "workspace.ceo_agent", "task_id": "this", "reason": "agent_error", "payload": map[string]interface{}{"mode": "one"}}}, + {Type: wfmodels.GenericActionQueueRun, Config: map[string]interface{}{"target": "workspace.ceo_agent", "task_id": "this", "reason": "agent_error", "payload": map[string]interface{}{"mode": "one"}}}, + {Type: wfmodels.GenericActionQueueRun, Config: map[string]interface{}{"target": "workspace.ceo_agent", "task_id": "child", "reason": "agent_error", "payload": map[string]interface{}{"mode": "one"}}}, + {Type: wfmodels.GenericActionQueueRun, Config: map[string]interface{}{"target": "workspace.ceo_agent", "task_id": "this", "reason": "agent_error", "payload": map[string]interface{}{"mode": "two"}}}, + {Type: wfmodels.GenericActionQueueRun, Config: map[string]interface{}{"reason": "agent_error"}}, + {Type: wfmodels.GenericActionQueueRun, Config: map[string]interface{}{"target": "primary", "task_id": "this", "reason": "agent_error"}}, + {Type: wfmodels.GenericActionMoveToStep, Config: map[string]interface{}{"step_id": "done"}}, + }, + }, + } + actions := templateAgentErrorActions(step) + if len(actions) != 4 { + t.Fatalf("templateAgentErrorActions len = %d, want 4: %+v", len(actions), actions) + } + if target, _ := actions[3].Config["target"].(string); target != "" { + t.Errorf("templateAgentErrorActions kept the default-target declaration as target %q, want it unchanged", target) + } +} + +func TestWarnNonQueueRunAgentErrorActions_LogsSkippedActionTypes(t *testing.T) { + repo := newRepoForBuiltinWorkflowTests(t) + core, logs := observer.New(zap.WarnLevel) + log, err := logger.NewFromZap(zap.New(core)) + if err != nil { + t.Fatalf("logger.NewFromZap: %v", err) + } + repo.log = log + + step := wfmodels.StepDefinition{ + Name: "Review", + Events: wfmodels.StepEvents{ + OnAgentError: []wfmodels.GenericAction{ + {Type: wfmodels.GenericActionQueueRun, Config: map[string]interface{}{"target": "workspace.ceo_agent", "reason": "agent_error"}}, + {Type: wfmodels.GenericActionMoveToStep, Config: map[string]interface{}{"step_id": "done"}}, + }, + }, + } + repo.warnNonQueueRunAgentErrorActions("office-default", step) + + if logs.Len() != 1 { + t.Fatalf("expected exactly one warning for the non-queue_run action, got %d", logs.Len()) + } + fields := logs.All()[0].ContextMap() + if fields["step_name"] != "Review" || fields["action_type"] != string(wfmodels.GenericActionMoveToStep) { + t.Errorf("warning fields = %+v, want step_name=Review action_type=%s", fields, wfmodels.GenericActionMoveToStep) + } +} + +func TestWarnNonQueueRunAgentErrorActions_NilLoggerNoop(t *testing.T) { + repo := newRepoForBuiltinWorkflowTests(t) + step := wfmodels.StepDefinition{ + Name: "Review", + Events: wfmodels.StepEvents{ + OnAgentError: []wfmodels.GenericAction{ + {Type: wfmodels.GenericActionMoveToStep, Config: map[string]interface{}{"step_id": "done"}}, + }, + }, + } + repo.warnNonQueueRunAgentErrorActions("office-default", step) +} diff --git a/apps/backend/internal/task/repository/sqlite/builtin_workflow_reconcile.go b/apps/backend/internal/task/repository/sqlite/builtin_workflow_reconcile.go new file mode 100644 index 0000000000..2bd8debd90 --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/builtin_workflow_reconcile.go @@ -0,0 +1,80 @@ +// Package sqlite provides SQLite-based repository implementations. +package sqlite + +import ( + "database/sql" + "encoding/json" + "errors" + "fmt" + "time" + + wfmodels "github.com/kandev/kandev/internal/workflow/models" +) + +// tryHealWorkflowStepEvents performs the shared read-modify-write operation +// used by builtin workflow reconcilers. The mutator returns true when the +// desired state is already present. A zero-row compare-and-swap result asks +// the caller to retry with a fresh events value. +func (r *Repository) tryHealWorkflowStepEvents( + stepID, operation string, + mutate func(*wfmodels.StepEvents) bool, +) (applied, retry bool, err error) { + tx, err := r.db.Beginx() + if err != nil { + return false, false, fmt.Errorf("begin %s reconciliation tx: %w", operation, err) + } + defer func() { _ = tx.Rollback() }() + + var rawEvents sql.NullString + if err := tx.QueryRow(tx.Rebind(`SELECT events FROM workflow_steps WHERE id = ?`), stepID).Scan(&rawEvents); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return true, false, nil + } + return false, false, fmt.Errorf("read step events: %w", err) + } + + var events wfmodels.StepEvents + if rawEvents.String != "" { + if err := json.Unmarshal([]byte(rawEvents.String), &events); err != nil { + return false, false, fmt.Errorf("parse step events: %w", err) + } + } + + if mutate(&events) { + return true, false, nil + } + updated, err := json.Marshal(events) + if err != nil { + return false, false, fmt.Errorf("marshal step events: %w", err) + } + + // events is TEXT and nullable; NULL requires "IS NULL" on both dialects + // since neither treats a bound parameter after IS/= as NULL-matching. + guardClause := "events = ?" + guardArg := any(rawEvents.String) + if !rawEvents.Valid { + guardClause = "events IS NULL" + guardArg = nil + } + args := []any{string(updated), time.Now().UTC(), stepID} + if guardArg != nil { + args = append(args, guardArg) + } + res, err := tx.Exec(tx.Rebind(` + UPDATE workflow_steps SET events = ?, updated_at = ? + WHERE id = ? AND `+guardClause), args...) + if err != nil { + return false, false, fmt.Errorf("write step events: %w", err) + } + affected, err := res.RowsAffected() + if err != nil { + return false, false, fmt.Errorf("check %s reconciliation write: %w", operation, err) + } + if affected == 0 { + return false, true, nil + } + if err := tx.Commit(); err != nil { + return false, false, fmt.Errorf("commit %s reconciliation: %w", operation, err) + } + return true, false, nil +} diff --git a/apps/backend/internal/task/repository/sqlite/builtin_workflow_seat_reconcile.go b/apps/backend/internal/task/repository/sqlite/builtin_workflow_seat_reconcile.go index 444a9a257d..08ac0bfd29 100644 --- a/apps/backend/internal/task/repository/sqlite/builtin_workflow_seat_reconcile.go +++ b/apps/backend/internal/task/repository/sqlite/builtin_workflow_seat_reconcile.go @@ -2,12 +2,8 @@ package sqlite import ( - "database/sql" - "encoding/json" - "errors" "fmt" "strings" - "time" "go.uber.org/zap" @@ -78,29 +74,10 @@ func templateSeatRoles(step wfmodels.StepDefinition) []string { // transactional retry loop below never runs with an open read cursor over // the same table. func (r *Repository) healStepParticipantSeatRole(templateID, stepName, role string) error { - rows, err := r.db.Query(r.db.Rebind(` - SELECT ws.id, ws.workflow_id FROM workflow_steps ws - JOIN workflows w ON w.id = ws.workflow_id - WHERE w.is_system = 1 AND w.workflow_template_id = ? AND ws.name = ? - `), templateID, stepName) + targets, err := r.findSystemOwnedWorkflowSteps(templateID, stepName) if err != nil { - return fmt.Errorf("find system-owned step rows: %w", err) + return err } - type seatStepRow struct{ stepID, workflowID string } - var targets []seatStepRow - for rows.Next() { - var row seatStepRow - if scanErr := rows.Scan(&row.stepID, &row.workflowID); scanErr != nil { - _ = rows.Close() - return fmt.Errorf("scan step row: %w", scanErr) - } - targets = append(targets, row) - } - if err := rows.Err(); err != nil { - _ = rows.Close() - return fmt.Errorf("iterate step rows: %w", err) - } - _ = rows.Close() for _, target := range targets { if err := r.healParticipantSeatRowWithRetry(target.stepID, target.workflowID, templateID, stepName, role); err != nil { @@ -151,66 +128,13 @@ func (r *Repository) tryHealParticipantSeatRow(stepID, role string) (applied, re return false, true, nil } - tx, err := r.db.Beginx() - if err != nil { - return false, false, fmt.Errorf("begin seat reconciliation tx: %w", err) - } - defer func() { _ = tx.Rollback() }() - - var rawEvents sql.NullString - if err := tx.QueryRow(tx.Rebind(`SELECT events FROM workflow_steps WHERE id = ?`), stepID).Scan(&rawEvents); err != nil { - if errors.Is(err, sql.ErrNoRows) { - return true, false, nil + return r.tryHealWorkflowStepEvents(stepID, "participant seat", func(events *wfmodels.StepEvents) bool { + if hasSeatRole(events.OnEnter, role) { + return true } - return false, false, fmt.Errorf("read step events: %w", err) - } - - var events wfmodels.StepEvents - if rawEvents.String != "" { - if err := json.Unmarshal([]byte(rawEvents.String), &events); err != nil { - return false, false, fmt.Errorf("parse step events: %w", err) - } - } - - if hasSeatRole(events.OnEnter, role) { - return true, false, nil - } - events.OnEnter = insertSeatAction(events.OnEnter, role) - - updated, err := json.Marshal(events) - if err != nil { - return false, false, fmt.Errorf("marshal step events: %w", err) - } - - // events is TEXT and nullable; NULL requires "IS NULL" on both dialects - // since neither treats a bound parameter after IS/= as NULL-matching. - guardClause := "events = ?" - guardArg := any(rawEvents.String) - if !rawEvents.Valid { - guardClause = "events IS NULL" - guardArg = nil - } - args := []any{string(updated), time.Now().UTC(), stepID} - if guardArg != nil { - args = append(args, guardArg) - } - res, err := tx.Exec(tx.Rebind(` - UPDATE workflow_steps SET events = ?, updated_at = ? - WHERE id = ? AND `+guardClause), args...) - if err != nil { - return false, false, fmt.Errorf("write step events: %w", err) - } - affected, err := res.RowsAffected() - if err != nil { - return false, false, fmt.Errorf("check seat reconciliation write: %w", err) - } - if affected == 0 { - return false, true, nil - } - if err := tx.Commit(); err != nil { - return false, false, fmt.Errorf("commit seat reconciliation: %w", err) - } - return true, false, nil + events.OnEnter = insertSeatAction(events.OnEnter, role) + return false + }) } // hasSeatRole reports whether actions already declares an diff --git a/apps/backend/internal/task/repository/sqlite/builtin_workflow_step_rows.go b/apps/backend/internal/task/repository/sqlite/builtin_workflow_step_rows.go new file mode 100644 index 0000000000..866909bc51 --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/builtin_workflow_step_rows.go @@ -0,0 +1,40 @@ +// Package sqlite provides SQLite-based repository implementations. +package sqlite + +import "fmt" + +// systemWorkflowStepRow identifies a single system-owned workflow_steps row +// materialized from a given (template, step name) pair. +type systemWorkflowStepRow struct{ stepID, workflowID string } + +// findSystemOwnedWorkflowSteps returns every system-owned (is_system = 1) +// workflow_steps row materialized from (templateID, stepName) across every +// workspace. Shared by the startup healers that reconcile +// workflow_steps.events onto already-materialized rows (participant seats, +// on_agent_error escalation): rows are collected up front so their +// transactional per-row retry loops never run with an open read cursor over +// the same table. +func (r *Repository) findSystemOwnedWorkflowSteps(templateID, stepName string) ([]systemWorkflowStepRow, error) { + rows, err := r.db.Query(r.db.Rebind(` + SELECT ws.id, ws.workflow_id FROM workflow_steps ws + JOIN workflows w ON w.id = ws.workflow_id + WHERE w.is_system = 1 AND w.workflow_template_id = ? AND ws.name = ? + `), templateID, stepName) + if err != nil { + return nil, fmt.Errorf("find system-owned step rows: %w", err) + } + defer func() { _ = rows.Close() }() + + var targets []systemWorkflowStepRow + for rows.Next() { + var row systemWorkflowStepRow + if scanErr := rows.Scan(&row.stepID, &row.workflowID); scanErr != nil { + return nil, fmt.Errorf("scan step row: %w", scanErr) + } + targets = append(targets, row) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate step rows: %w", err) + } + return targets, nil +} diff --git a/apps/backend/internal/workflow/engine/office_default_smoke_test.go b/apps/backend/internal/workflow/engine/office_default_smoke_test.go index 46caff90ea..43799fcdfb 100644 --- a/apps/backend/internal/workflow/engine/office_default_smoke_test.go +++ b/apps/backend/internal/workflow/engine/office_default_smoke_test.go @@ -195,6 +195,72 @@ func TestOfficeDefaultWorkflow_RejectRoutesBackToWork(t *testing.T) { } } +// TestOfficeDefaultWorkflow_OnAgentErrorEscalatesFromEveryStep proves that a +// reviewer or approver session failure now dispatches the same CEO +// escalation as a worker failure, instead of silently no-opping because the +// step's compiled on_agent_error action list was empty (ActionCount == 0, +// which the dispatcher reads as "not handled"). +func TestOfficeDefaultWorkflow_OnAgentErrorEscalatesFromEveryStep(t *testing.T) { + ctx := context.Background() + tmpl := loadEmbeddedTemplate(t, "office-default") + steps := compileWorkflow(tmpl) + + for _, stepName := range []string{"work", "review", "approval", "done"} { + t.Run(stepName, func(t *testing.T) { + store := newSmokeStore(steps) + queue := &fakeRunQueue{} + parts := newSmokeParticipants(steps) + decisions := newFakeDecisionStore() + + registry := MapRegistry{ + ActionQueueRun: QueueRunCallback{ + Adapter: queue, + Primary: stubPrimary{id: "agent-primary"}, + CEOResolver: stubCEO{id: "agent-ceo"}, + Participants: parts, + }, + } + eng := New(store, registry, + WithRunQueue(queue), + WithParticipantStore(parts), + WithDecisionStore(decisions), + ) + + stepID := steps[stepName].ID + store.setCurrentStep(stepID) + + res, err := eng.HandleTrigger(ctx, HandleInput{ + TaskID: "task-1", SessionID: "sess-1", + Trigger: TriggerOnAgentError, + OperationID: "op-agent-error-" + stepName, + }) + if err != nil { + t.Fatalf("%s.on_agent_error: %v", stepName, err) + } + if res.ActionCount != 1 { + t.Fatalf("%s.on_agent_error ActionCount = %d, want 1", stepName, res.ActionCount) + } + + var found *QueueRunRequest + for i := range queue.calls { + if queue.calls[i].WorkflowStepID == stepID { + found = &queue.calls[i] + break + } + } + if found == nil { + t.Fatalf("%s.on_agent_error: no QueueRunRequest recorded for step %s", stepName, stepID) + } + if found.AgentProfileID != "agent-ceo" { + t.Errorf("%s.on_agent_error queued AgentProfileID = %q, want %q", stepName, found.AgentProfileID, "agent-ceo") + } + if found.Reason != "agent_error" { + t.Errorf("%s.on_agent_error queued Reason = %q, want %q", stepName, found.Reason, "agent_error") + } + }) + } +} + // loadEmbeddedTemplate loads a single workflow template from the embedded YAML. func loadEmbeddedTemplate(t *testing.T, id string) *wfmodels.WorkflowTemplate { t.Helper() diff --git a/docs/specs/office/requirements/runtime.md b/docs/specs/office/requirements/runtime.md index b1411c5afe..bb1f08b5ed 100644 --- a/docs/specs/office/requirements/runtime.md +++ b/docs/specs/office/requirements/runtime.md @@ -9,19 +9,19 @@ owners: ## Overview -When an agent run fails mid-turn (invalid model, auth failure, malformed response, transient upstream), today's UX is: +When an agent run fails mid-turn (invalid model, auth failure, malformed response, transient upstream), the failed wakeup is terminal. The workflow engine can still evaluate a configured `on_agent_error` action and queue a separate coordinator run for escalation. ## Requirements ### REQ-OFFICE-RUNTIME-001: Office Agent Runtime — Error Handling Contract -**Intent:** When an agent run fails mid-turn (invalid model, auth failure, malformed response, transient upstream), today's UX is: +**Intent:** When an agent run fails mid-turn (invalid model, auth failure, malformed response, transient upstream), the failed agent run is terminal. A workflow can use `on_agent_error` to queue a separate coordinator run for escalation. #### Acceptance criteria -- **AC-OFFICE-RUNTIME-001.1:** No automatic retry. Every adapter error is terminal for the wakeup that produced it. -- **AC-OFFICE-RUNTIME-001.2:** The wakeup row is stamped with `status = failed` and `error_message`. No follow-up wakeup is queued from the failure path. -- **AC-OFFICE-RUNTIME-001.3:** Re-runs only happen via explicit user action: **Resume session** in chat, **Mark fixed** on an inbox entry, or task reassignment to a different agent. +- **AC-OFFICE-RUNTIME-001.1:** No automatic retry of the failed agent run. Every adapter error is terminal for the wakeup that produced it. +- **AC-OFFICE-RUNTIME-001.2:** The wakeup row is stamped with `status = failed` and `error_message`. The failure path does not queue another wakeup for the same failed agent. A configured `on_agent_error` action can queue a separate coordinator run. +- **AC-OFFICE-RUNTIME-001.3:** Re-runs of the failed agent happen only via explicit user action: **Resume session** in chat, **Mark fixed** on an inbox entry, or task reassignment to a different agent. A coordinator run from `on_agent_error` is escalation, not a retry. - **AC-OFFICE-RUNTIME-001.4:** **`agent_run_failed`** - one entry per failed (task, agent) wakeup while the agent is below threshold. Title: " failed on ". Action: **Mark fixed** -> dismiss + retry. - **AC-OFFICE-RUNTIME-001.5:** **`agent_paused_after_failures`** - one entry per auto-paused agent. Title: " auto-paused after failures (tasks A, B, C)". Action: **Mark fixed** -> unpause + retry the affected tasks. - **AC-OFFICE-RUNTIME-001.6:** Changing `assignee_agent_instance_id` on a task fires the existing reactivity pipeline, which queues a fresh `task_assigned` wakeup for the new agent.