diff --git a/apps/backend/internal/office/repository/sqlite/blockers.go b/apps/backend/internal/office/repository/sqlite/blockers.go index 5053bb4284..5d0f04f645 100644 --- a/apps/backend/internal/office/repository/sqlite/blockers.go +++ b/apps/backend/internal/office/repository/sqlite/blockers.go @@ -175,6 +175,35 @@ func (r *Repository) AreAllChildrenTerminal(ctx context.Context, parentID string return nonTerminal == 0, nil } +// ChildState is a minimal id+state pair over a parent's child tasks. +type ChildState struct { + TaskID string `db:"id"` + State string `db:"state"` +} + +// ListChildStates returns id+state for every child of a parent task, +// ordered by id. Callers that wake a parent once AreAllChildrenTerminal +// reports true use this to build a wake-idempotency key that changes +// across delegation waves — a parent that fans out again after one wave +// completes gets a distinct child set (and therefore a distinct key), so +// it can be woken again instead of colliding with the permanently-unique +// {reason}:{taskID}:{agentID} key QueueRunCtx mints by default. +func (r *Repository) ListChildStates(ctx context.Context, parentID string) ([]ChildState, error) { + var rows []ChildState + err := r.ro.SelectContext(ctx, &rows, r.ro.Rebind(` + SELECT id, COALESCE(state, '') AS state FROM tasks + WHERE parent_id = ? + ORDER BY id + `), parentID) + if err != nil { + return nil, err + } + if rows == nil { + rows = []ChildState{} + } + return rows, nil +} + // ChildSummary holds summary data for a completed child task. type ChildSummary struct { TaskID string `db:"id" json:"id"` diff --git a/apps/backend/internal/office/repository/sqlite/blockers_test.go b/apps/backend/internal/office/repository/sqlite/blockers_test.go index bede188bca..6d0ccf1475 100644 --- a/apps/backend/internal/office/repository/sqlite/blockers_test.go +++ b/apps/backend/internal/office/repository/sqlite/blockers_test.go @@ -170,6 +170,60 @@ func TestGetChildSummaries_NoChildren(t *testing.T) { } } +func TestListChildStates(t *testing.T) { + repo := newSearchTestRepo(t) + ctx := context.Background() + + insertTask(t, repo, ctx, "parent-1", "ws-1", "Parent Task", "", "") + insertTask(t, repo, ctx, "child-z", "ws-1", "Child Z", "", "") + insertTask(t, repo, ctx, "child-a", "ws-1", "Child A", "", "") + insertTask(t, repo, ctx, "child-null", "ws-1", "Child Null", "", "") + if _, err := repo.ExecRaw(ctx, ` + UPDATE tasks + SET parent_id = 'parent-1', state = CASE id + WHEN 'child-z' THEN 'CANCELLED' + WHEN 'child-a' THEN 'COMPLETED' + WHEN 'child-null' THEN NULL + END + WHERE id IN ('child-z', 'child-a', 'child-null') + `); err != nil { + t.Fatalf("set child states: %v", err) + } + + states, err := repo.ListChildStates(ctx, "parent-1") + if err != nil { + t.Fatalf("ListChildStates: %v", err) + } + if len(states) != 3 { + t.Fatalf("child state count = %d, want 3", len(states)) + } + want := []struct { + id string + state string + }{ + {id: "child-a", state: "COMPLETED"}, + {id: "child-null", state: ""}, + {id: "child-z", state: "CANCELLED"}, + } + for i, got := range states { + if got.TaskID != want[i].id || got.State != want[i].state { + t.Errorf("state[%d] = {%q, %q}, want {%q, %q}", + i, got.TaskID, got.State, want[i].id, want[i].state) + } + } + + empty, err := repo.ListChildStates(ctx, "missing-parent") + if err != nil { + t.Fatalf("ListChildStates empty: %v", err) + } + if empty == nil { + t.Fatal("empty child state result is nil") + } + if len(empty) != 0 { + t.Fatalf("empty child state count = %d, want 0", len(empty)) + } +} + func TestListBlockersForTasks(t *testing.T) { repo := newTestRepo(t) ctx := context.Background() diff --git a/apps/backend/internal/office/scheduler/reactivity.go b/apps/backend/internal/office/scheduler/reactivity.go index 32a05665a1..0b57aa1589 100644 --- a/apps/backend/internal/office/scheduler/reactivity.go +++ b/apps/backend/internal/office/scheduler/reactivity.go @@ -2,11 +2,14 @@ package scheduler import ( "context" + "crypto/sha256" "fmt" + "strings" "go.uber.org/zap" "github.com/kandev/kandev/internal/office/models" + "github.com/kandev/kandev/internal/office/repository/sqlite" ) // Canonical lowercase status values used inside the pipeline. Backend @@ -368,6 +371,16 @@ func (ss *SchedulerService) cascadeBlockersResolved( // cascadeChildrenCompleted wakes the parent if all its children are now // in a terminal state. +// +// The wake's idempotency key is a digest over the parent's current child +// set, not the default {reason}:{taskID}:{agentID} key — that default is +// permanently unique per (reason, task, agent), so a parent that +// delegates iteratively (finish wave 1, read the result, fan out wave 2) +// would only ever be woken for its first wave: wave 2's key would +// collide with wave 1's already-consumed one. Keying on the child set +// instead means a new delegation wave — which necessarily adds new child +// task IDs — produces a distinct key, while a duplicate cascade over the +// same terminal children still dedupes to the same key. func (ss *SchedulerService) cascadeChildrenCompleted( ctx context.Context, task *TaskSnapshot, queue func(string, RunContext), ) { @@ -382,14 +395,53 @@ func (ss *SchedulerService) cascadeChildrenCompleted( if err != nil || parentAssignee == "" { return } + // ListChildStates is a separate read from AreAllChildrenTerminal, with no + // enclosing transaction. A concurrent child insert between the two reads + // can produce a snapshot that is no longer all-terminal. Do not queue from + // that snapshot. A newly inserted child then adds its ID to the later + // terminal snapshot, so that transition can queue a distinct wake. + children, err := ss.repo.ListChildStates(ctx, task.ParentID) + if err != nil { + ss.logger.Error("list child states failed", + zap.String("parent_id", task.ParentID), zap.Error(err)) + return + } + for _, child := range children { + if child.State != "COMPLETED" && child.State != "CANCELLED" { + return + } + } queue(parentAssignee, RunContext{ - Reason: RunReasonTaskChildrenCompleted, - TaskID: task.ParentID, - WorkspaceID: task.WorkspaceID, - ChildTaskID: task.ID, + Reason: RunReasonTaskChildrenCompleted, + TaskID: task.ParentID, + WorkspaceID: task.WorkspaceID, + ChildTaskID: task.ID, + IdempotencyKey: childrenCompletedIdempotencyKey(task.ParentID, parentAssignee, children), }) } +// childrenCompletedIdempotencyKey digests the parent's child ID set +// (already ordered by id via ListChildStates) into an idempotency key for +// the task_children_completed wake. See cascadeChildrenCompleted for why +// this must vary across delegation waves instead of being permanently +// unique per (parent, agent). +// +// Only IDs go into the digest, not each child's state string. By the time +// this runs, cascadeChildrenCompleted has already confirmed every child is +// terminal, so a wave is identified by WHICH children exist, not which +// terminal state each one happens to be in — hashing the state as well +// would make an unrelated terminal-to-terminal edit (e.g. a cancelled +// sibling later marked completed by a user) look like a new wave and +// produce a spurious wake, even though the child set never changed. +func childrenCompletedIdempotencyKey(parentID, agentID string, children []sqlite.ChildState) string { + parts := make([]string, 0, len(children)) + for _, c := range children { + parts = append(parts, c.TaskID) + } + sum := sha256.Sum256([]byte(strings.Join(parts, ","))) + return fmt.Sprintf("%s:%s:%s:%x", RunReasonTaskChildrenCompleted, parentID, agentID, sum) +} + // allBlockersResolvedExcept returns true if every blocker on `taskID` // other than `excludeBlockerID` is in a terminal step. func (ss *SchedulerService) allBlockersResolvedExcept( diff --git a/apps/backend/internal/office/scheduler/reactivity_children_completed_test.go b/apps/backend/internal/office/scheduler/reactivity_children_completed_test.go new file mode 100644 index 0000000000..01ae6fbabe --- /dev/null +++ b/apps/backend/internal/office/scheduler/reactivity_children_completed_test.go @@ -0,0 +1,293 @@ +package scheduler + +import ( + "context" + "testing" + "time" + + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/office/models" + officesqlite "github.com/kandev/kandev/internal/office/repository/sqlite" + "github.com/kandev/kandev/internal/office/service" +) + +// newChildrenCompletedTestScheduler wires a real *service.Service (not +// nil, unlike newReactivityTestScheduler) because these tests exercise +// QueueRunCtx end to end — QueueRun's guardAgentStatus calls +// svc.GetAgentFromConfig, which panics on a nil Service. +func newChildrenCompletedTestScheduler(t *testing.T, repo *officesqlite.Repository) *SchedulerService { + t.Helper() + log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "console"}) + if err != nil { + t.Fatalf("logger: %v", err) + } + svc := service.NewService(service.ServiceOptions{Repo: repo, Logger: log}) + return NewSchedulerService(repo, log, svc) +} + +// createChildrenCompletedAgent registers an idle agent instance so +// guardAgentStatus (paused/stopped/pending-approval checks) passes. +func createChildrenCompletedAgent(t *testing.T, repo *officesqlite.Repository, id string) { + t.Helper() + if err := repo.CreateAgentInstance(context.Background(), &models.AgentInstance{ + ID: id, + WorkspaceID: "ws-1", + Name: id, + Status: models.AgentStatusIdle, + }); err != nil { + t.Fatalf("create agent instance %s: %v", id, err) + } +} + +// newChildrenCompletedQueue returns a queue closure that persists through +// the real QueueRunCtx path (mirroring the closure ApplyTaskMutation +// builds), so tests can assert against actually-persisted `runs` rows +// instead of a summary — QueueRun returns nil both when a run is +// persisted and when it is skipped as an idempotent duplicate +// (scheduler/run.go, "run skipped (idempotent)"), so a summary or a +// bare "no error" check false-greens on the exact defect these tests +// guard against. +func newChildrenCompletedQueue(t *testing.T, ss *SchedulerService) func(string, RunContext) { + t.Helper() + return func(agentID string, c RunContext) { + if err := ss.QueueRunCtx(context.Background(), agentID, c); err != nil { + t.Fatalf("QueueRunCtx: %v", err) + } + } +} + +func insertChildTask(t *testing.T, ss *SchedulerService, id, parentID, state string) { + t.Helper() + if _, err := ss.repo.ExecRaw(context.Background(), ` + INSERT INTO tasks (id, workspace_id, parent_id, state) + VALUES (?, 'ws-1', ?, ?) + `, id, parentID, state); err != nil { + t.Fatalf("insert child task %s: %v", id, err) + } +} + +func runsCountForReason(t *testing.T, ss *SchedulerService, reason string) int { + t.Helper() + var count int + row := ss.repo.ReaderDB().QueryRowx( + ss.repo.ReaderDB().Rebind(`SELECT COUNT(*) FROM runs WHERE reason = ?`), reason) + if err := row.Scan(&count); err != nil { + t.Fatalf("count runs: %v", err) + } + return count +} + +// ageRunsRequestedAt pushes every persisted run's requested_at back by +// the given duration, so tests can exercise the >24h idempotency-window +// path (IdempotencyWindowHours) without a real sleep. +func ageRunsRequestedAt(t *testing.T, ss *SchedulerService, by time.Duration) { + t.Helper() + // Bind a native time.Time (not a pre-formatted string): the sqlite + // driver's own time encoding is what CreateRun's INSERT and + // CoalesceRun's SELECT both use, and a hand-formatted string that + // looks "earlier" can still sort later once TEXT-compared against a + // driver-formatted cutoff, since the two encodings aren't identical. + cutoff := time.Now().UTC().Add(-by) + if _, err := ss.repo.ExecRaw(context.Background(), + `UPDATE runs SET requested_at = ?`, cutoff); err != nil { + t.Fatalf("age runs: %v", err) + } +} + +// setupChildrenCompletedParent creates a parent task with a +// workflow_step_participants "runner" row so GetTaskAssignee resolves it +// to agentID (RunnerProjection's task-scoped runner fallback). An empty +// workflow_steps table is also required: RunnerProjection references it +// in a COALESCE branch that SQLite still needs to compile even when that +// branch never matches. +func setupChildrenCompletedParent(t *testing.T, ss *SchedulerService, parentID, agentID string) { + t.Helper() + ctx := context.Background() + if _, err := ss.repo.ExecRaw(ctx, ` + CREATE TABLE IF NOT EXISTS workflow_steps ( + id TEXT PRIMARY KEY, + agent_profile_id TEXT DEFAULT '' + ) + `); err != nil { + t.Fatalf("create workflow_steps table: %v", err) + } + if _, err := ss.repo.ExecRaw(ctx, ` + INSERT INTO tasks (id, workspace_id) VALUES (?, 'ws-1') + `, parentID); err != nil { + t.Fatalf("insert parent task: %v", err) + } + if _, err := ss.repo.ExecRaw(ctx, ` + INSERT INTO workflow_step_participants + (id, step_id, task_id, role, agent_profile_id, decision_required, position) + VALUES ('p-runner', 'step-x', ?, 'runner', ?, 0, 0) + `, parentID, agentID); err != nil { + t.Fatalf("insert runner participant: %v", err) + } +} + +// TestCascadeChildrenCompleted_SecondWaveDifferentChildSet_PersistsNewRun +// is the regression test for the once-ever idempotency-key defect: a +// parent that delegates in two sequential waves (finish wave 1, then fan +// out wave 2) must be woken for BOTH waves, not just the first. Before +// the fix, QueueRunCtx derived the idempotency key from only +// {reason}:{taskID}:{agentID} — identical for every wave — so wave 2's +// wake collided with wave 1's already-consumed key and QueueRun silently +// returned nil without persisting a row (scheduler/run.go, "run skipped +// (idempotent)"). That is why this test counts persisted `runs` rows +// rather than checking for an error. +func TestCascadeChildrenCompleted_SecondWaveDifferentChildSet_PersistsNewRun(t *testing.T) { + repo := newReactivityTestRepo(t) + ss := newChildrenCompletedTestScheduler(t, repo) + createChildrenCompletedAgent(t, repo, "agent-1") + setupChildrenCompletedParent(t, ss, "parent-1", "agent-1") + queue := newChildrenCompletedQueue(t, ss) + ctx := context.Background() + + // Wave 1: child-1 completes, parent has no other children. + insertChildTask(t, ss, "child-1", "parent-1", "COMPLETED") + ss.cascadeChildrenCompleted(ctx, &TaskSnapshot{ID: "child-1", WorkspaceID: "ws-1", ParentID: "parent-1"}, queue) + + if got := runsCountForReason(t, ss, RunReasonTaskChildrenCompleted); got != 1 { + t.Fatalf("after wave 1: persisted runs = %d, want 1", got) + } + + // Push wave 1's run outside CoalesceWindowSeconds (5s) so wave 2 + // creates its own row instead of merging into wave 1's still-queued + // one — in production the two waves are minutes/hours apart (wave 2 + // only exists because the parent woke, read wave 1's result, and + // delegated again), so this just makes that separation real in a + // synchronous test. + ageRunsRequestedAt(t, ss, 10*time.Second) + + // Wave 2: parent delegates again, a new child completes. + insertChildTask(t, ss, "child-2", "parent-1", "COMPLETED") + ss.cascadeChildrenCompleted(ctx, &TaskSnapshot{ID: "child-2", WorkspaceID: "ws-1", ParentID: "parent-1"}, queue) + + if got := runsCountForReason(t, ss, RunReasonTaskChildrenCompleted); got != 2 { + t.Fatalf("after wave 2: persisted runs = %d, want 2 (parent must be woken again for the new wave)", got) + } +} + +// TestCascadeChildrenCompleted_SameChildSetTwice_StillDedupes proves the +// fix didn't trade "once-ever" for "always fires": a duplicate cascade +// over the exact same terminal child set (e.g. a re-delivered event) must +// still dedupe to a single persisted row. +func TestCascadeChildrenCompleted_SameChildSetTwice_StillDedupes(t *testing.T) { + repo := newReactivityTestRepo(t) + ss := newChildrenCompletedTestScheduler(t, repo) + createChildrenCompletedAgent(t, repo, "agent-1") + setupChildrenCompletedParent(t, ss, "parent-1", "agent-1") + queue := newChildrenCompletedQueue(t, ss) + ctx := context.Background() + + insertChildTask(t, ss, "child-1", "parent-1", "COMPLETED") + task := &TaskSnapshot{ID: "child-1", WorkspaceID: "ws-1", ParentID: "parent-1"} + + ss.cascadeChildrenCompleted(ctx, task, queue) + ss.cascadeChildrenCompleted(ctx, task, queue) + + if got := runsCountForReason(t, ss, RunReasonTaskChildrenCompleted); got != 1 { + t.Fatalf("persisted runs = %d, want exactly 1 (same child set must dedupe)", got) + } +} + +// TestCascadeChildrenCompleted_DistinctWaveBeyondIdempotencyWindow_NoInsertError +// covers the shape-change the triage evidence documented: before the fix, +// ANY second wake for the same parent+agent past IdempotencyWindowHours +// (24h) fell through CheckIdempotencyKey's time-bounded guard into +// CreateRun and hit "UNIQUE constraint failed: runs.idempotency_key", +// because the key never changed. A genuinely distinct wave (new child +// set) must insert cleanly regardless of how much time elapsed since the +// previous wave — the fix makes the key vary with content, not just with +// the guard's time window. +func TestCascadeChildrenCompleted_DistinctWaveBeyondIdempotencyWindow_NoInsertError(t *testing.T) { + repo := newReactivityTestRepo(t) + ss := newChildrenCompletedTestScheduler(t, repo) + createChildrenCompletedAgent(t, repo, "agent-1") + setupChildrenCompletedParent(t, ss, "parent-1", "agent-1") + queue := newChildrenCompletedQueue(t, ss) + ctx := context.Background() + + insertChildTask(t, ss, "child-1", "parent-1", "COMPLETED") + ss.cascadeChildrenCompleted(ctx, &TaskSnapshot{ID: "child-1", WorkspaceID: "ws-1", ParentID: "parent-1"}, queue) + + // Push wave 1's row outside the 24h idempotency window. + ageRunsRequestedAt(t, ss, 25*time.Hour) + + // Wave 2, a distinct child set, fires more than 24h after wave 1. + insertChildTask(t, ss, "child-2", "parent-1", "COMPLETED") + ss.cascadeChildrenCompleted(ctx, &TaskSnapshot{ID: "child-2", WorkspaceID: "ws-1", ParentID: "parent-1"}, queue) + + if got := runsCountForReason(t, ss, RunReasonTaskChildrenCompleted); got != 2 { + t.Fatalf("persisted runs = %d, want 2 (distinct wave past the window must not collide)", got) + } +} + +// TestCascadeChildrenCompleted_ChildTerminalStateEdited_DoesNotPersistNewRun +// is the regression test for a review finding (PR #3059): reactToStatusChange +// invokes cascadeChildrenCompleted on any transition into "done", regardless +// of the child's PRIOR state, so a sibling that is later edited from +// CANCELLED to COMPLETED (a terminal-to-terminal transition, not a new +// delegation wave) fires this cascade again. If the idempotency key digested +// each child's state string, that edit alone would change the digest and +// persist a spurious second run even though the child ID set never changed. +// The parent must be woken exactly once for an unchanged child set. +func TestCascadeChildrenCompleted_ChildTerminalStateEdited_DoesNotPersistNewRun(t *testing.T) { + repo := newReactivityTestRepo(t) + ss := newChildrenCompletedTestScheduler(t, repo) + createChildrenCompletedAgent(t, repo, "agent-1") + setupChildrenCompletedParent(t, ss, "parent-1", "agent-1") + queue := newChildrenCompletedQueue(t, ss) + ctx := context.Background() + + insertChildTask(t, ss, "child-1", "parent-1", "CANCELLED") + task := &TaskSnapshot{ID: "child-1", WorkspaceID: "ws-1", ParentID: "parent-1"} + ss.cascadeChildrenCompleted(ctx, task, queue) + + if got := runsCountForReason(t, ss, RunReasonTaskChildrenCompleted); got != 1 { + t.Fatalf("after first terminal state: persisted runs = %d, want 1", got) + } + + ageRunsRequestedAt(t, ss, 10*time.Second) + + // Same child, edited from CANCELLED to COMPLETED — still the only + // child of parent-1, so the child ID set is unchanged. + if _, err := ss.repo.ExecRaw(ctx, `UPDATE tasks SET state = 'COMPLETED' WHERE id = 'child-1'`); err != nil { + t.Fatalf("edit child-1 state: %v", err) + } + ss.cascadeChildrenCompleted(ctx, task, queue) + + if got := runsCountForReason(t, ss, RunReasonTaskChildrenCompleted); got != 1 { + t.Fatalf("after terminal-to-terminal edit: persisted runs = %d, want 1 (unchanged child set must not re-wake)", got) + } +} + +// TestCascadeChildrenCompleted_NonterminalSnapshotDoesNotQueueWake covers the +// inconsistent-read shape where the terminal check succeeds but the child +// snapshot contains a nonterminal row. A NULL state is normalized to an empty +// state by ListChildStates and is not terminal for the completion contract. +func TestCascadeChildrenCompleted_NonterminalSnapshotDoesNotQueueWake(t *testing.T) { + repo := newReactivityTestRepo(t) + ss := newChildrenCompletedTestScheduler(t, repo) + createChildrenCompletedAgent(t, repo, "agent-1") + setupChildrenCompletedParent(t, ss, "parent-1", "agent-1") + ctx := context.Background() + + insertChildTask(t, ss, "child-1", "parent-1", "COMPLETED") + if _, err := ss.repo.ExecRaw(ctx, ` + INSERT INTO tasks (id, workspace_id, parent_id, state) + VALUES ('child-2', 'ws-1', 'parent-1', NULL) + `); err != nil { + t.Fatalf("insert child-2: %v", err) + } + + var queued int + queue := func(_ string, _ RunContext) { queued++ } + ss.cascadeChildrenCompleted(ctx, &TaskSnapshot{ + ID: "child-1", WorkspaceID: "ws-1", ParentID: "parent-1", + }, queue) + + if queued != 0 { + t.Fatalf("queued wakes = %d, want 0 when the child snapshot is nonterminal", queued) + } +} diff --git a/apps/backend/internal/office/scheduler/reactivity_test.go b/apps/backend/internal/office/scheduler/reactivity_test.go index a0a92dd6a1..9dea4e2c61 100644 --- a/apps/backend/internal/office/scheduler/reactivity_test.go +++ b/apps/backend/internal/office/scheduler/reactivity_test.go @@ -38,7 +38,9 @@ func newReactivityTestRepo(t *testing.T) *officesqlite.Repository { CREATE TABLE IF NOT EXISTS tasks ( id TEXT PRIMARY KEY, workspace_id TEXT DEFAULT '', - workflow_step_id TEXT DEFAULT '' + workflow_step_id TEXT DEFAULT '', + parent_id TEXT DEFAULT '', + state TEXT DEFAULT '' ) `); err != nil { t.Fatalf("create tasks table: %v", err) diff --git a/apps/backend/internal/office/scheduler/run.go b/apps/backend/internal/office/scheduler/run.go index bd99e8a724..48c66c01c7 100644 --- a/apps/backend/internal/office/scheduler/run.go +++ b/apps/backend/internal/office/scheduler/run.go @@ -107,6 +107,18 @@ type RunContext struct { // changes_requested decision so the assignee receiving // task_changes_requested has the context inline. DecisionComment string `json:"decision_comment,omitempty"` + + // IdempotencyKey, when non-empty, overrides the default + // "{reason}:{taskID}:{agentID}" key QueueRunCtx mints. The default + // key is permanently unique per (reason, task, agent) — fine for + // reasons that fire at most once per task, wrong for a reason that + // can legitimately recur (e.g. task_children_completed across + // repeated delegation waves). Callers that recur build their own + // key that changes with the thing that makes each occurrence + // distinct. Excluded from the JSON payload: it must never change + // encodeRunContext's output shape, which CoalesceRun compares for + // equality and taskIDFromPayload parses. + IdempotencyKey string `json:"-"` } // Run status constants. @@ -293,9 +305,10 @@ func (ss *SchedulerService) QueueRun( // QueueRunCtx is the typed variant of QueueRun that takes a // structured RunContext. The context is JSON-encoded into the // payload column so the agent runtime can deserialise it. The -// idempotencyKey is derived as "{reason}:{taskID}:{agentID}" so the -// same agent never gets two runs for the same task+reason within -// the idempotency window. +// idempotency key is c.IdempotencyKey when the caller set one; +// otherwise it defaults to "{reason}:{taskID}:{agentID}" so the same +// agent never gets two runs for the same task+reason within the +// idempotency window. func (ss *SchedulerService) QueueRunCtx( ctx context.Context, agentInstanceID string, c RunContext, ) error { @@ -303,7 +316,10 @@ func (ss *SchedulerService) QueueRunCtx( if err != nil { return fmt.Errorf("encode run context: %w", err) } - idempotencyKey := fmt.Sprintf("%s:%s:%s", c.Reason, c.TaskID, agentInstanceID) + idempotencyKey := c.IdempotencyKey + if idempotencyKey == "" { + idempotencyKey = fmt.Sprintf("%s:%s:%s", c.Reason, c.TaskID, agentInstanceID) + } return ss.QueueRun(ctx, agentInstanceID, c.Reason, payload, idempotencyKey) }