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 <carlosmflorencio@gmail.com>

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

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)
 	}

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)
 	}
 

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,

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)
+}

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)
 	}

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).

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.