From d7e337ff475eedea973ade52830c93bda65fdaa4 Mon Sep 17 00:00:00 2001 From: Henry Date: Thu, 10 Sep 2026 23:34:25 +0800 Subject: [PATCH] fix(backend): stop different tasks' queued runs from silently merging (#3517) Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com> Co-authored-by: Carlos Florencio --- apps/backend/internal/db/dialect/json.go | 18 ++ .../office/repository/sqlite/runs_test.go | 2 +- .../internal/office/service/run_test.go | 2 +- .../internal/runs/repository/sqlite/runs.go | 58 +++-- .../sqlite/runs_queue_postgres_test.go | 220 ++++++++++++++++++ .../runs/repository/sqlite/runs_queue_test.go | 175 +++++++++++++- apps/backend/internal/runs/service/service.go | 14 +- .../tasks/system-design/model-unification.md | 5 +- 8 files changed, 460 insertions(+), 34 deletions(-) create mode 100644 apps/backend/internal/runs/repository/sqlite/runs_queue_postgres_test.go diff --git a/apps/backend/internal/db/dialect/json.go b/apps/backend/internal/db/dialect/json.go index 41c0cc744b..0ddc445e08 100644 --- a/apps/backend/internal/db/dialect/json.go +++ b/apps/backend/internal/db/dialect/json.go @@ -44,6 +44,24 @@ func JSONExtractPath(driver, col string, segments ...string) string { return fmt.Sprintf("json_extract(%s, '$.%s')", col, strings.Join(segments, ".")) } +// JSONTypeIsString returns a WHERE-clause fragment asserting the JSON value +// at path is itself a string. JSONExtract's Postgres ->> operator converts +// any JSON type to text, so a caller that compares its result against a +// bound string parameter can match a non-string stored value (e.g. a JSON +// number) whose text form happens to equal the parameter. SQLite's +// json_extract mostly avoids this by preserving storage class, except that +// a JSON object/array is itself returned as text. Checking the stored +// value's declared JSON type closes both gaps identically. +// +// SQLite: json_type(col, '$.path') = 'text' +// Postgres: jsonb_typeof(col::jsonb->'path') = 'string' +func JSONTypeIsString(driver, col, path string) string { + if IsPostgres(driver) { + return fmt.Sprintf("jsonb_typeof(%s::jsonb->'%s') = 'string'", col, path) + } + return fmt.Sprintf("json_type(%s, '$.%s') = 'text'", col, path) +} + // JSONExtractIsNotNull returns the SQL fragment to check that a JSON path is not null. // // SQLite: json_extract(col, '$.path') IS NOT NULL diff --git a/apps/backend/internal/office/repository/sqlite/runs_test.go b/apps/backend/internal/office/repository/sqlite/runs_test.go index 256d2e45bc..f6a0b1918e 100644 --- a/apps/backend/internal/office/repository/sqlite/runs_test.go +++ b/apps/backend/internal/office/repository/sqlite/runs_test.go @@ -177,7 +177,7 @@ func TestCoalesceRun(t *testing.T) { t.Fatalf("create: %v", err) } - coalesced, err := repo.CoalesceRun(ctx, "a1", "task_comment", 10, `{"task_id":"t2"}`) + coalesced, err := repo.CoalesceRun(ctx, "a1", "task_comment", 10, `{"task_id":"t1"}`) if err != nil { t.Fatalf("coalesce: %v", err) } diff --git a/apps/backend/internal/office/service/run_test.go b/apps/backend/internal/office/service/run_test.go index 192be9117c..16da1241bb 100644 --- a/apps/backend/internal/office/service/run_test.go +++ b/apps/backend/internal/office/service/run_test.go @@ -116,7 +116,7 @@ func TestQueueRun_Coalesce(t *testing.T) { if err := svc.QueueRun(ctx, agent.ID, service.RunReasonTaskComment, `{"task_id":"t1"}`, ""); err != nil { t.Fatalf("first: %v", err) } - if err := svc.QueueRun(ctx, agent.ID, service.RunReasonTaskComment, `{"task_id":"t2"}`, ""); err != nil { + if err := svc.QueueRun(ctx, agent.ID, service.RunReasonTaskComment, `{"task_id":"t1"}`, ""); err != nil { t.Fatalf("second: %v", err) } diff --git a/apps/backend/internal/runs/repository/sqlite/runs.go b/apps/backend/internal/runs/repository/sqlite/runs.go index 9438b5ce1f..5dc73a3b0f 100644 --- a/apps/backend/internal/runs/repository/sqlite/runs.go +++ b/apps/backend/internal/runs/repository/sqlite/runs.go @@ -452,16 +452,34 @@ func (r *Repository) CoalesceRun( ctx context.Context, agentInstanceID, reason string, windowSecs int, payload string, ) (bool, error) { cutoff := time.Now().UTC().Add(-time.Duration(windowSecs) * time.Second) - taskID := taskIDFromPayload(payload) - taskPredicate := "" + taskID, invalidTaskID := taskIDFromPayload(payload) args := []interface{}{payload, agentInstanceID, reason, cutoff, commentkeys.TaskCommentPrefix + "%"} - // Task assignment and failure recovery wakes are task-specific: merging - // two tasks for the same agent would replace the first task's payload and - // silently drop its launch. Other reasons, such as task comments, - // intentionally retain their existing cross-task coalescing behaviour. - if taskID != "" && isTaskScopedCoalescingReason(reason) { - taskPredicate = fmt.Sprintf(" AND %s = ?", dialect.JSONExtract(r.db.DriverName(), "payload", "task_id")) + // A payload's task_id identifies which launch it belongs to: merging + // across two different task_ids (present or absent) would replace one + // launch's payload with an unrelated one and silently drop it. The + // check is symmetric so both directions are covered. json_extract (and + // its Postgres ->> equivalent) yields NULL for both an absent key and + // an explicit JSON null, and '' for a present-but-empty string, so the + // taskless branch coalesces all three shapes together via COALESCE. + jsonExtract := dialect.JSONExtract(r.db.DriverName(), "payload", "task_id") + var taskPredicate string + switch { + case invalidTaskID: + // task_id is present but not a string (e.g. a number): it names a + // task we can't compare textually, so it must not be treated as + // taskless and must not match any queued row at all. + taskPredicate = " AND 1 = 0" + case taskID != "": + // The stored value must itself be a JSON string, not merely equal + // as text: Postgres's ->> converts a stored JSON number (or + // object) to text before the comparison, so an untyped payload + // with e.g. {"task_id":42} could otherwise textually match an + // incoming {"task_id":"42"} and get overwritten. + taskPredicate = fmt.Sprintf(" AND %s AND %s = ?", + dialect.JSONTypeIsString(r.db.DriverName(), "payload", "task_id"), jsonExtract) args = append(args, taskID) + default: + taskPredicate = fmt.Sprintf(" AND COALESCE(%s, '') = ''", jsonExtract) } query := fmt.Sprintf(` UPDATE runs @@ -487,17 +505,25 @@ func (r *Repository) CoalesceRun( return rows > 0, nil } -func isTaskScopedCoalescingReason(reason string) bool { - return reason == "task_assigned" || reason == "manual_resume_after_failure" -} - -func taskIDFromPayload(payload string) string { +// taskIDFromPayload extracts payload.task_id for CoalesceRun's task-scoping +// predicate. invalidTaskID is true only when the key is present with a +// non-string value: that shape names some task_id, just not one comparable +// as a string, so it must be kept out of the taskless bucket (an absent key, +// a JSON null, or a present empty string all return "", invalidTaskID=false). +func taskIDFromPayload(payload string) (taskID string, invalidTaskID bool) { var raw map[string]any if err := json.Unmarshal([]byte(payload), &raw); err != nil { - return "" + return "", false } - taskID, _ := raw["task_id"].(string) - return taskID + v, present := raw["task_id"] + if !present || v == nil { + return "", false + } + s, ok := v.(string) + if !ok { + return "", true + } + return s, false } // ClaimNextEligibleRun atomically claims the next queued run, diff --git a/apps/backend/internal/runs/repository/sqlite/runs_queue_postgres_test.go b/apps/backend/internal/runs/repository/sqlite/runs_queue_postgres_test.go new file mode 100644 index 0000000000..eb92e62032 --- /dev/null +++ b/apps/backend/internal/runs/repository/sqlite/runs_queue_postgres_test.go @@ -0,0 +1,220 @@ +package sqlite_test + +import ( + "context" + "testing" + "time" + + "github.com/kandev/kandev/internal/office/models" + officesqlite "github.com/kandev/kandev/internal/office/repository/sqlite" + runssqlite "github.com/kandev/kandev/internal/runs/repository/sqlite" + taskrepo "github.com/kandev/kandev/internal/task/repository/sqlite" + "github.com/kandev/kandev/internal/testutil" +) + +// newTestRepoPostgres mirrors newTestRepoWithHandles but backs the runs +// repository with an isolated Postgres schema instead of a temp SQLite +// file, following the boot order used elsewhere (task repo schema before +// office repo schema — see failure_postgres_test.go). +func newTestRepoPostgres(t *testing.T) *runssqlite.Repository { + t.Helper() + db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) + if _, err := taskrepo.NewWithDB(db, db, nil); err != nil { + t.Fatalf("init task repo: %v", err) + } + officeRepo, err := officesqlite.NewWithDB(db, db, nil) + if err != nil { + t.Fatalf("init office repo: %v", err) + } + return officeRepo.RunsRepository() +} + +// TestPostgresCoalesceRun_DoesNotMergeDifferentTaskRuns is the PostgreSQL +// twin of TestCoalesceRun_DoesNotMergeDifferentTaskRuns and +// TestCoalesceRun_DoesNotMergeDifferentTaskRuns_ForOtherReasons. +// CoalesceRun's task-scoping predicate is built from +// dialect.JSONExtract, which emits payload::jsonb->>'task_id' on +// Postgres versus json_extract(payload, '$.task_id') on SQLite — two +// different SQL fragments doing the same job, so a passing SQLite +// assertion is not evidence the Postgres fragment even parses, let alone +// scopes correctly. Runs the predicate for both the original +// "task_assigned" reason and a second reactivity reason, since the fix +// (d79cec9f3) generalized the guard from one hardcoded reason to any +// task-scoped payload. Skips unless KANDEV_TEST_POSTGRES_DSN is set. +func TestPostgresCoalesceRun_DoesNotMergeDifferentTaskRuns(t *testing.T) { + repo := newTestRepoPostgres(t) + ctx := context.Background() + + for _, reason := range []string{"task_assigned", "task_blockers_resolved"} { + t.Run(reason, func(t *testing.T) { + queued := mustCreateRun(t, repo, &models.Run{ + ID: "pg-task-a-" + reason, AgentProfileID: "a1", Reason: reason, + Payload: `{"task_id":"pg-task-a"}`, Status: "queued", CoalescedCount: 1, + }) + setRequestedAt(t, repo, queued.ID, time.Now().UTC()) + + merged, err := repo.CoalesceRun(ctx, "a1", reason, 3600, `{"task_id":"pg-task-b"}`) + if err != nil { + t.Fatalf("coalesce: %v", err) + } + if merged { + t.Fatal("coalesce = true for a different task, want false") + } + + got := mustGetRun(t, repo, queued.ID) + checkInt(t, "coalesced_count", got.CoalescedCount, 1) + checkString(t, "payload", got.Payload, `{"task_id":"pg-task-a"}`) + }) + } +} + +// TestPostgresCoalesceRun_DoesNotMergeTasklessIntoTaskCarryingRun is the +// PostgreSQL twin of TestCoalesceRun_DoesNotMergeTasklessIntoTaskCarryingRun: +// the guard must also reject a taskless incoming payload against a +// task-carrying queued row, not just the reverse. +func TestPostgresCoalesceRun_DoesNotMergeTasklessIntoTaskCarryingRun(t *testing.T) { + repo := newTestRepoPostgres(t) + ctx := context.Background() + + queued := mustCreateRun(t, repo, &models.Run{ + ID: "pg-task-c", AgentProfileID: "a1", Reason: "task_assigned", + Payload: `{"task_id":"pg-task-c"}`, Status: "queued", CoalescedCount: 1, + }) + setRequestedAt(t, repo, queued.ID, time.Now().UTC()) + + merged, err := repo.CoalesceRun(ctx, "a1", "task_assigned", 3600, `{}`) + if err != nil { + t.Fatalf("coalesce: %v", err) + } + if merged { + t.Fatal("coalesce = true for a taskless payload into a task-carrying run, want false") + } + + got := mustGetRun(t, repo, queued.ID) + checkInt(t, "coalesced_count", got.CoalescedCount, 1) + checkString(t, "payload", got.Payload, `{"task_id":"pg-task-c"}`) +} + +// TestPostgresCoalesceRun_MergesTasklessIntoTasklessRun is the Postgres +// twin of TestCoalesceRun_MergesTasklessIntoTasklessRun: the ->>'task_id' +// IS NULL fragment must still merge a taskless incoming payload into a +// taskless queued row, not just reject a task-carrying one. Untested, +// this fragment could be silently rewritten to always-false (e.g. an +// extra AND clause) and every other Postgres coalesce test would still +// pass, since none of them assert a taskless merge succeeds. +func TestPostgresCoalesceRun_MergesTasklessIntoTasklessRun(t *testing.T) { + for _, tc := range []struct { + name string + payload string + }{ + {"empty object", `{}`}, + {"other field set", `{"routine_id":"r1"}`}, + {"explicit null task_id", `{"task_id":null}`}, + {"present empty string task_id", `{"task_id":""}`}, + } { + t.Run(tc.name, func(t *testing.T) { + repo := newTestRepoPostgres(t) + ctx := context.Background() + + queued := mustCreateRun(t, repo, &models.Run{ + ID: "pg-taskless", AgentProfileID: "a1", Reason: "heartbeat", + Payload: tc.payload, Status: "queued", CoalescedCount: 1, + }) + setRequestedAt(t, repo, queued.ID, time.Now().UTC()) + + merged, err := repo.CoalesceRun(ctx, "a1", "heartbeat", 3600, `{"merged":true}`) + if err != nil { + t.Fatalf("coalesce: %v", err) + } + if !merged { + t.Fatalf("coalesce = false for taskless into taskless, want true") + } + + got := mustGetRun(t, repo, queued.ID) + checkInt(t, "coalesced_count", got.CoalescedCount, 2) + checkString(t, "payload", got.Payload, `{"merged":true}`) + }) + } +} + +// TestPostgresCoalesceRun_DoesNotMergeNonStringTaskIDIntoTasklessRun is the +// Postgres twin of TestCoalesceRun_DoesNotMergeNonStringTaskIDIntoTasklessRun: +// a payload with a present, non-string task_id must not be classified +// taskless and must not absorb a genuinely taskless queued run's payload. +func TestPostgresCoalesceRun_DoesNotMergeNonStringTaskIDIntoTasklessRun(t *testing.T) { + repo := newTestRepoPostgres(t) + ctx := context.Background() + + queued := mustCreateRun(t, repo, &models.Run{ + ID: "pg-taskless-real", AgentProfileID: "a1", Reason: "custom_reason", + Payload: `{"note":"real taskless launch"}`, Status: "queued", CoalescedCount: 1, + }) + setRequestedAt(t, repo, queued.ID, time.Now().UTC()) + + merged, err := repo.CoalesceRun(ctx, "a1", "custom_reason", 3600, `{"task_id":42}`) + if err != nil { + t.Fatalf("coalesce: %v", err) + } + if merged { + t.Fatal("coalesce = true for a non-string task_id into a taskless run, want false") + } + + got := mustGetRun(t, repo, queued.ID) + checkInt(t, "coalesced_count", got.CoalescedCount, 1) + checkString(t, "payload", got.Payload, `{"note":"real taskless launch"}`) +} + +// TestPostgresCoalesceRun_DoesNotMergeIntoNonStringQueuedTaskID is the +// Postgres twin of TestCoalesceRun_DoesNotMergeIntoNonStringQueuedTaskID. +// Postgres's ->> operator converts a stored JSON number to text before the +// comparison, so a naive text-only predicate would let {"task_id":"42"} +// match a queued {"task_id":42} and silently overwrite it; this exercises +// exactly that dialect-specific coercion. +func TestPostgresCoalesceRun_DoesNotMergeIntoNonStringQueuedTaskID(t *testing.T) { + repo := newTestRepoPostgres(t) + ctx := context.Background() + + queued := mustCreateRun(t, repo, &models.Run{ + ID: "pg-malformed", AgentProfileID: "a1", Reason: "task_assigned", + Payload: `{"task_id":42}`, Status: "queued", CoalescedCount: 1, + }) + setRequestedAt(t, repo, queued.ID, time.Now().UTC()) + + merged, err := repo.CoalesceRun(ctx, "a1", "task_assigned", 3600, `{"task_id":"42"}`) + if err != nil { + t.Fatalf("coalesce: %v", err) + } + if merged { + t.Fatal("coalesce = true into a queued row with a non-string task_id, want false") + } + + got := mustGetRun(t, repo, queued.ID) + checkInt(t, "coalesced_count", got.CoalescedCount, 1) + checkString(t, "payload", got.Payload, `{"task_id":42}`) +} + +// TestPostgresCoalesceRun_MergesSameTask is the positive-path twin: a +// same-task, same-agent, same-reason request still merges, proving the +// Postgres JSONExtract fragment is not just "always false" (which would +// also make the negative test above pass for the wrong reason). +func TestPostgresCoalesceRun_MergesSameTask(t *testing.T) { + repo := newTestRepoPostgres(t) + ctx := context.Background() + + queued := mustCreateRun(t, repo, &models.Run{ + ID: "pg-merge-task", AgentProfileID: "a1", Reason: "task_assigned", + Payload: `{"task_id":"pg-merge-task"}`, Status: "queued", CoalescedCount: 1, + }) + setRequestedAt(t, repo, queued.ID, time.Now().UTC()) + + merged, err := repo.CoalesceRun(ctx, "a1", "task_assigned", 3600, `{"task_id":"pg-merge-task"}`) + if err != nil { + t.Fatalf("coalesce: %v", err) + } + if !merged { + t.Fatal("coalesce = false for the same task, want true") + } + + got := mustGetRun(t, repo, queued.ID) + checkInt(t, "coalesced_count", got.CoalescedCount, 2) +} diff --git a/apps/backend/internal/runs/repository/sqlite/runs_queue_test.go b/apps/backend/internal/runs/repository/sqlite/runs_queue_test.go index 86250445b0..f24c297b10 100644 --- a/apps/backend/internal/runs/repository/sqlite/runs_queue_test.go +++ b/apps/backend/internal/runs/repository/sqlite/runs_queue_test.go @@ -29,10 +29,19 @@ func TestCoalesceRun_MergesIntoMostRecentQueuedRun(t *testing.T) { ctx := context.Background() now := time.Now().UTC() - older := queueRunAt(t, repo, "older", "a1", now.Add(-30*time.Second)) - newer := queueRunAt(t, repo, "newer", "a1", now.Add(-10*time.Second)) + older := mustCreateRun(t, repo, &models.Run{ + ID: "older", AgentProfileID: "a1", Reason: "task_assigned", + Payload: `{"task_id":"t1"}`, Status: "queued", CoalescedCount: 1, + }) + setRequestedAt(t, repo, older.ID, now.Add(-30*time.Second)) - merged, err := repo.CoalesceRun(ctx, "a1", "task_assigned", 3600, `{"merged":true}`) + newer := mustCreateRun(t, repo, &models.Run{ + ID: "newer", AgentProfileID: "a1", Reason: "task_assigned", + Payload: `{"task_id":"t1"}`, Status: "queued", CoalescedCount: 1, + }) + setRequestedAt(t, repo, newer.ID, now.Add(-10*time.Second)) + + merged, err := repo.CoalesceRun(ctx, "a1", "task_assigned", 3600, `{"task_id":"t1","merged":true}`) if err != nil { t.Fatalf("coalesce: %v", err) } @@ -42,11 +51,11 @@ func TestCoalesceRun_MergesIntoMostRecentQueuedRun(t *testing.T) { got := mustGetRun(t, repo, newer.ID) checkInt(t, "coalesced_count", got.CoalescedCount, 2) - checkString(t, "payload", got.Payload, `{"merged":true}`) + checkString(t, "payload", got.Payload, `{"task_id":"t1","merged":true}`) untouched := mustGetRun(t, repo, older.ID) checkInt(t, "older coalesced_count", untouched.CoalescedCount, 1) - checkString(t, "older payload", untouched.Payload, `{"task_id":"older"}`) + checkString(t, "older payload", untouched.Payload, `{"task_id":"t1"}`) if n := countRuns(t, repo); n != 2 { t.Errorf("%d rows after coalescing, want 2 (no new run may be created)", n) @@ -98,6 +107,34 @@ func TestCoalesceRun_ManualResumeAfterFailure_DoesNotMergeDifferentTaskRuns(t *t checkString(t, "payload", got.Payload, `{"task_id":"task-a"}`) } +// TestCoalesceRun_DoesNotMergeDifferentTaskRuns_ForOtherReasons pins the +// same guarantee as TestCoalesceRun_DoesNotMergeDifferentTaskRuns for a +// task-scoped reactivity reason other than "task_assigned" — e.g. two +// blocked tasks unblocked by the same predecessor and assigned to the +// same agent within the coalesce window must not merge into one run. +func TestCoalesceRun_DoesNotMergeDifferentTaskRuns_ForOtherReasons(t *testing.T) { + repo := newTestRepo(t) + ctx := context.Background() + + queued := mustCreateRun(t, repo, &models.Run{ + ID: "task-a-run", AgentProfileID: "a1", Reason: "task_blockers_resolved", + Payload: `{"task_id":"task-a"}`, Status: "queued", CoalescedCount: 1, + }) + setRequestedAt(t, repo, queued.ID, time.Now().UTC()) + + merged, err := repo.CoalesceRun(ctx, "a1", "task_blockers_resolved", 3600, `{"task_id":"task-b"}`) + if err != nil { + t.Fatalf("coalesce: %v", err) + } + if merged { + t.Fatal("coalesce = true for a different task, want false") + } + + got := mustGetRun(t, repo, queued.ID) + checkInt(t, "coalesced_count", got.CoalescedCount, 1) + checkString(t, "payload", got.Payload, `{"task_id":"task-a"}`) +} + // TestCoalesceRun_ManualResumeAfterFailure_MergesSameTaskDuplicates proves // the task-scoping predicate added for manual_resume_after_failure still // allows genuine duplicates (repeated wakes for the same task) to coalesce @@ -126,6 +163,126 @@ func TestCoalesceRun_ManualResumeAfterFailure_MergesSameTaskDuplicates(t *testin } } +// TestCoalesceRun_DoesNotMergeTasklessIntoTaskCarryingRun pins the other +// direction of the same guarantee: a taskless incoming payload must not +// absorb a queued run that carries a task_id, or it would silently +// overwrite that task's launch. +func TestCoalesceRun_DoesNotMergeTasklessIntoTaskCarryingRun(t *testing.T) { + repo := newTestRepo(t) + ctx := context.Background() + queued := queueRunAt(t, repo, "task-a", "a1", time.Now().UTC()) + + merged, err := repo.CoalesceRun(ctx, "a1", "task_assigned", 3600, `{}`) + if err != nil { + t.Fatalf("coalesce: %v", err) + } + if merged { + t.Fatal("coalesce = true for a taskless payload into a task-carrying run, want false") + } + + got := mustGetRun(t, repo, queued.ID) + checkInt(t, "coalesced_count", got.CoalescedCount, 1) + checkString(t, "payload", got.Payload, `{"task_id":"task-a"}`) +} + +// TestCoalesceRun_MergesTasklessIntoTasklessRun pins the positive half of +// the taskless guarantee that TestCoalesceRun_DoesNotMergeTasklessIntoTaskCarryingRun +// only pins negatively: a taskless incoming payload must still merge into a +// taskless queued run, so heartbeat, routine_dispatch_* and agent_error wakes +// keep coalescing instead of growing the queue unboundedly. +func TestCoalesceRun_MergesTasklessIntoTasklessRun(t *testing.T) { + for _, tc := range []struct { + name string + payload string + }{ + {"empty object", `{}`}, + {"other field set", `{"routine_id":"r1"}`}, + {"explicit null task_id", `{"task_id":null}`}, + {"present empty string task_id", `{"task_id":""}`}, + } { + t.Run(tc.name, func(t *testing.T) { + repo := newTestRepo(t) + ctx := context.Background() + + queued := mustCreateRun(t, repo, &models.Run{ + ID: "taskless", AgentProfileID: "a1", Reason: "heartbeat", + Payload: tc.payload, Status: "queued", CoalescedCount: 1, + }) + setRequestedAt(t, repo, queued.ID, time.Now().UTC()) + + merged, err := repo.CoalesceRun(ctx, "a1", "heartbeat", 3600, `{"merged":true}`) + if err != nil { + t.Fatalf("coalesce: %v", err) + } + if !merged { + t.Fatalf("coalesce = false for taskless into taskless, want true") + } + + got := mustGetRun(t, repo, queued.ID) + checkInt(t, "coalesced_count", got.CoalescedCount, 2) + checkString(t, "payload", got.Payload, `{"merged":true}`) + }) + } +} + +// TestCoalesceRun_DoesNotMergeNonStringTaskIDIntoTasklessRun pins the R4-2 +// fix: a payload whose task_id is present but not a string (e.g. a number) +// names some task, just not one taskIDFromPayload can compare textually. It +// must not be classified taskless and must not silently absorb — and +// overwrite the payload of — a genuinely taskless queued run. +func TestCoalesceRun_DoesNotMergeNonStringTaskIDIntoTasklessRun(t *testing.T) { + repo := newTestRepo(t) + ctx := context.Background() + + queued := mustCreateRun(t, repo, &models.Run{ + ID: "taskless", AgentProfileID: "a1", Reason: "custom_reason", + Payload: `{"note":"real taskless launch"}`, Status: "queued", CoalescedCount: 1, + }) + setRequestedAt(t, repo, queued.ID, time.Now().UTC()) + + merged, err := repo.CoalesceRun(ctx, "a1", "custom_reason", 3600, `{"task_id":42}`) + if err != nil { + t.Fatalf("coalesce: %v", err) + } + if merged { + t.Fatal("coalesce = true for a non-string task_id into a taskless run, want false") + } + + got := mustGetRun(t, repo, queued.ID) + checkInt(t, "coalesced_count", got.CoalescedCount, 1) + checkString(t, "payload", got.Payload, `{"note":"real taskless launch"}`) +} + +// TestCoalesceRun_DoesNotMergeIntoNonStringQueuedTaskID pins the other +// direction of TestCoalesceRun_DoesNotMergeNonStringTaskIDIntoTasklessRun: a +// queued row whose own task_id is malformed (present but not a string) must +// not absorb an incoming request just because the malformed value's textual +// form happens to equal the incoming task_id's string. Both dialects yield a +// text-comparable value for a JSON number, so this predicate must inspect +// the stored value's JSON type rather than only its text. +func TestCoalesceRun_DoesNotMergeIntoNonStringQueuedTaskID(t *testing.T) { + repo := newTestRepo(t) + ctx := context.Background() + + queued := mustCreateRun(t, repo, &models.Run{ + ID: "malformed", AgentProfileID: "a1", Reason: "task_assigned", + Payload: `{"task_id":42}`, Status: "queued", CoalescedCount: 1, + }) + setRequestedAt(t, repo, queued.ID, time.Now().UTC()) + + merged, err := repo.CoalesceRun(ctx, "a1", "task_assigned", 3600, `{"task_id":"42"}`) + if err != nil { + t.Fatalf("coalesce: %v", err) + } + if merged { + t.Fatal("coalesce = true into a queued row with a non-string task_id, want false") + } + + got := mustGetRun(t, repo, queued.ID) + checkInt(t, "coalesced_count", got.CoalescedCount, 1) + checkString(t, "payload", got.Payload, `{"task_id":42}`) +} + // TestCoalesceRun_RepeatedCoalesceAccumulatesOnOneRow pins the counter // arithmetic across several merges into the same run. func TestCoalesceRun_RepeatedCoalesceAccumulatesOnOneRow(t *testing.T) { @@ -134,7 +291,7 @@ func TestCoalesceRun_RepeatedCoalesceAccumulatesOnOneRow(t *testing.T) { run := queueRunAt(t, repo, "target", "a1", time.Now().UTC().Add(-time.Second)) for i := range 3 { - merged, err := repo.CoalesceRun(ctx, "a1", "task_assigned", 3600, `{"n":1}`) + merged, err := repo.CoalesceRun(ctx, "a1", "task_assigned", 3600, `{"task_id":"target","n":1}`) if err != nil { t.Fatalf("coalesce %d: %v", i, err) } @@ -183,7 +340,7 @@ func TestCoalesceRun_HonoursTheWindow(t *testing.T) { ctx := context.Background() run := queueRunAt(t, repo, "stale", "a1", time.Now().UTC().Add(-10*time.Second)) - merged, err := repo.CoalesceRun(ctx, "a1", "task_assigned", 5, `{"merged":true}`) + merged, err := repo.CoalesceRun(ctx, "a1", "task_assigned", 5, `{"task_id":"stale","merged":true}`) if err != nil { t.Fatalf("coalesce (narrow window): %v", err) } @@ -192,7 +349,7 @@ func TestCoalesceRun_HonoursTheWindow(t *testing.T) { } checkInt(t, "coalesced_count", mustGetRun(t, repo, run.ID).CoalescedCount, 1) - merged, err = repo.CoalesceRun(ctx, "a1", "task_assigned", 600, `{"merged":true}`) + merged, err = repo.CoalesceRun(ctx, "a1", "task_assigned", 600, `{"task_id":"stale","merged":true}`) if err != nil { t.Fatalf("coalesce (wide window): %v", err) } @@ -223,7 +380,7 @@ func TestCoalesceRun_SkipsCommentKeyedRuns(t *testing.T) { }) setRequestedAt(t, repo, keyed.ID, now.Add(-5*time.Second)) - merged, err := repo.CoalesceRun(ctx, "a1", "task_comment", 600, `{"merged":true}`) + merged, err := repo.CoalesceRun(ctx, "a1", "task_comment", 600, `{"task_id":"t1","merged":true}`) if err != nil { t.Fatalf("coalesce: %v", err) } diff --git a/apps/backend/internal/runs/service/service.go b/apps/backend/internal/runs/service/service.go index 7f83aaac1d..ea0c42e449 100644 --- a/apps/backend/internal/runs/service/service.go +++ b/apps/backend/internal/runs/service/service.go @@ -60,8 +60,8 @@ const ( // already exists in the durable idempotency index, so nothing was inserted. QueueOutcomeDeduped QueueOutcome = "deduped" // QueueOutcomeCoalesced means the request was merged into an existing - // queued row for the same agent + reason within the coalescing - // window, so nothing new was inserted. + // queued row for the same agent, reason, and task bucket within the + // coalescing window, so nothing new was inserted. QueueOutcomeCoalesced QueueOutcome = "coalesced" ) @@ -92,9 +92,11 @@ type QueueRunRequest struct { } // CoalesceWindowSeconds is the default coalescing window. When two -// queue_run requests for the same (agent, reason) land within this -// window, the second is merged into the first by bumping -// coalesced_count and replacing the payload. +// queue_run requests for the same agent, reason, and task bucket land +// within this window, the second is merged into the first by bumping +// coalesced_count and replacing the payload. A task bucket is the +// nonempty task_id, or the taskless bucket when task_id is missing, +// null, or empty. const CoalesceWindowSeconds = 5 // IdempotencyWindowHours is the lookback used by the fast duplicate query. @@ -167,7 +169,7 @@ func (s *Service) SubscribeSignal() <-chan struct{} { return s.signalCh } // 1. Resolve agent_profile_id (from the request field, payload fallback, // or a wired resolver). // 2. Recent idempotency check on req.IdempotencyKey if set. -// 3. Coalescing (5s window for same agent + reason). +// 3. Coalescing (5s window for same agent, reason, and task bucket). // 4. Insert into runs table. // 5. Publish OfficeRunQueued. // 6. Signal the scheduler (B3.5 — event-driven claim). diff --git a/docs/specs/tasks/system-design/model-unification.md b/docs/specs/tasks/system-design/model-unification.md index 194f875043..f97e57c3db 100644 --- a/docs/specs/tasks/system-design/model-unification.md +++ b/docs/specs/tasks/system-design/model-unification.md @@ -184,12 +184,15 @@ A reviewer rejecting a Review: `office_runs` is renamed `runs`. Universal queue for engine-emitted launches. Every `queue_run` action creates a row. Coalescing (5s -window for same agent + reason), idempotency (a 24h lookup plus a durable +window for same agent, reason, and task bucket), idempotency (a 24h lookup plus a durable unique key), per-agent serialisation (one claimed run per agent), cooldown (per-agent, default 10s), and atomic task checkout (one agent per task at a time) all apply — same machinery as today's office_runs. +A task bucket is the nonempty `task_id`. Missing, null, or empty +`task_id` values use the taskless bucket. + User-initiated launches (clicking Start on a kanban task) bypass the queue and call `runtime.Launch` directly — those don't need coalescing and shouldn't pay the scheduler's tick latency.