test(runs): cover the runs queue sqlite repository (#2628)

Co-authored-by: Kandev Agent <agent@kandev.dev>

apps/backend/internal/runs/repository/sqlite/base_test.go

@@ -0,0 +1,231 @@
+// Package sqlite_test covers the runs queue repository.
+//
+// The runs / run_events tables are created by the office repository's
+// initSchema (see internal/office/repository/sqlite/base.go), so the
+// harness builds an office repository and takes its embedded runs
+// handle rather than hand-rolling the DDL — that keeps these tests
+// honest about the schema production actually runs against.
+package sqlite_test
+
+import (
+	"context"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/jmoiron/sqlx"
+
+	settingsstore "github.com/kandev/kandev/internal/agent/settings/store"
+	"github.com/kandev/kandev/internal/db"
+	"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"
+)
+
+// newTestRepo returns a runs repository backed by a temp-dir SQLite
+// database with the production writer/reader handle split.
+func newTestRepo(t *testing.T) *runssqlite.Repository {
+	t.Helper()
+	repo, _, _ := newTestRepoWithHandles(t)
+	return repo
+}
+
+// newTestRepoWithDB additionally hands back the writer handle so tests
+// that need to seed foreign tables (agent_profiles, office_cost_events)
+// can insert rows the runs repository has no typed method for.
+func newTestRepoWithDB(t *testing.T) (*runssqlite.Repository, *sqlx.DB) {
+	t.Helper()
+	repo, writer, _ := newTestRepoWithHandles(t)
+	return repo, writer
+}
+
+// newTestRepoWithHandles builds the repository over two genuinely
+// distinct pools — the single-connection writer and the multi-
+// connection reader — exactly as internal/persistence wires them.
+// Keeping them distinct is what makes the writer/reader split
+// testable: closing one pool disables every method that uses it.
+func newTestRepoWithHandles(t *testing.T) (*runssqlite.Repository, *sqlx.DB, *sqlx.DB) {
+	t.Helper()
+	dbPath := filepath.Join(t.TempDir(), "runs.db")
+
+	writerRaw, err := db.OpenSQLite(dbPath)
+	if err != nil {
+		t.Fatalf("open writer: %v", err)
+	}
+	writer := sqlx.NewDb(writerRaw, "sqlite3")
+	t.Cleanup(func() { _ = writer.Close() })
+
+	readerRaw, err := db.OpenSQLiteReader(dbPath)
+	if err != nil {
+		t.Fatalf("open reader: %v", err)
+	}
+	reader := sqlx.NewDb(readerRaw, "sqlite3")
+	t.Cleanup(func() { _ = reader.Close() })
+
+	// agent_profiles lives in the settings store; ListRuns joins it.
+	if _, _, err := settingsstore.Provide(writer, reader, nil); err != nil {
+		t.Fatalf("settings store init: %v", err)
+	}
+	officeRepo, err := officesqlite.NewWithDB(writer, reader, nil)
+	if err != nil {
+		t.Fatalf("office repo init: %v", err)
+	}
+	return officeRepo.RunsRepository(), writer, reader
+}
+
+// seedAgentProfile inserts the minimal agent_profiles row ListRuns
+// needs to resolve a run's workspace. agent_profiles.agent_id is a
+// foreign key into agents (and the harness DSN enables FK enforcement),
+// so each profile gets its own parent agent row in the same workspace —
+// sharing one agent row across profiles would leave it pointing at
+// whichever workspace was seeded first.
+func seedAgentProfile(t *testing.T, writer *sqlx.DB, id, workspaceID string) {
+	t.Helper()
+	now := time.Now().UTC()
+	if _, err := writer.Exec(`INSERT INTO agents
+		(id, name, workspace_id, created_at, updated_at)
+		VALUES (?, ?, ?, ?, ?)`, id, "agent-"+id, workspaceID, now, now); err != nil {
+		t.Fatalf("seed agent %q: %v", id, err)
+	}
+	_, err := writer.Exec(`INSERT INTO agent_profiles
+		(id, agent_id, name, agent_display_name, workspace_id, role,
+		 created_at, updated_at)
+		VALUES (?, ?, ?, ?, ?, '', ?, ?)`,
+		id, id, "name", "display", workspaceID, now, now)
+	if err != nil {
+		t.Fatalf("seed agent_profile %q: %v", id, err)
+	}
+}
+
+// seedCostEvent inserts one office_cost_events row for GetRunWithCosts.
+func seedCostEvent(
+	t *testing.T, writer *sqlx.DB, id, taskID string,
+	tokensIn, tokensCachedIn, tokensOut, costSubcents int64,
+) {
+	t.Helper()
+	now := time.Now().UTC()
+	_, err := writer.Exec(`INSERT INTO office_cost_events
+		(id, session_id, task_id, agent_profile_id, project_id, model, provider,
+		 tokens_in, tokens_cached_in, tokens_out, cost_subcents, estimated,
+		 occurred_at, created_at)
+		VALUES (?, '', ?, 'a1', '', 'model', 'provider', ?, ?, ?, ?, 0, ?, ?)`,
+		id, taskID, tokensIn, tokensCachedIn, tokensOut, costSubcents, now, now)
+	if err != nil {
+		t.Fatalf("seed cost event %q: %v", id, err)
+	}
+}
+
+// mustCreateRun creates a run and fails the test on error.
+func mustCreateRun(t *testing.T, repo *runssqlite.Repository, run *models.Run) *models.Run {
+	t.Helper()
+	if err := repo.CreateRun(context.Background(), run); err != nil {
+		t.Fatalf("create run: %v", err)
+	}
+	return run
+}
+
+// mustGetRun reads a run back and fails the test when it is missing.
+func mustGetRun(t *testing.T, repo *runssqlite.Repository, id string) *models.Run {
+	t.Helper()
+	run, err := repo.GetRunByID(context.Background(), id)
+	if err != nil {
+		t.Fatalf("get run %q: %v", id, err)
+	}
+	return run
+}
+
+// setRequestedAt backdates a run so ordering assertions do not depend
+// on wall-clock resolution between two CreateRun calls.
+func setRequestedAt(t *testing.T, repo *runssqlite.Repository, id string, ts time.Time) {
+	t.Helper()
+	if err := repo.SetRunRequestedAtForTest(context.Background(), id, ts); err != nil {
+		t.Fatalf("set requested_at for %q: %v", id, err)
+	}
+}
+
+// setStatus forces status + timing fields on a seeded run.
+func setStatus(
+	t *testing.T, repo *runssqlite.Repository, id, status string,
+	claimedAt, finishedAt *time.Time,
+) {
+	t.Helper()
+	if err := repo.SetRunStatusForTest(context.Background(), id, status, claimedAt, finishedAt); err != nil {
+		t.Fatalf("set status for %q: %v", id, err)
+	}
+}
+
+func strPtr(s string) *string { return &s }
+
+func timePtr(ts time.Time) *time.Time { return &ts }
+
+// TestWriterAndReaderExposeDistinctHandles pins the accessors used by
+// callers that join runs against their own tables: they must hand back
+// the two handles NewWithDB was given, not the same one twice.
+func TestWriterAndReaderExposeDistinctHandles(t *testing.T) {
+	repo, writer, reader := newTestRepoWithHandles(t)
+
+	if repo.Writer() != writer {
+		t.Errorf("Writer() returned a different handle than the one passed to NewWithDB")
+	}
+	if repo.Reader() != reader {
+		t.Errorf("Reader() returned a different handle than the one passed to NewWithDB")
+	}
+	if repo.Writer() == repo.Reader() {
+		t.Fatalf("Writer() and Reader() returned the same handle")
+	}
+
+	if _, err := repo.Writer().Exec(
+		`INSERT INTO runs (id, agent_profile_id, reason, requested_at) VALUES ('w1','a1','r',?)`,
+		time.Now().UTC(),
+	); err != nil {
+		t.Fatalf("writer handle should accept writes: %v", err)
+	}
+	var seen int
+	if err := repo.Reader().Get(&seen, `SELECT COUNT(*) FROM runs WHERE id = 'w1'`); err != nil {
+		t.Fatalf("read through the reader handle: %v", err)
+	}
+	if seen != 1 {
+		t.Errorf("reader saw %d rows for the writer's insert, want 1", seen)
+	}
+}
+
+// TestReadPathsUseTheReadOnlyHandle is the behavioural half of the
+// writer/reader split: every read method must go through r.ro, and
+// every write through r.db. A write routed to the read-only reader
+// would fail outright, and a read that still works after the writer
+// pool is closed proves it did not use r.db.
+func TestReadPathsUseTheReadOnlyHandle(t *testing.T) {
+	repo, writer, _ := newTestRepoWithHandles(t)
+	ctx := context.Background()
+
+	run := mustCreateRun(t, repo, &models.Run{
+		AgentProfileID: "a1",
+		Reason:         "task_assigned",
+		Payload:        `{"task_id":"t1"}`,
+		Status:         "queued",
+		CoalescedCount: 1,
+	})
+
+	if err := writer.Close(); err != nil {
+		t.Fatalf("close writer: %v", err)
+	}
+
+	got, err := repo.GetRunByID(ctx, run.ID)
+	if err != nil {
+		t.Fatalf("GetRunByID after closing the writer: %v", err)
+	}
+	if got.ID != run.ID {
+		t.Errorf("GetRunByID returned %q, want %q", got.ID, run.ID)
+	}
+	if _, err := repo.ListPendingRunsForTask(ctx, "t1"); err != nil {
+		t.Errorf("ListPendingRunsForTask after closing the writer: %v", err)
+	}
+	if _, err := repo.ListRunEvents(ctx, run.ID, -1, 0); err != nil {
+		t.Errorf("ListRunEvents after closing the writer: %v", err)
+	}
+
+	// Writes must not fall back to the reader.
+	if err := repo.FinishRun(ctx, run.ID, "finished"); err == nil {
+		t.Errorf("FinishRun succeeded with a closed writer; the write path is not using r.db")
+	}
+}

apps/backend/internal/runs/repository/sqlite/cancel_test.go

@@ -0,0 +1,186 @@
+package sqlite_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/kandev/kandev/internal/office/models"
+)
+
+// TestCancelRun_CancelsQueuedAndClaimedRuns pins every column the
+// terminal cancel write owns, for both cancellable states.
+func TestCancelRun_CancelsQueuedAndClaimedRuns(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	now := time.Now().UTC()
+	before := now.Add(-time.Second)
+
+	queued := queueRunAt(t, repo, "queued", "a1", now)
+	claimed := queueRunAt(t, repo, "claimed", "a2", now)
+	setStatus(t, repo, claimed.ID, "claimed", timePtr(now), nil)
+
+	for _, id := range []string{queued.ID, claimed.ID} {
+		if err := repo.CancelRun(ctx, id, "user stopped the task"); err != nil {
+			t.Fatalf("cancel %q: %v", id, err)
+		}
+		got := mustGetRun(t, repo, id)
+		checkString(t, id+" status", string(got.Status), "cancelled")
+		checkStringPtr(t, id+" cancel_reason", got.CancelReason, strPtr("user stopped the task"))
+		if got.FinishedAt == nil {
+			t.Errorf("%s finished_at = nil, want a stamp", id)
+		} else if !got.FinishedAt.After(before) {
+			t.Errorf("%s finished_at = %s, want after %s", id, got.FinishedAt, before)
+		}
+	}
+}
+
+// TestCancelRun_TerminalRunIsLeftAlone is the guard that keeps a run
+// which completed between the caller's SELECT and this UPDATE from
+// having its real outcome overwritten.
+func TestCancelRun_TerminalRunIsLeftAlone(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	finishedAt := time.Date(2026, 2, 3, 4, 5, 6, 0, time.UTC)
+
+	for _, status := range []string{"finished", "failed", "cancelled"} {
+		run := queueRunAt(t, repo, "run-"+status, "a1", finishedAt.Add(-time.Hour))
+		setStatus(t, repo, run.ID, status, nil, timePtr(finishedAt))
+
+		if err := repo.CancelRun(ctx, run.ID, "too late"); err != nil {
+			t.Fatalf("cancel %q: %v", run.ID, err)
+		}
+		got := mustGetRun(t, repo, run.ID)
+		checkString(t, run.ID+" status", string(got.Status), status)
+		if got.CancelReason != nil {
+			t.Errorf("%s cancel_reason = %q, want nil", run.ID, *got.CancelReason)
+		}
+		if got.FinishedAt == nil || !got.FinishedAt.Equal(finishedAt) {
+			t.Errorf("%s finished_at = %v, want the original %s", run.ID, got.FinishedAt, finishedAt)
+		}
+	}
+}
+
+// TestCancelRun_UnknownRunIsANoOp pins that a missing id is not an error.
+func TestCancelRun_UnknownRunIsANoOp(t *testing.T) {
+	repo := newTestRepo(t)
+	if err := repo.CancelRun(context.Background(), "nope", "reason"); err != nil {
+		t.Fatalf("cancel unknown run: %v", err)
+	}
+}
+
+// TestCancelRunsWhere_ReportsRowsActuallyCancelled pins the returned
+// count against a mixed set: only the queued/claimed rows count.
+func TestCancelRunsWhere_ReportsRowsActuallyCancelled(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	now := time.Now().UTC()
+
+	queueRunAt(t, repo, "q1", "a1", now)
+	claimed := queueRunAt(t, repo, "c1", "a1", now)
+	setStatus(t, repo, claimed.ID, "claimed", timePtr(now), nil)
+	finished := queueRunAt(t, repo, "f1", "a1", now)
+	setStatus(t, repo, finished.ID, "finished", nil, timePtr(now))
+	queueRunAt(t, repo, "other-agent", "a2", now)
+
+	cancelled, err := repo.CancelRunsWhere(ctx, "agent removed", `agent_profile_id = ?`, "a1")
+	if err != nil {
+		t.Fatalf("cancel where: %v", err)
+	}
+	if cancelled != 2 {
+		t.Errorf("cancelled = %d, want 2", cancelled)
+	}
+	checkString(t, "q1", string(mustGetRun(t, repo, "q1").Status), "cancelled")
+	checkString(t, "c1", string(mustGetRun(t, repo, "c1").Status), "cancelled")
+	checkString(t, "f1", string(mustGetRun(t, repo, "f1").Status), "finished")
+	checkString(t, "other-agent", string(mustGetRun(t, repo, "other-agent").Status), "queued")
+
+	// Re-running the same selector cancels nothing: everything eligible
+	// has already reached a terminal state.
+	cancelled, err = repo.CancelRunsWhere(ctx, "agent removed", `agent_profile_id = ?`, "a1")
+	if err != nil {
+		t.Fatalf("cancel where (second pass): %v", err)
+	}
+	if cancelled != 0 {
+		t.Errorf("second pass cancelled = %d, want 0", cancelled)
+	}
+}
+
+// TestCancelRunsWhere_MultiArgSelectorBindsInOrder covers a selector
+// with several placeholders, the shape office's by-task selector uses.
+func TestCancelRunsWhere_MultiArgSelectorBindsInOrder(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+
+	mustCreateRun(t, repo, &models.Run{
+		ID: "match", AgentProfileID: "a1", Reason: "task_comment",
+		Payload: `{"task_id":"t1"}`, Status: "queued",
+	})
+	mustCreateRun(t, repo, &models.Run{
+		ID: "wrong-reason", AgentProfileID: "a1", Reason: "heartbeat",
+		Payload: `{"task_id":"t1"}`, Status: "queued",
+	})
+
+	cancelled, err := repo.CancelRunsWhere(ctx, "superseded",
+		`agent_profile_id = ? AND reason = ?`, "a1", "task_comment")
+	if err != nil {
+		t.Fatalf("cancel where: %v", err)
+	}
+	if cancelled != 1 {
+		t.Errorf("cancelled = %d, want 1", cancelled)
+	}
+	checkString(t, "match", string(mustGetRun(t, repo, "match").Status), "cancelled")
+	checkString(t, "wrong-reason", string(mustGetRun(t, repo, "wrong-reason").Status), "queued")
+}
+
+// TestBulkCancelRuns_CancelsOnlyTheListedEligibleRuns pins the id
+// selector: rows in the set that are terminal, and rows outside it,
+// are both left alone.
+func TestBulkCancelRuns_CancelsOnlyTheListedEligibleRuns(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	now := time.Now().UTC()
+
+	queueRunAt(t, repo, "in-set-queued", "a1", now)
+	claimed := queueRunAt(t, repo, "in-set-claimed", "a1", now)
+	setStatus(t, repo, claimed.ID, "claimed", timePtr(now), nil)
+	finished := queueRunAt(t, repo, "in-set-finished", "a1", now)
+	setStatus(t, repo, finished.ID, "finished", nil, timePtr(now))
+	queueRunAt(t, repo, "out-of-set", "a1", now)
+
+	err := repo.BulkCancelRuns(ctx,
+		[]string{"in-set-queued", "in-set-claimed", "in-set-finished", "missing-id"},
+		"workspace deleted")
+	if err != nil {
+		t.Fatalf("bulk cancel: %v", err)
+	}
+
+	checkString(t, "in-set-queued", string(mustGetRun(t, repo, "in-set-queued").Status), "cancelled")
+	checkString(t, "in-set-claimed", string(mustGetRun(t, repo, "in-set-claimed").Status), "cancelled")
+	checkString(t, "in-set-finished", string(mustGetRun(t, repo, "in-set-finished").Status), "finished")
+	checkString(t, "out-of-set", string(mustGetRun(t, repo, "out-of-set").Status), "queued")
+	checkStringPtr(t, "in-set-queued reason",
+		mustGetRun(t, repo, "in-set-queued").CancelReason, strPtr("workspace deleted"))
+	if reason := mustGetRun(t, repo, "out-of-set").CancelReason; reason != nil {
+		t.Errorf("out-of-set cancel_reason = %q, want nil", *reason)
+	}
+}
+
+// TestBulkCancelRuns_EmptyIDListIsANoOp pins that the degenerate call
+// does not fall through to an unbounded UPDATE.
+func TestBulkCancelRuns_EmptyIDListIsANoOp(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	queueRunAt(t, repo, "survivor", "a1", time.Now().UTC())
+
+	for _, ids := range [][]string{nil, {}} {
+		if err := repo.BulkCancelRuns(ctx, ids, "nothing"); err != nil {
+			t.Fatalf("bulk cancel %v: %v", ids, err)
+		}
+	}
+	got := mustGetRun(t, repo, "survivor")
+	checkString(t, "status", string(got.Status), "queued")
+	if got.CancelReason != nil {
+		t.Errorf("cancel_reason = %q, want nil", *got.CancelReason)
+	}
+}

apps/backend/internal/runs/repository/sqlite/errors_test.go

@@ -0,0 +1,203 @@
+package sqlite_test
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/kandev/kandev/internal/office/models"
+)
+
+// TestReadMethodsPropagateDatabaseErrors closes the reader pool and
+// asserts every read surfaces the failure instead of degrading to an
+// empty result — a swallowed error here would render an empty run list
+// as "this agent has no runs".
+func TestReadMethodsPropagateDatabaseErrors(t *testing.T) {
+	repo, _, reader := newTestRepoWithHandles(t)
+	ctx := context.Background()
+	run := mustCreateRun(t, repo, &models.Run{
+		ID: "run-1", AgentProfileID: "a1", Reason: "task_assigned",
+		Payload: `{"task_id":"t1","comment_id":"cm-1"}`,
+	})
+	if err := reader.Close(); err != nil {
+		t.Fatalf("close reader: %v", err)
+	}
+
+	checks := map[string]func() error{
+		"GetRunByID": func() error {
+			_, err := repo.GetRunByID(ctx, run.ID)
+			return err
+		},
+		"ListRuns": func() error {
+			_, err := repo.ListRuns(ctx, "ws-1")
+			return err
+		},
+		"ListRunEvents": func() error {
+			_, err := repo.ListRunEvents(ctx, run.ID, -1, 0)
+			return err
+		},
+		"CheckIdempotencyKey": func() error {
+			_, err := repo.CheckIdempotencyKey(ctx, "key", 1)
+			return err
+		},
+		"ListPendingRunsForTask": func() error {
+			_, err := repo.ListPendingRunsForTask(ctx, "t1")
+			return err
+		},
+		"ListRunsForAgentPaged(first page)": func() error {
+			_, err := repo.ListRunsForAgentPaged(ctx, "a1", time.Time{}, "", 10)
+			return err
+		},
+		"ListRunsForAgentPaged(cursor)": func() error {
+			_, err := repo.ListRunsForAgentPaged(ctx, "a1", time.Now().UTC(), "cursor", 10)
+			return err
+		},
+		"FindInflightRunForAgent": func() error {
+			_, err := repo.FindInflightRunForAgent(ctx, "a1")
+			return err
+		},
+		"GetClaimedRunByTaskID": func() error {
+			_, err := repo.GetClaimedRunByTaskID(ctx, "t1")
+			return err
+		},
+		"GetClaimedTasklessRunForAgent": func() error {
+			_, err := repo.GetClaimedTasklessRunForAgent(ctx, "a1")
+			return err
+		},
+		"GetRunsByCommentIDs": func() error {
+			_, err := repo.GetRunsByCommentIDs(ctx, []string{"cm-1"})
+			return err
+		},
+		"GetRunWithCosts": func() error {
+			_, _, err := repo.GetRunWithCosts(ctx, run.ID)
+			return err
+		},
+	}
+	for name, call := range checks {
+		err := call()
+		if err == nil {
+			t.Errorf("%s returned nil error against a closed reader", name)
+			continue
+		}
+		if errors.Is(err, sql.ErrNoRows) {
+			t.Errorf("%s reported sql.ErrNoRows for a closed reader, want the underlying failure", name)
+		}
+	}
+}
+
+// TestWriteMethodsPropagateDatabaseErrors is the same contract for the
+// writer pool.
+func TestWriteMethodsPropagateDatabaseErrors(t *testing.T) {
+	repo, writer, _ := newTestRepoWithHandles(t)
+	ctx := context.Background()
+	run := mustCreateRun(t, repo, &models.Run{
+		ID: "run-1", AgentProfileID: "a1", Reason: "task_assigned",
+	})
+	if err := writer.Close(); err != nil {
+		t.Fatalf("close writer: %v", err)
+	}
+
+	checks := map[string]func() error{
+		"CreateRun": func() error {
+			return repo.CreateRun(ctx, &models.Run{AgentProfileID: "a1", Reason: "r"})
+		},
+		"UpdateRunRuntimeSnapshot": func() error {
+			return repo.UpdateRunRuntimeSnapshot(ctx, run.ID, "{}", "{}", "s")
+		},
+		"UpdateRunPromptArtifacts": func() error {
+			return repo.UpdateRunPromptArtifacts(ctx, run.ID, "p", "s")
+		},
+		"UpdateRunResultJSON": func() error {
+			return repo.UpdateRunResultJSON(ctx, run.ID, "{}")
+		},
+		"UpdateRunOutputSummary": func() error {
+			return repo.UpdateRunOutputSummary(ctx, run.ID, "o", "f")
+		},
+		"FinishRun": func() error {
+			return repo.FinishRun(ctx, run.ID, "finished")
+		},
+		"ClaimRun": func() error {
+			_, err := repo.ClaimRun(ctx, "a1")
+			return err
+		},
+		"ClaimNextEligibleRun": func() error {
+			_, err := repo.ClaimNextEligibleRun(ctx)
+			return err
+		},
+		"CoalesceRun": func() error {
+			_, err := repo.CoalesceRun(ctx, "a1", "r", 60, "{}")
+			return err
+		},
+		"ScheduleRetry": func() error {
+			return repo.ScheduleRetry(ctx, run.ID, time.Now().UTC(), 1)
+		},
+		"CleanExpired": func() error {
+			_, err := repo.CleanExpired(ctx, time.Now().UTC())
+			return err
+		},
+		"RecoverStale": func() error {
+			_, err := repo.RecoverStale(ctx, time.Now().UTC())
+			return err
+		},
+		"CancelRunsWhere": func() error {
+			_, err := repo.CancelRunsWhere(ctx, "reason", `id = ?`, run.ID)
+			return err
+		},
+		"CancelRun": func() error {
+			return repo.CancelRun(ctx, run.ID, "reason")
+		},
+		"BulkCancelRuns": func() error {
+			return repo.BulkCancelRuns(ctx, []string{run.ID}, "reason")
+		},
+		"AppendRunEvent": func() error {
+			_, err := repo.AppendRunEvent(ctx, run.ID, "run.progress", "info", "{}")
+			return err
+		},
+		"SetRunRequestedAtForTest": func() error {
+			return repo.SetRunRequestedAtForTest(ctx, run.ID, time.Now().UTC())
+		},
+		"SetRunStatusForTest": func() error {
+			return repo.SetRunStatusForTest(ctx, run.ID, "queued", nil, nil)
+		},
+		"SetRunErrorMessageForTest": func() error {
+			return repo.SetRunErrorMessageForTest(ctx, run.ID, "boom")
+		},
+	}
+	for name, call := range checks {
+		if err := call(); err == nil {
+			t.Errorf("%s returned nil error against a closed writer", name)
+		}
+	}
+}
+
+// TestGetRunWithCosts_RollupFailureDegradesToZeroes pins the documented
+// asymmetry: the run row is mandatory, the cost rollup is best-effort.
+// Losing the cost table must still return the run with a zero rollup
+// rather than failing the whole read.
+func TestGetRunWithCosts_RollupFailureDegradesToZeroes(t *testing.T) {
+	repo, writer := newTestRepoWithDB(t)
+	ctx := context.Background()
+	run := seedTaskRun(t, repo, "costed", "a1", "t1", "queued")
+	seedCostEvent(t, writer, "c1", "t1", 100, 10, 20, 5)
+
+	if _, err := writer.Exec(`DROP TABLE office_cost_events`); err != nil {
+		t.Fatalf("drop cost table: %v", err)
+	}
+
+	got, rollup, err := repo.GetRunWithCosts(ctx, run.ID)
+	if err != nil {
+		t.Fatalf("get run with costs: %v", err)
+	}
+	if got == nil || got.ID != run.ID {
+		t.Fatalf("run = %v, want %q", got, run.ID)
+	}
+	if rollup == nil {
+		t.Fatalf("rollup = nil, want a zero rollup")
+	}
+	checkInt64(t, "input_tokens", rollup.InputTokens, 0)
+	checkInt64(t, "output_tokens", rollup.OutputTokens, 0)
+	checkInt64(t, "cached_tokens", rollup.CachedTokens, 0)
+	checkInt64(t, "cost_subcents", rollup.CostSubcents, 0)
+}

apps/backend/internal/runs/repository/sqlite/run_events_test.go

@@ -0,0 +1,225 @@
+package sqlite_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/kandev/kandev/internal/office/models"
+)
+
+// TestAppendRunEvent_ReturnedEventMatchesTheStoredRow is the round-trip
+// guard for run_events: every column written must come back out of the
+// list query, and the value returned to the caller (which is published
+// on the event bus without re-reading) must equal what landed on disk.
+func TestAppendRunEvent_ReturnedEventMatchesTheStoredRow(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	run := mustCreateRun(t, repo, &models.Run{
+		ID: "run-1", AgentProfileID: "a1", Reason: "task_assigned",
+	})
+	before := time.Now().UTC().Add(-time.Second)
+
+	got, err := repo.AppendRunEvent(ctx, run.ID, "run.started", "warn", `{"detail":"x"}`)
+	if err != nil {
+		t.Fatalf("append: %v", err)
+	}
+	checkString(t, "run_id", got.RunID, run.ID)
+	checkInt(t, "seq", got.Seq, 0)
+	checkString(t, "event_type", string(got.EventType), "run.started")
+	checkString(t, "level", string(got.Level), "warn")
+	checkString(t, "payload", got.Payload, `{"detail":"x"}`)
+	if !got.CreatedAt.After(before) {
+		t.Errorf("created_at = %s, want after %s", got.CreatedAt, before)
+	}
+
+	stored, err := repo.ListRunEvents(ctx, run.ID, -1, 0)
+	if err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	if len(stored) != 1 {
+		t.Fatalf("%d stored events, want 1", len(stored))
+	}
+	checkString(t, "stored run_id", stored[0].RunID, got.RunID)
+	checkInt(t, "stored seq", stored[0].Seq, got.Seq)
+	checkString(t, "stored event_type", string(stored[0].EventType), string(got.EventType))
+	checkString(t, "stored level", string(stored[0].Level), string(got.Level))
+	checkString(t, "stored payload", stored[0].Payload, got.Payload)
+	if !stored[0].CreatedAt.Equal(got.CreatedAt) {
+		t.Errorf("stored created_at = %s, want %s", stored[0].CreatedAt, got.CreatedAt)
+	}
+}
+
+// TestAppendRunEvent_AppliesLevelAndPayloadDefaults pins the two
+// documented fallbacks.
+func TestAppendRunEvent_AppliesLevelAndPayloadDefaults(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+
+	got, err := repo.AppendRunEvent(ctx, "run-1", "run.progress", "", "")
+	if err != nil {
+		t.Fatalf("append: %v", err)
+	}
+	checkString(t, "returned level", string(got.Level), "info")
+	checkString(t, "returned payload", got.Payload, "{}")
+
+	stored, err := repo.ListRunEvents(ctx, "run-1", -1, 0)
+	if err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	if len(stored) != 1 {
+		t.Fatalf("%d stored events, want 1", len(stored))
+	}
+	checkString(t, "stored level", string(stored[0].Level), "info")
+	checkString(t, "stored payload", stored[0].Payload, "{}")
+}
+
+// TestAppendRunEvent_SeqIsMonotonicPerRun pins that the sequence is
+// scoped to one run: each run starts at 0 and increments independently.
+//
+// Sequential only, deliberately. AppendRunEvent's doc comment claims
+// events "land monotonically even when multiple writers race", but the
+// implementation reads MAX(seq)+1 and inserts in two separate
+// statements, so concurrent appends to one run collide on the
+// (run_id, seq) primary key. Ten concurrent appends were measured
+// failing 6-8 times with "UNIQUE constraint failed: run_events.run_id,
+// run_events.seq". Making the allocation atomic is a production change
+// and is out of scope for this test-only change; a concurrent test is
+// therefore not added here, since it would pin a bug rather than a
+// contract.
+func TestAppendRunEvent_SeqIsMonotonicPerRun(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+
+	for want := range 3 {
+		got, err := repo.AppendRunEvent(ctx, "run-1", "run.progress", "info", "{}")
+		if err != nil {
+			t.Fatalf("append %d: %v", want, err)
+		}
+		checkInt(t, "seq", got.Seq, want)
+	}
+
+	other, err := repo.AppendRunEvent(ctx, "run-2", "run.progress", "info", "{}")
+	if err != nil {
+		t.Fatalf("append to second run: %v", err)
+	}
+	checkInt(t, "seq of a different run", other.Seq, 0)
+
+	next, err := repo.AppendRunEvent(ctx, "run-1", "run.progress", "info", "{}")
+	if err != nil {
+		t.Fatalf("append after the second run: %v", err)
+	}
+	checkInt(t, "seq resumes on the first run", next.Seq, 3)
+}
+
+// TestListRunEvents_OrdersBySeqAscendingAndScopesToTheRun inserts rows
+// out of seq order so the ORDER BY is what produces the result order,
+// not the insertion order.
+func TestListRunEvents_OrdersBySeqAscendingAndScopesToTheRun(t *testing.T) {
+	repo, writer := newTestRepoWithDB(t)
+	ctx := context.Background()
+	now := time.Now().UTC()
+
+	for _, seq := range []int{5, 1, 3} {
+		if _, err := writer.Exec(`INSERT INTO run_events
+			(run_id, seq, event_type, level, payload, created_at)
+			VALUES ('run-1', ?, 'run.progress', 'info', '{}', ?)`, seq, now); err != nil {
+			t.Fatalf("insert seq %d: %v", seq, err)
+		}
+	}
+	if _, err := writer.Exec(`INSERT INTO run_events
+		(run_id, seq, event_type, level, payload, created_at)
+		VALUES ('run-2', 0, 'run.progress', 'info', '{}', ?)`, now); err != nil {
+		t.Fatalf("insert other run: %v", err)
+	}
+
+	got, err := repo.ListRunEvents(ctx, "run-1", -1, 0)
+	if err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	checkSeqOrder(t, "run-1 events", got, 1, 3, 5)
+}
+
+// TestListRunEvents_AfterSeqReturnsStrictlyNewerEvents pins the live
+// tail's cursor semantics.
+func TestListRunEvents_AfterSeqReturnsStrictlyNewerEvents(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	for range 4 {
+		if _, err := repo.AppendRunEvent(ctx, "run-1", "run.progress", "info", "{}"); err != nil {
+			t.Fatalf("append: %v", err)
+		}
+	}
+
+	all, err := repo.ListRunEvents(ctx, "run-1", -1, 0)
+	if err != nil {
+		t.Fatalf("list all: %v", err)
+	}
+	checkSeqOrder(t, "all events", all, 0, 1, 2, 3)
+
+	tail, err := repo.ListRunEvents(ctx, "run-1", 1, 0)
+	if err != nil {
+		t.Fatalf("list tail: %v", err)
+	}
+	checkSeqOrder(t, "events after seq 1", tail, 2, 3)
+
+	caughtUp, err := repo.ListRunEvents(ctx, "run-1", 3, 0)
+	if err != nil {
+		t.Fatalf("list caught up: %v", err)
+	}
+	if len(caughtUp) != 0 {
+		t.Errorf("got %d events after seq 3, want none", len(caughtUp))
+	}
+}
+
+// TestListRunEvents_LimitCapsTheOldestEvents pins that the cap applies
+// after the ascending sort, and that limit 0 means "no cap".
+func TestListRunEvents_LimitCapsTheOldestEvents(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	for range 5 {
+		if _, err := repo.AppendRunEvent(ctx, "run-1", "run.progress", "info", "{}"); err != nil {
+			t.Fatalf("append: %v", err)
+		}
+	}
+
+	capped, err := repo.ListRunEvents(ctx, "run-1", -1, 2)
+	if err != nil {
+		t.Fatalf("list capped: %v", err)
+	}
+	checkSeqOrder(t, "limit 2", capped, 0, 1)
+
+	uncapped, err := repo.ListRunEvents(ctx, "run-1", -1, 0)
+	if err != nil {
+		t.Fatalf("list uncapped: %v", err)
+	}
+	checkSeqOrder(t, "limit 0", uncapped, 0, 1, 2, 3, 4)
+}
+
+// TestListRunEvents_UnknownRunReturnsNoEvents pins the empty case.
+func TestListRunEvents_UnknownRunReturnsNoEvents(t *testing.T) {
+	repo := newTestRepo(t)
+	got, err := repo.ListRunEvents(context.Background(), "nope", -1, 0)
+	if err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	if len(got) != 0 {
+		t.Errorf("got %d events, want none", len(got))
+	}
+}
+
+func checkSeqOrder(t *testing.T, label string, got []*models.RunEvent, want ...int) {
+	t.Helper()
+	seqs := make([]int, 0, len(got))
+	for _, event := range got {
+		seqs = append(seqs, event.Seq)
+	}
+	if len(seqs) != len(want) {
+		t.Fatalf("%s = %v, want %v", label, seqs, want)
+	}
+	for i := range want {
+		if seqs[i] != want[i] {
+			t.Fatalf("%s = %v, want %v", label, seqs, want)
+		}
+	}
+}

apps/backend/internal/runs/repository/sqlite/runs_claim_test.go

@@ -0,0 +1,354 @@
+package sqlite_test
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/jmoiron/sqlx"
+
+	"github.com/kandev/kandev/internal/office/models"
+	runssqlite "github.com/kandev/kandev/internal/runs/repository/sqlite"
+)
+
+// queueRunAt seeds a queued run for an agent and backdates it so
+// ordering assertions are deterministic.
+func queueRunAt(
+	t *testing.T, repo *runssqlite.Repository, id, agentID string, requestedAt time.Time,
+) *models.Run {
+	t.Helper()
+	run := mustCreateRun(t, repo, &models.Run{
+		ID:             id,
+		AgentProfileID: agentID,
+		Reason:         "task_assigned",
+		Payload:        `{"task_id":"` + id + `"}`,
+		Status:         "queued",
+		CoalescedCount: 1,
+	})
+	setRequestedAt(t, repo, run.ID, requestedAt)
+	return run
+}
+
+// TestClaimRun_ClaimsTheOldestQueuedRunForTheAgent asserts which run
+// comes back, not merely that one did: the queue is FIFO by
+// requested_at, so reversing the ORDER BY must fail here.
+func TestClaimRun_ClaimsTheOldestQueuedRunForTheAgent(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	base := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC)
+
+	oldest := queueRunAt(t, repo, "oldest", "a1", base)
+	queueRunAt(t, repo, "middle", "a1", base.Add(time.Minute))
+	queueRunAt(t, repo, "newest", "a1", base.Add(2*time.Minute))
+	// An older run belonging to a different agent must not be picked.
+	queueRunAt(t, repo, "other-agent", "a2", base.Add(-time.Hour))
+
+	before := time.Now().UTC().Add(-time.Second)
+	claimed, err := repo.ClaimRun(ctx, "a1")
+	if err != nil {
+		t.Fatalf("claim: %v", err)
+	}
+	if claimed.ID != oldest.ID {
+		t.Errorf("claimed %q, want %q (oldest queued run for a1)", claimed.ID, oldest.ID)
+	}
+	checkString(t, "returned status", string(claimed.Status), "claimed")
+	if claimed.ClaimedAt == nil || !claimed.ClaimedAt.After(before) {
+		t.Errorf("returned claimed_at = %v, want a fresh stamp after %s", claimed.ClaimedAt, before)
+	}
+
+	stored := mustGetRun(t, repo, oldest.ID)
+	checkString(t, "stored status", string(stored.Status), "claimed")
+	if stored.ClaimedAt == nil {
+		t.Errorf("stored claimed_at = nil, want a stamp")
+	}
+	checkString(t, "sibling status", string(mustGetRun(t, repo, "middle").Status), "queued")
+	checkString(t, "other agent status", string(mustGetRun(t, repo, "other-agent").Status), "queued")
+}
+
+// TestClaimRun_NoQueuedRunReturnsNoRows pins the sentinel the scheduler
+// treats as "nothing to do" — including the case where the agent's only
+// runs are already claimed or terminal.
+func TestClaimRun_NoQueuedRunReturnsNoRows(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+
+	run := queueRunAt(t, repo, "claimed-already", "a1", time.Now().UTC())
+	setStatus(t, repo, run.ID, "claimed", timePtr(time.Now().UTC()), nil)
+	done := queueRunAt(t, repo, "finished", "a1", time.Now().UTC())
+	setStatus(t, repo, done.ID, "finished", nil, timePtr(time.Now().UTC()))
+
+	if _, err := repo.ClaimRun(ctx, "a1"); !errors.Is(err, sql.ErrNoRows) {
+		t.Fatalf("err = %v, want sql.ErrNoRows", err)
+	}
+	if _, err := repo.ClaimRun(ctx, "unknown-agent"); !errors.Is(err, sql.ErrNoRows) {
+		t.Fatalf("unknown agent err = %v, want sql.ErrNoRows", err)
+	}
+}
+
+// TestClaimRun_ConcurrentClaimsAreExclusive races real goroutines at a
+// fixed set of queued runs. Every run must be handed to exactly one
+// caller: a weakened status guard hands the same row out twice, which
+// two sequential calls would never reveal.
+func TestClaimRun_ConcurrentClaimsAreExclusive(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	base := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC)
+
+	const queued = 3
+	const claimers = 8
+	for i := range queued {
+		queueRunAt(t, repo, string(rune('a'+i))+"-run", "a1", base.Add(time.Duration(i)*time.Minute))
+	}
+
+	ids, noRows := runConcurrentClaims(t, claimers, func() (*models.Run, error) {
+		return repo.ClaimRun(ctx, "a1")
+	})
+
+	if len(ids) != queued {
+		t.Errorf("%d successful claims, want %d", len(ids), queued)
+	}
+	if noRows != claimers-queued {
+		t.Errorf("%d sql.ErrNoRows results, want %d", noRows, claimers-queued)
+	}
+	assertNoDuplicates(t, ids)
+	assertClaimedCount(t, repo, queued)
+}
+
+// TestClaimNextEligibleRun_PicksTheOldestEligibleRun asserts the queue
+// order across agents.
+func TestClaimNextEligibleRun_PicksTheOldestEligibleRun(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	base := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC)
+
+	queueRunAt(t, repo, "newer", "a1", base.Add(time.Minute))
+	oldest := queueRunAt(t, repo, "older", "a2", base)
+
+	claimed, err := repo.ClaimNextEligibleRun(ctx)
+	if err != nil {
+		t.Fatalf("claim: %v", err)
+	}
+	if claimed.ID != oldest.ID {
+		t.Errorf("claimed %q, want %q (oldest queued run)", claimed.ID, oldest.ID)
+	}
+	checkString(t, "status", string(claimed.Status), "claimed")
+	checkString(t, "untouched sibling", string(mustGetRun(t, repo, "newer").Status), "queued")
+}
+
+// TestClaimNextEligibleRun_SkipsAgentThatAlreadyHasAClaimedRun is the
+// one-run-per-agent guard. The busy agent owns the *older* queued run,
+// so a guard that stops counting claimed runs correctly returns the
+// wrong row rather than merely returning nothing.
+func TestClaimNextEligibleRun_SkipsAgentThatAlreadyHasAClaimedRun(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	base := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC)
+
+	busy := queueRunAt(t, repo, "busy-agent-inflight", "a1", base.Add(-time.Hour))
+	setStatus(t, repo, busy.ID, "claimed", timePtr(base), nil)
+	queueRunAt(t, repo, "busy-agent-queued", "a1", base)
+	idle := queueRunAt(t, repo, "idle-agent-queued", "a2", base.Add(time.Minute))
+
+	claimed, err := repo.ClaimNextEligibleRun(ctx)
+	if err != nil {
+		t.Fatalf("claim: %v", err)
+	}
+	if claimed.ID != idle.ID {
+		t.Errorf("claimed %q, want %q (a1 already has a claimed run)", claimed.ID, idle.ID)
+	}
+	checkString(t, "busy agent queued run", string(mustGetRun(t, repo, "busy-agent-queued").Status), "queued")
+}
+
+// TestClaimNextEligibleRun_HonoursScheduledRetryWindow pins the retry
+// gate in both directions.
+func TestClaimNextEligibleRun_HonoursScheduledRetryWindow(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	base := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC)
+
+	future := queueRunAt(t, repo, "retry-future", "a1", base)
+	if err := repo.ScheduleRetry(ctx, future.ID, time.Now().UTC().Add(time.Hour), 1); err != nil {
+		t.Fatalf("schedule future retry: %v", err)
+	}
+	setRequestedAt(t, repo, future.ID, base)
+	due := queueRunAt(t, repo, "retry-due", "a2", base.Add(time.Minute))
+	if err := repo.ScheduleRetry(ctx, due.ID, time.Now().UTC().Add(-time.Hour), 1); err != nil {
+		t.Fatalf("schedule due retry: %v", err)
+	}
+	setRequestedAt(t, repo, due.ID, base.Add(time.Minute))
+
+	claimed, err := repo.ClaimNextEligibleRun(ctx)
+	if err != nil {
+		t.Fatalf("claim: %v", err)
+	}
+	if claimed.ID != due.ID {
+		t.Errorf("claimed %q, want %q (the future retry must be skipped)", claimed.ID, due.ID)
+	}
+
+	// With the due run taken, the future retry is still not eligible.
+	if _, err := repo.ClaimNextEligibleRun(ctx); !errors.Is(err, sql.ErrNoRows) {
+		t.Errorf("second claim err = %v, want sql.ErrNoRows", err)
+	}
+}
+
+// TestClaimNextEligibleRun_SkipsRoutingBlockedRuns pins the parked-run
+// filter: a run parked by the provider router is not dispatchable even
+// though it is queued and older than everything else.
+func TestClaimNextEligibleRun_SkipsRoutingBlockedRuns(t *testing.T) {
+	repo, writer := newTestRepoWithDB(t)
+	ctx := context.Background()
+	base := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC)
+
+	parked := queueRunAt(t, repo, "parked", "a1", base)
+	if _, err := writer.Exec(
+		`UPDATE runs SET routing_blocked_status = 'waiting_for_provider_capacity' WHERE id = ?`,
+		parked.ID,
+	); err != nil {
+		t.Fatalf("park run: %v", err)
+	}
+	open := queueRunAt(t, repo, "open", "a2", base.Add(time.Minute))
+
+	claimed, err := repo.ClaimNextEligibleRun(ctx)
+	if err != nil {
+		t.Fatalf("claim: %v", err)
+	}
+	if claimed.ID != open.ID {
+		t.Errorf("claimed %q, want %q (parked run must be skipped)", claimed.ID, open.ID)
+	}
+	checkString(t, "parked run status", string(mustGetRun(t, repo, parked.ID).Status), "queued")
+}
+
+// TestClaimNextEligibleRun_EmptyQueueReturnsNoRows pins the sentinel.
+func TestClaimNextEligibleRun_EmptyQueueReturnsNoRows(t *testing.T) {
+	repo := newTestRepo(t)
+	if _, err := repo.ClaimNextEligibleRun(context.Background()); !errors.Is(err, sql.ErrNoRows) {
+		t.Fatalf("err = %v, want sql.ErrNoRows", err)
+	}
+}
+
+// TestClaimNextEligibleRun_ConcurrentClaimsRespectOnePerAgent is the
+// reservation guard under real contention: three agents with two queued
+// runs each must yield exactly three claims, one per agent, no run
+// handed out twice. Sequential calls cannot distinguish a count guard
+// of "= 0" from one that accepts any count; concurrent ones do.
+func TestClaimNextEligibleRun_ConcurrentClaimsRespectOnePerAgent(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	base := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC)
+
+	agents := []string{"a1", "a2", "a3"}
+	for i, agent := range agents {
+		for run := range 2 {
+			queueRunAt(t, repo, agent+"-"+string(rune('0'+run)), agent,
+				base.Add(time.Duration(i*2+run)*time.Minute))
+		}
+	}
+
+	const claimers = 10
+	ids, noRows := runConcurrentClaims(t, claimers, func() (*models.Run, error) {
+		return repo.ClaimNextEligibleRun(ctx)
+	})
+
+	if len(ids) != len(agents) {
+		t.Errorf("%d successful claims, want %d (one per agent)", len(ids), len(agents))
+	}
+	if noRows != claimers-len(agents) {
+		t.Errorf("%d sql.ErrNoRows results, want %d", noRows, claimers-len(agents))
+	}
+	assertNoDuplicates(t, ids)
+	assertClaimedCount(t, repo, len(agents))
+	assertOneClaimPerAgent(t, repo)
+}
+
+// runConcurrentClaims fires n goroutines at claim, released together by
+// closing a start channel, and returns the claimed IDs plus the number
+// of sql.ErrNoRows results. Any other error fails the test.
+func runConcurrentClaims(
+	t *testing.T, n int, claim func() (*models.Run, error),
+) ([]string, int) {
+	t.Helper()
+	start := make(chan struct{})
+	results := make(chan *models.Run, n)
+	errs := make(chan error, n)
+
+	var wg sync.WaitGroup
+	wg.Add(n)
+	for range n {
+		go func() {
+			defer wg.Done()
+			<-start
+			run, err := claim()
+			if err != nil {
+				errs <- err
+				return
+			}
+			results <- run
+		}()
+	}
+	close(start)
+	wg.Wait()
+	close(results)
+	close(errs)
+
+	var ids []string
+	for run := range results {
+		ids = append(ids, run.ID)
+	}
+	noRows := 0
+	for err := range errs {
+		if !errors.Is(err, sql.ErrNoRows) {
+			t.Errorf("unexpected claim error: %v", err)
+			continue
+		}
+		noRows++
+	}
+	return ids, noRows
+}
+
+func assertNoDuplicates(t *testing.T, ids []string) {
+	t.Helper()
+	seen := make(map[string]int, len(ids))
+	for _, id := range ids {
+		seen[id]++
+	}
+	for id, count := range seen {
+		if count > 1 {
+			t.Errorf("run %q was claimed %d times, want exactly 1", id, count)
+		}
+	}
+}
+
+func assertClaimedCount(t *testing.T, repo *runssqlite.Repository, want int) {
+	t.Helper()
+	var got int
+	if err := repo.Reader().Get(&got, `SELECT COUNT(*) FROM runs WHERE status = 'claimed'`); err != nil {
+		t.Fatalf("count claimed rows: %v", err)
+	}
+	if got != want {
+		t.Errorf("%d rows in status=claimed, want %d", got, want)
+	}
+}
+
+func assertOneClaimPerAgent(t *testing.T, repo *runssqlite.Repository) {
+	t.Helper()
+	type agentCount struct {
+		AgentProfileID string `db:"agent_profile_id"`
+		Claimed        int    `db:"claimed"`
+	}
+	var rows []agentCount
+	if err := sqlx.Select(repo.Reader(), &rows, `
+		SELECT agent_profile_id, COUNT(*) AS claimed
+		FROM runs WHERE status = 'claimed'
+		GROUP BY agent_profile_id
+	`); err != nil {
+		t.Fatalf("group claimed rows: %v", err)
+	}
+	for _, row := range rows {
+		if row.Claimed != 1 {
+			t.Errorf("agent %q holds %d claimed runs, want 1", row.AgentProfileID, row.Claimed)
+		}
+	}
+}

apps/backend/internal/runs/repository/sqlite/runs_comments_test.go

@@ -0,0 +1,214 @@
+package sqlite_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/kandev/kandev/internal/office/models"
+	runssqlite "github.com/kandev/kandev/internal/runs/repository/sqlite"
+)
+
+// commentRun seeds a run for the comment-status lookup with an explicit
+// idempotency key (pass nil for a key-less run) and a fixed
+// requested_at so "most recent wins" is deterministic.
+func commentRun(
+	t *testing.T, repo *runssqlite.Repository,
+	id, reason, payload string, key *string, requestedAt time.Time,
+) *models.Run {
+	t.Helper()
+	run := mustCreateRun(t, repo, &models.Run{
+		ID:             id,
+		AgentProfileID: "a1",
+		Reason:         reason,
+		Payload:        payload,
+		Status:         "queued",
+		CoalescedCount: 1,
+		IdempotencyKey: key,
+	})
+	setRequestedAt(t, repo, run.ID, requestedAt)
+	return run
+}
+
+// TestGetRunsByCommentIDs_EmptyInputReturnsEmptyMap pins the degenerate
+// case callers range over without a nil check.
+func TestGetRunsByCommentIDs_EmptyInputReturnsEmptyMap(t *testing.T) {
+	repo := newTestRepo(t)
+	got, err := repo.GetRunsByCommentIDs(context.Background(), nil)
+	if err != nil {
+		t.Fatalf("get: %v", err)
+	}
+	if got == nil {
+		t.Fatalf("got nil map, want an empty non-nil map")
+	}
+	if len(got) != 0 {
+		t.Errorf("got %v, want an empty map", got)
+	}
+}
+
+// TestGetRunsByCommentIDs_MapsCanonicalKeysAndCarriesStatus covers the
+// same-task wake: the canonical idempotency key resolves the comment,
+// and the badge fields (run id, status, error message) come back intact.
+func TestGetRunsByCommentIDs_MapsCanonicalKeysAndCarriesStatus(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	base := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC)
+
+	runA := commentRun(t, repo, "run-a", "task_comment",
+		`{"task_id":"t1","comment_id":"cm-A"}`, strPtr("task_comment:cm-A"), base)
+	runB := commentRun(t, repo, "run-b", "task_comment",
+		`{"task_id":"t1","comment_id":"cm-B"}`, strPtr("task_comment:cm-B"), base)
+	setStatus(t, repo, runB.ID, "failed", nil, timePtr(base))
+	if err := repo.SetRunErrorMessageForTest(ctx, runB.ID, "agent exploded"); err != nil {
+		t.Fatalf("set error message: %v", err)
+	}
+
+	got, err := repo.GetRunsByCommentIDs(ctx, []string{"cm-A", "cm-B", "cm-none"})
+	if err != nil {
+		t.Fatalf("get: %v", err)
+	}
+	if len(got) != 2 {
+		t.Fatalf("got %d entries, want 2 (cm-none has no run): %+v", len(got), got)
+	}
+	checkCommentStatus(t, got, "cm-A", runssqlite.CommentRunStatus{
+		RunID: runA.ID, Status: "queued", ErrorMessage: "",
+	})
+	checkCommentStatus(t, got, "cm-B", runssqlite.CommentRunStatus{
+		RunID: runB.ID, Status: "failed", ErrorMessage: "agent exploded",
+	})
+	if _, present := got["cm-none"]; present {
+		t.Errorf("cm-none is present (%+v), want absent", got["cm-none"])
+	}
+}
+
+// TestGetRunsByCommentIDs_SaltedKeysResolveThroughThePayload covers the
+// cross-task fan-out: the key carries salt after the comment id, so the
+// persisted payload.comment_id is what links the run back.
+func TestGetRunsByCommentIDs_SaltedKeysResolveThroughThePayload(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	base := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC)
+
+	salted := commentRun(t, repo, "salted", "task_comment",
+		`{"task_id":"target","source_task_id":"source","comment_id":"cm-X"}`,
+		strPtr("task_comment:cm-X:work:target:a1:abcd1234"), base)
+	keyless := commentRun(t, repo, "keyless", "task_comment",
+		`{"task_id":"target","comment_id":"cm-Y"}`, nil, base)
+
+	got, err := repo.GetRunsByCommentIDs(ctx, []string{"cm-X", "cm-Y"})
+	if err != nil {
+		t.Fatalf("get: %v", err)
+	}
+	checkCommentStatus(t, got, "cm-X", runssqlite.CommentRunStatus{RunID: salted.ID, Status: "queued"})
+	checkCommentStatus(t, got, "cm-Y", runssqlite.CommentRunStatus{RunID: keyless.ID, Status: "queued"})
+}
+
+// TestGetRunsByCommentIDs_MostRecentRunWinsPerComment pins the
+// ordering: with a canonical run and a later salted run for the same
+// comment, the badge must reflect the newer one.
+func TestGetRunsByCommentIDs_MostRecentRunWinsPerComment(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	base := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC)
+
+	older := commentRun(t, repo, "older", "task_comment",
+		`{"task_id":"t1","comment_id":"cm-A"}`, strPtr("task_comment:cm-A"), base)
+	setStatus(t, repo, older.ID, "finished", nil, timePtr(base))
+	newer := commentRun(t, repo, "newer", "task_comment",
+		`{"task_id":"t2","comment_id":"cm-A"}`,
+		strPtr("task_comment:cm-A:work:t2:a1:beef"), base.Add(time.Hour))
+
+	got, err := repo.GetRunsByCommentIDs(ctx, []string{"cm-A"})
+	if err != nil {
+		t.Fatalf("get: %v", err)
+	}
+	if len(got) != 1 {
+		t.Fatalf("got %d entries, want 1: %+v", len(got), got)
+	}
+	checkCommentStatus(t, got, "cm-A", runssqlite.CommentRunStatus{RunID: newer.ID, Status: "queued"})
+}
+
+// TestGetRunsByCommentIDs_IgnoresUnrelatedRuns pins the filters: a run
+// for another comment, and a key-less run whose reason is not
+// task_comment, must both stay out of the map.
+func TestGetRunsByCommentIDs_IgnoresUnrelatedRuns(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	base := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC)
+
+	commentRun(t, repo, "other-comment", "task_comment",
+		`{"task_id":"t1","comment_id":"cm-other"}`, strPtr("task_comment:cm-other"), base)
+	commentRun(t, repo, "wrong-reason", "task_assigned",
+		`{"task_id":"t1","comment_id":"cm-A"}`, nil, base.Add(time.Hour))
+	wanted := commentRun(t, repo, "wanted", "task_comment",
+		`{"task_id":"t1","comment_id":"cm-A"}`, strPtr("task_comment:cm-A"), base)
+
+	got, err := repo.GetRunsByCommentIDs(ctx, []string{"cm-A"})
+	if err != nil {
+		t.Fatalf("get: %v", err)
+	}
+	if len(got) != 1 {
+		t.Fatalf("got %d entries, want 1: %+v", len(got), got)
+	}
+	checkCommentStatus(t, got, "cm-A", runssqlite.CommentRunStatus{RunID: wanted.ID, Status: "queued"})
+}
+
+// TestGetRunsByCommentIDs_CanonicalKeyResolvesUnusablePayloads covers
+// the payload-parsing fallbacks: a payload with no comment_id, or with
+// a non-string one, must not stop the canonical key from resolving.
+// (A syntactically malformed payload cannot be seeded — the runs table
+// carries a partial index over json_extract(payload, '$.comment_id'),
+// so SQLite rejects the INSERT outright.)
+func TestGetRunsByCommentIDs_CanonicalKeyResolvesUnusablePayloads(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	base := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC)
+
+	cases := []struct {
+		commentID string
+		payload   string
+	}{
+		{"cm-numeric", `{"comment_id":42}`},
+		{"cm-absent", `{"task_id":"t1"}`},
+		// The payload names a comment that was not asked about: the
+		// canonical key still decides which comment this run belongs to.
+		{"cm-mismatched", `{"comment_id":"cm-not-requested"}`},
+	}
+	wantIDs := make(map[string]string, len(cases))
+	for i, tc := range cases {
+		run := commentRun(t, repo, "run-"+tc.commentID, "task_comment", tc.payload,
+			strPtr("task_comment:"+tc.commentID), base.Add(time.Duration(i)*time.Minute))
+		wantIDs[tc.commentID] = run.ID
+	}
+
+	ids := make([]string, 0, len(cases))
+	for _, tc := range cases {
+		ids = append(ids, tc.commentID)
+	}
+	got, err := repo.GetRunsByCommentIDs(ctx, ids)
+	if err != nil {
+		t.Fatalf("get: %v", err)
+	}
+	if len(got) != len(cases) {
+		t.Fatalf("got %d entries, want %d: %+v", len(got), len(cases), got)
+	}
+	for commentID, runID := range wantIDs {
+		checkCommentStatus(t, got, commentID, runssqlite.CommentRunStatus{RunID: runID, Status: "queued"})
+	}
+}
+
+func checkCommentStatus(
+	t *testing.T,
+	got map[string]runssqlite.CommentRunStatus,
+	commentID string,
+	want runssqlite.CommentRunStatus,
+) {
+	t.Helper()
+	status, ok := got[commentID]
+	if !ok {
+		t.Fatalf("comment %q missing from the map: %+v", commentID, got)
+	}
+	if status != want {
+		t.Errorf("comment %q = %+v, want %+v", commentID, status, want)
+	}
+}

apps/backend/internal/runs/repository/sqlite/runs_crud_test.go

@@ -0,0 +1,348 @@
+package sqlite_test
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/kandev/kandev/internal/office/models"
+)
+
+// fixedRetryAt is a whole-second timestamp so the round-trip assertion
+// compares an exact value rather than tolerating driver precision.
+var fixedRetryAt = time.Date(2026, 3, 4, 5, 6, 7, 0, time.UTC)
+
+// fullRun is a run with every column CreateRun writes set to a value
+// that differs from the schema default, so dropping any one column
+// from the INSERT changes what comes back out.
+func fullRun() *models.Run {
+	return &models.Run{
+		ID:               "run-full",
+		AgentProfileID:   "agent-42",
+		Reason:           "task_assigned",
+		Payload:          `{"task_id":"t-9"}`,
+		Status:           "claimed",
+		CoalescedCount:   7,
+		IdempotencyKey:   strPtr("idem-key-1"),
+		ContextSnapshot:  `{"ctx":1}`,
+		Capabilities:     `{"cap":true}`,
+		InputSnapshot:    `{"in":"x"}`,
+		OutputSummary:    "output summary",
+		FailureReason:    "failure reason",
+		SessionID:        "sess-7",
+		RetryCount:       3,
+		ScheduledRetryAt: timePtr(fixedRetryAt),
+		ErrorMessage:     "boom",
+		CancelReason:     strPtr("user cancelled"),
+	}
+}
+
+// TestCreateRun_PersistsEveryWrittenColumn is the round-trip guard for
+// the INSERT: write a row with every field populated, read it back,
+// compare all fields.
+func TestCreateRun_PersistsEveryWrittenColumn(t *testing.T) {
+	repo := newTestRepo(t)
+	want := fullRun()
+	mustCreateRun(t, repo, want)
+
+	got := mustGetRun(t, repo, want.ID)
+
+	checkString(t, "id", got.ID, want.ID)
+	checkString(t, "agent_profile_id", got.AgentProfileID, want.AgentProfileID)
+	checkString(t, "reason", got.Reason, want.Reason)
+	checkString(t, "payload", got.Payload, want.Payload)
+	checkString(t, "status", string(got.Status), string(want.Status))
+	checkString(t, "context_snapshot", got.ContextSnapshot, want.ContextSnapshot)
+	checkString(t, "capabilities", got.Capabilities, want.Capabilities)
+	checkString(t, "input_snapshot", got.InputSnapshot, want.InputSnapshot)
+	checkString(t, "output_summary", got.OutputSummary, want.OutputSummary)
+	checkString(t, "failure_reason", got.FailureReason, want.FailureReason)
+	checkString(t, "session_id", got.SessionID, want.SessionID)
+	checkString(t, "error_message", got.ErrorMessage, want.ErrorMessage)
+	checkInt(t, "coalesced_count", got.CoalescedCount, want.CoalescedCount)
+	checkInt(t, "retry_count", got.RetryCount, want.RetryCount)
+	checkStringPtr(t, "idempotency_key", got.IdempotencyKey, want.IdempotencyKey)
+	checkStringPtr(t, "cancel_reason", got.CancelReason, want.CancelReason)
+
+	if got.ScheduledRetryAt == nil {
+		t.Errorf("scheduled_retry_at = nil, want %s", fixedRetryAt)
+	} else if !got.ScheduledRetryAt.Equal(fixedRetryAt) {
+		t.Errorf("scheduled_retry_at = %s, want %s", got.ScheduledRetryAt, fixedRetryAt)
+	}
+}
+
+// TestCreateRun_LeavesUnwrittenColumnsAtTheirDefaults pins the other
+// half of the INSERT contract: the columns CreateRun deliberately does
+// not write (routing, prompt artifacts, timing) must land on their
+// schema defaults rather than NULL-scanning into a broken struct.
+func TestCreateRun_LeavesUnwrittenColumnsAtTheirDefaults(t *testing.T) {
+	repo := newTestRepo(t)
+	run := mustCreateRun(t, repo, fullRun())
+
+	got := mustGetRun(t, repo, run.ID)
+
+	checkString(t, "result_json", got.ResultJSON, "{}")
+	checkString(t, "assembled_prompt", got.AssembledPrompt, "")
+	checkString(t, "summary_injected", got.SummaryInjected, "")
+	checkInt(t, "current_route_attempt_seq", got.CurrentRouteAttemptSeq, 0)
+	checkInt(t, "route_cycle_baseline_seq", got.RouteCycleBaselineSeq, 0)
+	if got.ClaimedAt != nil {
+		t.Errorf("claimed_at = %s, want nil", got.ClaimedAt)
+	}
+	if got.FinishedAt != nil {
+		t.Errorf("finished_at = %s, want nil", got.FinishedAt)
+	}
+	if got.RoutingBlockedStatus != nil {
+		t.Errorf("routing_blocked_status = %v, want nil", *got.RoutingBlockedStatus)
+	}
+	if got.ResolvedProviderID != nil {
+		t.Errorf("resolved_provider_id = %v, want nil", *got.ResolvedProviderID)
+	}
+	if got.EarliestRetryAt != nil {
+		t.Errorf("earliest_retry_at = %s, want nil", got.EarliestRetryAt)
+	}
+}
+
+// TestCreateRun_AppliesDefaultsAndStampsIdentity covers ensureRunDefaults
+// plus the two fields CreateRun owns outright: a generated ID and a
+// server-side requested_at that overwrites whatever the caller passed.
+func TestCreateRun_AppliesDefaultsAndStampsIdentity(t *testing.T) {
+	repo := newTestRepo(t)
+	stale := time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC)
+	before := time.Now().UTC().Add(-time.Second)
+
+	run := &models.Run{
+		AgentProfileID: "a1",
+		Reason:         "heartbeat",
+		RequestedAt:    stale,
+	}
+	mustCreateRun(t, repo, run)
+
+	if run.ID == "" {
+		t.Fatalf("CreateRun left ID empty; it must generate one")
+	}
+	if !run.RequestedAt.After(before) {
+		t.Errorf("requested_at = %s, want a fresh stamp after %s (caller value must be ignored)",
+			run.RequestedAt, before)
+	}
+
+	got := mustGetRun(t, repo, run.ID)
+	checkString(t, "payload", got.Payload, "{}")
+	checkString(t, "context_snapshot", got.ContextSnapshot, "{}")
+	checkString(t, "capabilities", got.Capabilities, "{}")
+	checkString(t, "input_snapshot", got.InputSnapshot, "{}")
+	checkString(t, "status", string(got.Status), "queued")
+	checkString(t, "result_json", got.ResultJSON, "{}")
+	if got.RequestedAt.Equal(stale) {
+		t.Errorf("stored requested_at = %s, want the CreateRun stamp, not the caller value", got.RequestedAt)
+	}
+	if !got.RequestedAt.After(before) {
+		t.Errorf("stored requested_at = %s, want after %s", got.RequestedAt, before)
+	}
+}
+
+// TestCreateRun_KeepsCallerSuppliedID pins that a caller-chosen ID is
+// preserved (the idempotency and coalescing callers depend on it).
+func TestCreateRun_KeepsCallerSuppliedID(t *testing.T) {
+	repo := newTestRepo(t)
+	run := mustCreateRun(t, repo, &models.Run{
+		ID:             "chosen-id",
+		AgentProfileID: "a1",
+		Reason:         "task_assigned",
+	})
+	if run.ID != "chosen-id" {
+		t.Fatalf("ID = %q, want chosen-id", run.ID)
+	}
+	if got := mustGetRun(t, repo, "chosen-id"); got.ID != "chosen-id" {
+		t.Errorf("stored ID = %q, want chosen-id", got.ID)
+	}
+}
+
+// TestGetRunByID_UnknownRunReturnsNoRows pins the sentinel every caller
+// switches on.
+func TestGetRunByID_UnknownRunReturnsNoRows(t *testing.T) {
+	repo := newTestRepo(t)
+	_, err := repo.GetRunByID(context.Background(), "nope")
+	if !errors.Is(err, sql.ErrNoRows) {
+		t.Fatalf("err = %v, want sql.ErrNoRows", err)
+	}
+}
+
+// TestUpdateRunRuntimeSnapshot_WritesAllThreeColumns covers the
+// pre-launch snapshot write and asserts it touches nothing else.
+func TestUpdateRunRuntimeSnapshot_WritesAllThreeColumns(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	run := mustCreateRun(t, repo, fullRun())
+	other := mustCreateRun(t, repo, &models.Run{
+		ID: "other", AgentProfileID: "a2", Reason: "heartbeat",
+		Capabilities: `{"untouched":true}`,
+	})
+
+	if err := repo.UpdateRunRuntimeSnapshot(ctx, run.ID,
+		`{"cap":"new"}`, `{"in":"new"}`, "sess-new"); err != nil {
+		t.Fatalf("update runtime snapshot: %v", err)
+	}
+
+	got := mustGetRun(t, repo, run.ID)
+	checkString(t, "capabilities", got.Capabilities, `{"cap":"new"}`)
+	checkString(t, "input_snapshot", got.InputSnapshot, `{"in":"new"}`)
+	checkString(t, "session_id", got.SessionID, "sess-new")
+	checkString(t, "reason", got.Reason, "task_assigned")
+
+	untouched := mustGetRun(t, repo, other.ID)
+	checkString(t, "other capabilities", untouched.Capabilities, `{"untouched":true}`)
+	checkString(t, "other session_id", untouched.SessionID, "")
+}
+
+// TestUpdateRunPromptArtifacts_RoundTripsBothColumns covers the
+// dispatch-time prompt snapshot, including the documented empty case.
+func TestUpdateRunPromptArtifacts_RoundTripsBothColumns(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	run := mustCreateRun(t, repo, fullRun())
+
+	if err := repo.UpdateRunPromptArtifacts(ctx, run.ID, "the prompt", "the summary"); err != nil {
+		t.Fatalf("update prompt artifacts: %v", err)
+	}
+	got := mustGetRun(t, repo, run.ID)
+	checkString(t, "assembled_prompt", got.AssembledPrompt, "the prompt")
+	checkString(t, "summary_injected", got.SummaryInjected, "the summary")
+
+	if err := repo.UpdateRunPromptArtifacts(ctx, run.ID, "", ""); err != nil {
+		t.Fatalf("update prompt artifacts (empty): %v", err)
+	}
+	got = mustGetRun(t, repo, run.ID)
+	checkString(t, "assembled_prompt after empty write", got.AssembledPrompt, "")
+	checkString(t, "summary_injected after empty write", got.SummaryInjected, "")
+}
+
+// TestUpdateRunResultJSON_EmptyFallsBackToEmptyObject pins the column
+// default invariant the continuation-summary builder relies on.
+func TestUpdateRunResultJSON_EmptyFallsBackToEmptyObject(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	run := mustCreateRun(t, repo, fullRun())
+
+	if err := repo.UpdateRunResultJSON(ctx, run.ID, `{"actions":["a"]}`); err != nil {
+		t.Fatalf("update result json: %v", err)
+	}
+	checkString(t, "result_json", mustGetRun(t, repo, run.ID).ResultJSON, `{"actions":["a"]}`)
+
+	if err := repo.UpdateRunResultJSON(ctx, run.ID, ""); err != nil {
+		t.Fatalf("update result json (empty): %v", err)
+	}
+	checkString(t, "result_json after empty write", mustGetRun(t, repo, run.ID).ResultJSON, "{}")
+}
+
+// TestUpdateRunOutputSummary_WritesSummaryAndFailureReason covers the
+// completion write.
+func TestUpdateRunOutputSummary_WritesSummaryAndFailureReason(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	run := mustCreateRun(t, repo, fullRun())
+
+	if err := repo.UpdateRunOutputSummary(ctx, run.ID, "did the thing", "none"); err != nil {
+		t.Fatalf("update output summary: %v", err)
+	}
+	got := mustGetRun(t, repo, run.ID)
+	checkString(t, "output_summary", got.OutputSummary, "did the thing")
+	checkString(t, "failure_reason", got.FailureReason, "none")
+	checkString(t, "status", string(got.Status), "claimed")
+}
+
+// TestFinishRun_SetsStatusAndFinishedAt pins both columns the terminal
+// write owns.
+func TestFinishRun_SetsStatusAndFinishedAt(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	before := time.Now().UTC().Add(-time.Second)
+	run := mustCreateRun(t, repo, fullRun())
+
+	if err := repo.FinishRun(ctx, run.ID, "finished"); err != nil {
+		t.Fatalf("finish run: %v", err)
+	}
+	got := mustGetRun(t, repo, run.ID)
+	checkString(t, "status", string(got.Status), "finished")
+	if got.FinishedAt == nil {
+		t.Fatalf("finished_at = nil, want a stamp")
+	}
+	if !got.FinishedAt.After(before) {
+		t.Errorf("finished_at = %s, want after %s", got.FinishedAt, before)
+	}
+}
+
+// TestSetRunStatusForTest_WritesStatusAndBothTimestamps covers the
+// harness affordance used by the E2E seeder, including its NULL case.
+func TestSetRunStatusForTest_WritesStatusAndBothTimestamps(t *testing.T) {
+	repo := newTestRepo(t)
+	run := mustCreateRun(t, repo, fullRun())
+	claimed := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
+	finished := claimed.Add(time.Hour)
+
+	setStatus(t, repo, run.ID, "failed", timePtr(claimed), timePtr(finished))
+	got := mustGetRun(t, repo, run.ID)
+	checkString(t, "status", string(got.Status), "failed")
+	if got.ClaimedAt == nil || !got.ClaimedAt.Equal(claimed) {
+		t.Errorf("claimed_at = %v, want %s", got.ClaimedAt, claimed)
+	}
+	if got.FinishedAt == nil || !got.FinishedAt.Equal(finished) {
+		t.Errorf("finished_at = %v, want %s", got.FinishedAt, finished)
+	}
+
+	setStatus(t, repo, run.ID, "queued", nil, nil)
+	got = mustGetRun(t, repo, run.ID)
+	checkString(t, "status", string(got.Status), "queued")
+	if got.ClaimedAt != nil || got.FinishedAt != nil {
+		t.Errorf("timestamps = (%v, %v), want both nil", got.ClaimedAt, got.FinishedAt)
+	}
+}
+
+// TestSetRunRequestedAtAndErrorMessageForTest covers the two remaining
+// test affordances; the E2E ordering fixtures depend on both landing.
+func TestSetRunRequestedAtAndErrorMessageForTest(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	run := mustCreateRun(t, repo, fullRun())
+	backdated := time.Date(2025, 6, 7, 8, 9, 10, 0, time.UTC)
+
+	setRequestedAt(t, repo, run.ID, backdated)
+	if err := repo.SetRunErrorMessageForTest(ctx, run.ID, "agent exploded"); err != nil {
+		t.Fatalf("set error message: %v", err)
+	}
+
+	got := mustGetRun(t, repo, run.ID)
+	if !got.RequestedAt.Equal(backdated) {
+		t.Errorf("requested_at = %s, want %s", got.RequestedAt, backdated)
+	}
+	checkString(t, "error_message", got.ErrorMessage, "agent exploded")
+}
+
+func checkString(t *testing.T, field, got, want string) {
+	t.Helper()
+	if got != want {
+		t.Errorf("%s = %q, want %q", field, got, want)
+	}
+}
+
+func checkInt(t *testing.T, field string, got, want int) {
+	t.Helper()
+	if got != want {
+		t.Errorf("%s = %d, want %d", field, got, want)
+	}
+}
+
+func checkStringPtr(t *testing.T, field string, got, want *string) {
+	t.Helper()
+	switch {
+	case got == nil && want == nil:
+	case got == nil:
+		t.Errorf("%s = nil, want %q", field, *want)
+	case want == nil:
+		t.Errorf("%s = %q, want nil", field, *got)
+	case *got != *want:
+		t.Errorf("%s = %q, want %q", field, *got, *want)
+	}
+}

apps/backend/internal/runs/repository/sqlite/runs_list_test.go

@@ -0,0 +1,264 @@
+package sqlite_test
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/kandev/kandev/internal/office/models"
+	runssqlite "github.com/kandev/kandev/internal/runs/repository/sqlite"
+)
+
+// runIDs projects a run slice down to its IDs so ordering assertions
+// read as a single comparison.
+func runIDs(runs []*models.Run) []string {
+	out := make([]string, 0, len(runs))
+	for _, run := range runs {
+		out = append(out, run.ID)
+	}
+	return out
+}
+
+func checkIDOrder(t *testing.T, label string, got []*models.Run, want ...string) {
+	t.Helper()
+	gotIDs := runIDs(got)
+	if len(gotIDs) != len(want) {
+		t.Fatalf("%s = %v, want %v", label, gotIDs, want)
+	}
+	for i := range want {
+		if gotIDs[i] != want[i] {
+			t.Fatalf("%s = %v, want %v", label, gotIDs, want)
+		}
+	}
+}
+
+// TestListRuns_ScopesByWorkspaceAndOrdersNewestFirst pins the join and
+// the ordering: only runs whose agent profile belongs to the workspace,
+// newest first.
+func TestListRuns_ScopesByWorkspaceAndOrdersNewestFirst(t *testing.T) {
+	repo, writer := newTestRepoWithDB(t)
+	ctx := context.Background()
+	base := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC)
+
+	seedAgentProfile(t, writer, "a1", "ws-1")
+	seedAgentProfile(t, writer, "a2", "ws-1")
+	seedAgentProfile(t, writer, "a3", "ws-2")
+
+	queueRunAt(t, repo, "ws1-oldest", "a1", base)
+	queueRunAt(t, repo, "ws1-newest", "a2", base.Add(2*time.Minute))
+	queueRunAt(t, repo, "ws1-middle", "a1", base.Add(time.Minute))
+	queueRunAt(t, repo, "ws2-only", "a3", base.Add(3*time.Minute))
+	// A run whose agent profile no longer exists must not surface.
+	queueRunAt(t, repo, "orphan", "gone", base.Add(4*time.Minute))
+
+	got, err := repo.ListRuns(ctx, "ws-1")
+	if err != nil {
+		t.Fatalf("list runs: %v", err)
+	}
+	checkIDOrder(t, "ws-1 runs", got, "ws1-newest", "ws1-middle", "ws1-oldest")
+
+	got, err = repo.ListRuns(ctx, "ws-2")
+	if err != nil {
+		t.Fatalf("list runs ws-2: %v", err)
+	}
+	checkIDOrder(t, "ws-2 runs", got, "ws2-only")
+}
+
+// TestListRuns_UnknownWorkspaceReturnsEmptyNonNilSlice pins the
+// degenerate case handlers range over without a nil check.
+func TestListRuns_UnknownWorkspaceReturnsEmptyNonNilSlice(t *testing.T) {
+	repo := newTestRepo(t)
+	got, err := repo.ListRuns(context.Background(), "ws-none")
+	if err != nil {
+		t.Fatalf("list runs: %v", err)
+	}
+	if got == nil {
+		t.Fatalf("got nil slice, want an empty non-nil slice")
+	}
+	if len(got) != 0 {
+		t.Errorf("got %v, want no runs", runIDs(got))
+	}
+}
+
+// TestListRunsForAgentPaged_FirstPageIsNewestFirstAndScopedToTheAgent
+// pins ordering, the limit, and the agent filter.
+func TestListRunsForAgentPaged_FirstPageIsNewestFirstAndScopedToTheAgent(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	base := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC)
+
+	queueRunAt(t, repo, "r1", "a1", base)
+	queueRunAt(t, repo, "r2", "a1", base.Add(time.Minute))
+	queueRunAt(t, repo, "r3", "a1", base.Add(2*time.Minute))
+	queueRunAt(t, repo, "other", "a2", base.Add(3*time.Minute))
+
+	got, err := repo.ListRunsForAgentPaged(ctx, "a1", time.Time{}, "", 2)
+	if err != nil {
+		t.Fatalf("first page: %v", err)
+	}
+	checkIDOrder(t, "first page", got, "r3", "r2")
+}
+
+// TestListRunsForAgentPaged_CursorWalksTheWholeQueueExactlyOnce follows
+// the cursor contract the handler implements: feed the last row's
+// (requested_at, id) back in and get the strictly-next page.
+func TestListRunsForAgentPaged_CursorWalksTheWholeQueueExactlyOnce(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	base := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC)
+
+	for i := range 5 {
+		queueRunAt(t, repo, "r"+string(rune('0'+i)), "a1", base.Add(time.Duration(i)*time.Minute))
+	}
+
+	var seen []string
+	cursor := time.Time{}
+	cursorID := ""
+	for range 3 {
+		page, err := repo.ListRunsForAgentPaged(ctx, "a1", cursor, cursorID, 2)
+		if err != nil {
+			t.Fatalf("page: %v", err)
+		}
+		if len(page) == 0 {
+			break
+		}
+		seen = append(seen, runIDs(page)...)
+		last := page[len(page)-1]
+		cursor, cursorID = last.RequestedAt, last.ID
+	}
+
+	want := []string{"r4", "r3", "r2", "r1", "r0"}
+	if len(seen) != len(want) {
+		t.Fatalf("paged through %v, want %v", seen, want)
+	}
+	for i := range want {
+		if seen[i] != want[i] {
+			t.Fatalf("paged through %v, want %v", seen, want)
+		}
+	}
+}
+
+// TestListRunsForAgentPaged_TieBreaksOnIDWhenTimestampsMatch pins the
+// secondary sort: without it, rows sharing a requested_at could repeat
+// or vanish across pages.
+func TestListRunsForAgentPaged_TieBreaksOnIDWhenTimestampsMatch(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	same := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC)
+
+	queueRunAt(t, repo, "run-a", "a1", same)
+	queueRunAt(t, repo, "run-b", "a1", same)
+	queueRunAt(t, repo, "run-c", "a1", same)
+
+	first, err := repo.ListRunsForAgentPaged(ctx, "a1", time.Time{}, "", 2)
+	if err != nil {
+		t.Fatalf("first page: %v", err)
+	}
+	checkIDOrder(t, "first page", first, "run-c", "run-b")
+
+	last := first[len(first)-1]
+	next, err := repo.ListRunsForAgentPaged(ctx, "a1", last.RequestedAt, last.ID, 2)
+	if err != nil {
+		t.Fatalf("second page: %v", err)
+	}
+	checkIDOrder(t, "second page", next, "run-a")
+}
+
+// TestListRunsForAgentPaged_NonPositiveLimitFallsBackTo25 pins the
+// default page size.
+func TestListRunsForAgentPaged_NonPositiveLimitFallsBackTo25(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	base := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC)
+
+	const seeded = 27
+	for i := range seeded {
+		queueRunAt(t, repo, "r"+padded(i), "a1", base.Add(time.Duration(i)*time.Minute))
+	}
+
+	for _, limit := range []int{0, -5} {
+		got, err := repo.ListRunsForAgentPaged(ctx, "a1", time.Time{}, "", limit)
+		if err != nil {
+			t.Fatalf("list with limit %d: %v", limit, err)
+		}
+		if len(got) != 25 {
+			t.Errorf("limit %d returned %d rows, want the 25-row default", limit, len(got))
+		}
+		if len(got) > 0 && got[0].ID != "r"+padded(seeded-1) {
+			t.Errorf("limit %d returned %q first, want the newest run", limit, got[0].ID)
+		}
+	}
+}
+
+// TestListRunsForAgentPaged_UnknownAgentReturnsEmptyNonNilSlice pins the
+// degenerate case.
+func TestListRunsForAgentPaged_UnknownAgentReturnsEmptyNonNilSlice(t *testing.T) {
+	repo := newTestRepo(t)
+	got, err := repo.ListRunsForAgentPaged(context.Background(), "nobody", time.Time{}, "", 10)
+	if err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	if got == nil || len(got) != 0 {
+		t.Errorf("got %v, want an empty non-nil slice", runIDs(got))
+	}
+}
+
+// TestListPendingRunsForTask_OnlyQueuedRetriesForThatTask pins all
+// three predicates: queued, a scheduled retry, and the task id inside
+// the payload.
+func TestListPendingRunsForTask_OnlyQueuedRetriesForThatTask(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	now := time.Now().UTC()
+
+	pending := seedTaskRun(t, repo, "pending", "a1", "t1", "queued")
+	if err := repo.ScheduleRetry(ctx, pending.ID, now.Add(time.Hour), 1); err != nil {
+		t.Fatalf("schedule retry: %v", err)
+	}
+	seedTaskRun(t, repo, "queued-no-retry", "a1", "t1", "queued")
+	claimedRetry := seedTaskRun(t, repo, "claimed-retry", "a1", "t1", "queued")
+	if err := repo.ScheduleRetry(ctx, claimedRetry.ID, now.Add(time.Hour), 1); err != nil {
+		t.Fatalf("schedule retry (claimed): %v", err)
+	}
+	setStatus(t, repo, claimedRetry.ID, "claimed", timePtr(now), nil)
+	otherTask := seedTaskRun(t, repo, "other-task", "a1", "t2", "queued")
+	if err := repo.ScheduleRetry(ctx, otherTask.ID, now.Add(time.Hour), 1); err != nil {
+		t.Fatalf("schedule retry (other task): %v", err)
+	}
+
+	got, err := repo.ListPendingRunsForTask(ctx, "t1")
+	if err != nil {
+		t.Fatalf("list pending: %v", err)
+	}
+	checkIDOrder(t, "pending runs for t1", got, "pending")
+
+	empty, err := repo.ListPendingRunsForTask(ctx, "t-unknown")
+	if err != nil {
+		t.Fatalf("list pending (unknown task): %v", err)
+	}
+	if empty == nil || len(empty) != 0 {
+		t.Errorf("got %v, want an empty non-nil slice", runIDs(empty))
+	}
+}
+
+// seedTaskRun creates a run carrying a task id in its payload.
+func seedTaskRun(
+	t *testing.T, repo *runssqlite.Repository, id, agentID, taskID, status string,
+) *models.Run {
+	t.Helper()
+	return mustCreateRun(t, repo, &models.Run{
+		ID:             id,
+		AgentProfileID: agentID,
+		Reason:         "task_assigned",
+		Payload:        `{"task_id":"` + taskID + `"}`,
+		Status:         models.RunStatus(status),
+		CoalescedCount: 1,
+	})
+}
+
+// padded keeps seeded run IDs sortable as strings so an ID tie-break
+// cannot masquerade as a timestamp ordering.
+func padded(i int) string {
+	return fmt.Sprintf("%02d", i)
+}

apps/backend/internal/runs/repository/sqlite/runs_lookup_test.go

@@ -0,0 +1,228 @@
+package sqlite_test
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/kandev/kandev/internal/office/models"
+)
+
+// TestFindInflightRunForAgent_ReturnsTheNewestQueuedOrClaimedRun pins
+// both the in-flight status set and the newest-first ordering.
+func TestFindInflightRunForAgent_ReturnsTheNewestQueuedOrClaimedRun(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	base := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC)
+
+	queueRunAt(t, repo, "older-queued", "a1", base)
+	newest := queueRunAt(t, repo, "newer-claimed", "a1", base.Add(time.Minute))
+	setStatus(t, repo, newest.ID, "claimed", timePtr(base.Add(time.Minute)), nil)
+	// Newer still, but terminal — must not be reported as in-flight.
+	terminal := queueRunAt(t, repo, "newest-finished", "a1", base.Add(2*time.Minute))
+	setStatus(t, repo, terminal.ID, "finished", nil, timePtr(base.Add(3*time.Minute)))
+	// Newest of all, but a different agent.
+	queueRunAt(t, repo, "other-agent", "a2", base.Add(4*time.Minute))
+
+	got, err := repo.FindInflightRunForAgent(ctx, "a1")
+	if err != nil {
+		t.Fatalf("find inflight: %v", err)
+	}
+	if got.ID != newest.ID {
+		t.Errorf("inflight = %q, want %q", got.ID, newest.ID)
+	}
+	checkString(t, "status", string(got.Status), "claimed")
+}
+
+// TestFindInflightRunForAgent_TerminalRunsAreNotInflight covers the
+// negative half: every terminal status must yield the sentinel.
+func TestFindInflightRunForAgent_TerminalRunsAreNotInflight(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	now := time.Now().UTC()
+
+	for _, status := range []string{"finished", "failed", "cancelled"} {
+		run := queueRunAt(t, repo, "run-"+status, "a1", now)
+		setStatus(t, repo, run.ID, status, timePtr(now), timePtr(now))
+	}
+
+	if _, err := repo.FindInflightRunForAgent(ctx, "a1"); !errors.Is(err, sql.ErrNoRows) {
+		t.Fatalf("err = %v, want sql.ErrNoRows", err)
+	}
+}
+
+// TestGetClaimedRunByTaskID_MatchesThePayloadTaskAndPrefersTheLatestClaim
+// pins the payload lookup plus the claimed_at ordering.
+func TestGetClaimedRunByTaskID_MatchesThePayloadTaskAndPrefersTheLatestClaim(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	base := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC)
+
+	older := seedTaskRun(t, repo, "older-claim", "a1", "t1", "queued")
+	setStatus(t, repo, older.ID, "claimed", timePtr(base), nil)
+	newer := seedTaskRun(t, repo, "newer-claim", "a2", "t1", "queued")
+	setStatus(t, repo, newer.ID, "claimed", timePtr(base.Add(time.Hour)), nil)
+	// Same task but still queued, and another task's claimed run.
+	seedTaskRun(t, repo, "queued-same-task", "a3", "t1", "queued")
+	otherTask := seedTaskRun(t, repo, "other-task", "a4", "t2", "queued")
+	setStatus(t, repo, otherTask.ID, "claimed", timePtr(base.Add(2*time.Hour)), nil)
+
+	got, err := repo.GetClaimedRunByTaskID(ctx, "t1")
+	if err != nil {
+		t.Fatalf("get claimed run: %v", err)
+	}
+	if got.ID != newer.ID {
+		t.Errorf("claimed run = %q, want %q (most recent claim for t1)", got.ID, newer.ID)
+	}
+
+	if _, err := repo.GetClaimedRunByTaskID(ctx, "t-unknown"); !errors.Is(err, sql.ErrNoRows) {
+		t.Errorf("unknown task err = %v, want sql.ErrNoRows", err)
+	}
+}
+
+// TestGetClaimedTasklessRunForAgent_OnlyMatchesRunsWithoutATask pins the
+// heartbeat attribution rule: a missing or empty payload task_id counts
+// as taskless, a populated one never does.
+func TestGetClaimedTasklessRunForAgent_OnlyMatchesRunsWithoutATask(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	base := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC)
+
+	missing := mustCreateRun(t, repo, &models.Run{
+		ID: "missing-task-id", AgentProfileID: "a1", Reason: "heartbeat",
+		Payload: `{"note":"no task"}`,
+	})
+	setStatus(t, repo, missing.ID, "claimed", timePtr(base), nil)
+	empty := mustCreateRun(t, repo, &models.Run{
+		ID: "empty-task-id", AgentProfileID: "a1", Reason: "heartbeat",
+		Payload: `{"task_id":""}`,
+	})
+	setStatus(t, repo, empty.ID, "claimed", timePtr(base.Add(time.Hour)), nil)
+	// Newest claim, but it carries a task id.
+	withTask := seedTaskRun(t, repo, "with-task", "a1", "t1", "queued")
+	setStatus(t, repo, withTask.ID, "claimed", timePtr(base.Add(2*time.Hour)), nil)
+
+	got, err := repo.GetClaimedTasklessRunForAgent(ctx, "a1")
+	if err != nil {
+		t.Fatalf("get taskless run: %v", err)
+	}
+	if got.ID != empty.ID {
+		t.Errorf("taskless run = %q, want %q (most recent taskless claim)", got.ID, empty.ID)
+	}
+}
+
+// TestGetClaimedTasklessRunForAgent_ScopedToTheAgentAndToClaimedRuns
+// covers the two remaining predicates.
+func TestGetClaimedTasklessRunForAgent_ScopedToTheAgentAndToClaimedRuns(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	now := time.Now().UTC()
+
+	queued := mustCreateRun(t, repo, &models.Run{
+		ID: "queued", AgentProfileID: "a1", Reason: "heartbeat",
+	})
+	otherAgent := mustCreateRun(t, repo, &models.Run{
+		ID: "other-agent", AgentProfileID: "a2", Reason: "heartbeat",
+	})
+	setStatus(t, repo, otherAgent.ID, "claimed", timePtr(now), nil)
+
+	if _, err := repo.GetClaimedTasklessRunForAgent(ctx, "a1"); !errors.Is(err, sql.ErrNoRows) {
+		t.Fatalf("err = %v, want sql.ErrNoRows (a1's only run is queued)", err)
+	}
+
+	setStatus(t, repo, queued.ID, "claimed", timePtr(now), nil)
+	got, err := repo.GetClaimedTasklessRunForAgent(ctx, "a1")
+	if err != nil {
+		t.Fatalf("get taskless run: %v", err)
+	}
+	if got.ID != queued.ID {
+		t.Errorf("taskless run = %q, want %q", got.ID, queued.ID)
+	}
+}
+
+// TestGetRunWithCosts_SumsOnlyTheRunsOwnTaskEvents pins the rollup
+// join: cost events are matched on the task id inside the run payload.
+func TestGetRunWithCosts_SumsOnlyTheRunsOwnTaskEvents(t *testing.T) {
+	repo, writer := newTestRepoWithDB(t)
+	ctx := context.Background()
+
+	run := seedTaskRun(t, repo, "costed", "a1", "t1", "claimed")
+	seedCostEvent(t, writer, "c1", "t1", 100, 10, 20, 5)
+	seedCostEvent(t, writer, "c2", "t1", 200, 20, 30, 7)
+	seedCostEvent(t, writer, "c3", "t2", 999, 999, 999, 999)
+
+	got, rollup, err := repo.GetRunWithCosts(ctx, run.ID)
+	if err != nil {
+		t.Fatalf("get run with costs: %v", err)
+	}
+	if got.ID != run.ID {
+		t.Errorf("run = %q, want %q", got.ID, run.ID)
+	}
+	checkInt64(t, "input_tokens", rollup.InputTokens, 300)
+	checkInt64(t, "cached_tokens", rollup.CachedTokens, 30)
+	checkInt64(t, "output_tokens", rollup.OutputTokens, 50)
+	checkInt64(t, "cost_subcents", rollup.CostSubcents, 12)
+}
+
+// TestGetRunWithCosts_TasklessRunRollsUpToZero pins the non-empty
+// task_id guard: a run without a task must not absorb the cost events
+// that carry no task id either.
+func TestGetRunWithCosts_TasklessRunRollsUpToZero(t *testing.T) {
+	repo, writer := newTestRepoWithDB(t)
+	ctx := context.Background()
+
+	run := mustCreateRun(t, repo, &models.Run{
+		ID: "taskless", AgentProfileID: "a1", Reason: "heartbeat",
+	})
+	seedCostEvent(t, writer, "orphan", "", 500, 50, 60, 70)
+
+	got, rollup, err := repo.GetRunWithCosts(ctx, run.ID)
+	if err != nil {
+		t.Fatalf("get run with costs: %v", err)
+	}
+	if got.ID != run.ID {
+		t.Errorf("run = %q, want %q", got.ID, run.ID)
+	}
+	checkInt64(t, "input_tokens", rollup.InputTokens, 0)
+	checkInt64(t, "cached_tokens", rollup.CachedTokens, 0)
+	checkInt64(t, "output_tokens", rollup.OutputTokens, 0)
+	checkInt64(t, "cost_subcents", rollup.CostSubcents, 0)
+}
+
+// TestGetRunWithCosts_NoCostEventsYieldsAZeroRollup pins the documented
+// "run exists, costs not produced yet" shape.
+func TestGetRunWithCosts_NoCostEventsYieldsAZeroRollup(t *testing.T) {
+	repo := newTestRepo(t)
+	run := seedTaskRun(t, repo, "no-costs", "a1", "t1", "queued")
+
+	got, rollup, err := repo.GetRunWithCosts(context.Background(), run.ID)
+	if err != nil {
+		t.Fatalf("get run with costs: %v", err)
+	}
+	if got == nil || rollup == nil {
+		t.Fatalf("run = %v, rollup = %v, want both non-nil", got, rollup)
+	}
+	checkInt64(t, "input_tokens", rollup.InputTokens, 0)
+	checkInt64(t, "cost_subcents", rollup.CostSubcents, 0)
+}
+
+// TestGetRunWithCosts_UnknownRunReturnsNoRows pins the sentinel.
+func TestGetRunWithCosts_UnknownRunReturnsNoRows(t *testing.T) {
+	repo := newTestRepo(t)
+	run, rollup, err := repo.GetRunWithCosts(context.Background(), "nope")
+	if !errors.Is(err, sql.ErrNoRows) {
+		t.Fatalf("err = %v, want sql.ErrNoRows", err)
+	}
+	if run != nil || rollup != nil {
+		t.Errorf("run = %v, rollup = %v, want both nil on error", run, rollup)
+	}
+}
+
+func checkInt64(t *testing.T, field string, got, want int64) {
+	t.Helper()
+	if got != want {
+		t.Errorf("%s = %d, want %d", field, got, want)
+	}
+}

apps/backend/internal/runs/repository/sqlite/runs_queue_test.go

@@ -0,0 +1,403 @@
+package sqlite_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/kandev/kandev/internal/office/models"
+	runssqlite "github.com/kandev/kandev/internal/runs/repository/sqlite"
+)
+
+// countRuns is the row-count assertion that proves a coalesce merged
+// instead of inserting.
+func countRuns(t *testing.T, repo *runssqlite.Repository) int {
+	t.Helper()
+	var got int
+	if err := repo.Reader().Get(&got, `SELECT COUNT(*) FROM runs`); err != nil {
+		t.Fatalf("count runs: %v", err)
+	}
+	return got
+}
+
+// TestCoalesceRun_MergesIntoMostRecentQueuedRun asserts the identity of
+// the row that absorbed the request: the newest queued run for the same
+// agent + reason, with its payload replaced and its counter bumped, and
+// no second row created.
+func TestCoalesceRun_MergesIntoMostRecentQueuedRun(t *testing.T) {
+	repo := newTestRepo(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))
+
+	merged, err := repo.CoalesceRun(ctx, "a1", "task_assigned", 3600, `{"merged":true}`)
+	if err != nil {
+		t.Fatalf("coalesce: %v", err)
+	}
+	if !merged {
+		t.Fatalf("coalesce = false, want true")
+	}
+
+	got := mustGetRun(t, repo, newer.ID)
+	checkInt(t, "coalesced_count", got.CoalescedCount, 2)
+	checkString(t, "payload", got.Payload, `{"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"}`)
+
+	if n := countRuns(t, repo); n != 2 {
+		t.Errorf("%d rows after coalescing, want 2 (no new run may be created)", n)
+	}
+}
+
+// TestCoalesceRun_RepeatedCoalesceAccumulatesOnOneRow pins the counter
+// arithmetic across several merges into the same run.
+func TestCoalesceRun_RepeatedCoalesceAccumulatesOnOneRow(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	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}`)
+		if err != nil {
+			t.Fatalf("coalesce %d: %v", i, err)
+		}
+		if !merged {
+			t.Fatalf("coalesce %d = false, want true", i)
+		}
+	}
+
+	checkInt(t, "coalesced_count", mustGetRun(t, repo, run.ID).CoalescedCount, 4)
+	if n := countRuns(t, repo); n != 1 {
+		t.Errorf("%d rows, want 1", n)
+	}
+}
+
+// TestCoalesceRun_NoEligibleRunLeavesEverythingAlone covers the three
+// selector dimensions that disqualify a candidate.
+func TestCoalesceRun_NoEligibleRunLeavesEverythingAlone(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	now := time.Now().UTC()
+
+	sameAgent := queueRunAt(t, repo, "other-reason", "a1", now.Add(-time.Second))
+	otherAgent := queueRunAt(t, repo, "other-agent", "a2", now.Add(-time.Second))
+	claimed := queueRunAt(t, repo, "claimed", "a1", now.Add(-time.Second))
+	setStatus(t, repo, claimed.ID, "claimed", timePtr(now), nil)
+
+	merged, err := repo.CoalesceRun(ctx, "a1", "heartbeat", 3600, `{"merged":true}`)
+	if err != nil {
+		t.Fatalf("coalesce: %v", err)
+	}
+	if merged {
+		t.Fatalf("coalesce = true, want false (no queued run with reason=heartbeat for a1)")
+	}
+	for _, id := range []string{sameAgent.ID, otherAgent.ID, claimed.ID} {
+		got := mustGetRun(t, repo, id)
+		checkInt(t, "coalesced_count of "+id, got.CoalescedCount, 1)
+		checkString(t, "payload of "+id, got.Payload, `{"task_id":"`+id+`"}`)
+	}
+}
+
+// TestCoalesceRun_HonoursTheWindow pins the requested_at cutoff: the
+// same run is eligible under a wide window and ineligible under a
+// narrow one.
+func TestCoalesceRun_HonoursTheWindow(t *testing.T) {
+	repo := newTestRepo(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}`)
+	if err != nil {
+		t.Fatalf("coalesce (narrow window): %v", err)
+	}
+	if merged {
+		t.Fatalf("coalesce = true for a run older than the window, want false")
+	}
+	checkInt(t, "coalesced_count", mustGetRun(t, repo, run.ID).CoalescedCount, 1)
+
+	merged, err = repo.CoalesceRun(ctx, "a1", "task_assigned", 600, `{"merged":true}`)
+	if err != nil {
+		t.Fatalf("coalesce (wide window): %v", err)
+	}
+	if !merged {
+		t.Fatalf("coalesce = false inside the window, want true")
+	}
+	checkInt(t, "coalesced_count", mustGetRun(t, repo, run.ID).CoalescedCount, 2)
+}
+
+// TestCoalesceRun_SkipsCommentKeyedRuns pins the exclusion that keeps a
+// comment-triggered run addressable by its own idempotency key: a
+// salted or canonical task_comment run must never absorb an unrelated
+// wake, while a key-less run in the same window still does.
+func TestCoalesceRun_SkipsCommentKeyedRuns(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	now := time.Now().UTC()
+
+	plain := mustCreateRun(t, repo, &models.Run{
+		ID: "plain", AgentProfileID: "a1", Reason: "task_comment",
+		Payload: `{"task_id":"t1"}`, Status: "queued", CoalescedCount: 1,
+	})
+	setRequestedAt(t, repo, plain.ID, now.Add(-20*time.Second))
+	keyed := mustCreateRun(t, repo, &models.Run{
+		ID: "keyed", AgentProfileID: "a1", Reason: "task_comment",
+		Payload: `{"task_id":"t1","comment_id":"cm-1"}`, Status: "queued",
+		CoalescedCount: 1, IdempotencyKey: strPtr("task_comment:cm-1"),
+	})
+	setRequestedAt(t, repo, keyed.ID, now.Add(-5*time.Second))
+
+	merged, err := repo.CoalesceRun(ctx, "a1", "task_comment", 600, `{"merged":true}`)
+	if err != nil {
+		t.Fatalf("coalesce: %v", err)
+	}
+	if !merged {
+		t.Fatalf("coalesce = false, want true (the key-less run is eligible)")
+	}
+	checkInt(t, "comment-keyed coalesced_count", mustGetRun(t, repo, keyed.ID).CoalescedCount, 1)
+	checkString(t, "comment-keyed payload", mustGetRun(t, repo, keyed.ID).Payload,
+		`{"task_id":"t1","comment_id":"cm-1"}`)
+	checkInt(t, "key-less coalesced_count", mustGetRun(t, repo, plain.ID).CoalescedCount, 2)
+}
+
+// TestCheckIdempotencyKey_WindowEdges pins both sides of the cutoff for
+// the same key and row.
+func TestCheckIdempotencyKey_WindowEdges(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	run := mustCreateRun(t, repo, &models.Run{
+		ID: "idem", AgentProfileID: "a1", Reason: "task_assigned",
+		IdempotencyKey: strPtr("wake:t1"),
+	})
+
+	// Just inside a one-hour window.
+	setRequestedAt(t, repo, run.ID, time.Now().UTC().Add(-time.Hour+time.Minute))
+	seen, err := repo.CheckIdempotencyKey(ctx, "wake:t1", 1)
+	if err != nil {
+		t.Fatalf("check (inside window): %v", err)
+	}
+	if !seen {
+		t.Errorf("seen = false for a run inside the window, want true")
+	}
+
+	// Just outside the same window; still inside a wider one.
+	setRequestedAt(t, repo, run.ID, time.Now().UTC().Add(-time.Hour-time.Minute))
+	seen, err = repo.CheckIdempotencyKey(ctx, "wake:t1", 1)
+	if err != nil {
+		t.Fatalf("check (outside window): %v", err)
+	}
+	if seen {
+		t.Errorf("seen = true for a run older than the window, want false")
+	}
+	seen, err = repo.CheckIdempotencyKey(ctx, "wake:t1", 24)
+	if err != nil {
+		t.Fatalf("check (wide window): %v", err)
+	}
+	if !seen {
+		t.Errorf("seen = false under a 24h window, want true")
+	}
+}
+
+// TestCheckIdempotencyKey_MatchesOnTheExactKey guards against a
+// weakened comparison (prefix/LIKE) letting a neighbouring key satisfy
+// the check.
+func TestCheckIdempotencyKey_MatchesOnTheExactKey(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	mustCreateRun(t, repo, &models.Run{
+		ID: "idem", AgentProfileID: "a1", Reason: "task_assigned",
+		IdempotencyKey: strPtr("wake:t1"),
+	})
+	mustCreateRun(t, repo, &models.Run{
+		ID: "no-key", AgentProfileID: "a1", Reason: "task_assigned",
+	})
+
+	for _, key := range []string{"wake:t2", "wake:t1:salt", "wake:t", ""} {
+		seen, err := repo.CheckIdempotencyKey(ctx, key, 24)
+		if err != nil {
+			t.Fatalf("check %q: %v", key, err)
+		}
+		if seen {
+			t.Errorf("seen = true for key %q, want false", key)
+		}
+	}
+}
+
+// TestCreateRun_DuplicateIdempotencyKeyIsRejected pins the storage-level
+// half of the idempotency contract: the same key twice cannot produce a
+// second run, and the original row is the one that survives.
+func TestCreateRun_DuplicateIdempotencyKeyIsRejected(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	first := mustCreateRun(t, repo, &models.Run{
+		ID: "first", AgentProfileID: "a1", Reason: "task_comment",
+		Payload: `{"task_id":"t1"}`, IdempotencyKey: strPtr("task_comment:cm-1"),
+	})
+
+	err := repo.CreateRun(ctx, &models.Run{
+		ID: "second", AgentProfileID: "a1", Reason: "task_comment",
+		Payload: `{"task_id":"t2"}`, IdempotencyKey: strPtr("task_comment:cm-1"),
+	})
+	if err == nil {
+		t.Fatalf("second CreateRun with a duplicate idempotency key succeeded, want a constraint error")
+	}
+	if n := countRuns(t, repo); n != 1 {
+		t.Errorf("%d rows after the duplicate insert, want 1", n)
+	}
+	checkString(t, "surviving run payload", mustGetRun(t, repo, first.ID).Payload, `{"task_id":"t1"}`)
+}
+
+// TestScheduleRetry_RequeuesAndClearsTerminalTimestamps asserts every
+// column the retry write owns, from a run that had already finished.
+func TestScheduleRetry_RequeuesAndClearsTerminalTimestamps(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	now := time.Now().UTC()
+	run := queueRunAt(t, repo, "retryable", "a1", now.Add(-time.Hour))
+	setStatus(t, repo, run.ID, "failed", timePtr(now.Add(-30*time.Minute)), timePtr(now))
+	untouched := queueRunAt(t, repo, "bystander", "a1", now.Add(-time.Hour))
+
+	retryAt := time.Date(2026, 7, 8, 9, 10, 11, 0, time.UTC)
+	if err := repo.ScheduleRetry(ctx, run.ID, retryAt, 4); err != nil {
+		t.Fatalf("schedule retry: %v", err)
+	}
+
+	got := mustGetRun(t, repo, run.ID)
+	checkString(t, "status", string(got.Status), "queued")
+	checkInt(t, "retry_count", got.RetryCount, 4)
+	if got.ScheduledRetryAt == nil || !got.ScheduledRetryAt.Equal(retryAt) {
+		t.Errorf("scheduled_retry_at = %v, want %s", got.ScheduledRetryAt, retryAt)
+	}
+	if got.ClaimedAt != nil {
+		t.Errorf("claimed_at = %s, want nil", got.ClaimedAt)
+	}
+	if got.FinishedAt != nil {
+		t.Errorf("finished_at = %s, want nil", got.FinishedAt)
+	}
+	checkString(t, "payload preserved", got.Payload, `{"task_id":"retryable"}`)
+
+	bystander := mustGetRun(t, repo, untouched.ID)
+	checkInt(t, "bystander retry_count", bystander.RetryCount, 0)
+	if bystander.ScheduledRetryAt != nil {
+		t.Errorf("bystander scheduled_retry_at = %s, want nil", bystander.ScheduledRetryAt)
+	}
+}
+
+// TestScheduleRetry_WritesTheAttemptCountVerbatim pins that the caller
+// owns the counter — the repository must not derive or clamp it.
+func TestScheduleRetry_WritesTheAttemptCountVerbatim(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	run := queueRunAt(t, repo, "counter", "a1", time.Now().UTC())
+
+	for _, count := range []int{1, 5, 0} {
+		if err := repo.ScheduleRetry(ctx, run.ID, fixedRetryAt, count); err != nil {
+			t.Fatalf("schedule retry %d: %v", count, err)
+		}
+		checkInt(t, "retry_count", mustGetRun(t, repo, run.ID).RetryCount, count)
+	}
+}
+
+// TestRecoverStale_ResetsOnlyClaimedRunsStrictlyOlderThanTheCutoff
+// exercises the staleness boundary in both directions. The cutoff is a
+// parameter, so the equality case is exact rather than clock-dependent.
+func TestRecoverStale_ResetsOnlyClaimedRunsStrictlyOlderThanTheCutoff(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	cutoff := time.Date(2026, 4, 1, 12, 0, 0, 0, time.UTC)
+
+	stale := queueRunAt(t, repo, "stale", "a1", cutoff.Add(-time.Hour))
+	setStatus(t, repo, stale.ID, "claimed", timePtr(cutoff.Add(-time.Minute)), nil)
+	atCutoff := queueRunAt(t, repo, "at-cutoff", "a2", cutoff.Add(-time.Hour))
+	setStatus(t, repo, atCutoff.ID, "claimed", timePtr(cutoff), nil)
+	fresh := queueRunAt(t, repo, "fresh", "a3", cutoff.Add(-time.Hour))
+	setStatus(t, repo, fresh.ID, "claimed", timePtr(cutoff.Add(time.Minute)), nil)
+	finished := queueRunAt(t, repo, "finished", "a4", cutoff.Add(-time.Hour))
+	setStatus(t, repo, finished.ID, "finished",
+		timePtr(cutoff.Add(-time.Hour)), timePtr(cutoff.Add(-time.Minute)))
+
+	recovered, err := repo.RecoverStale(ctx, cutoff)
+	if err != nil {
+		t.Fatalf("recover stale: %v", err)
+	}
+	if recovered != 1 {
+		t.Errorf("recovered = %d, want 1", recovered)
+	}
+
+	got := mustGetRun(t, repo, stale.ID)
+	checkString(t, "stale status", string(got.Status), "queued")
+	if got.ClaimedAt != nil {
+		t.Errorf("stale claimed_at = %s, want nil", got.ClaimedAt)
+	}
+	checkString(t, "at-cutoff status", string(mustGetRun(t, repo, atCutoff.ID).Status), "claimed")
+	checkString(t, "fresh status", string(mustGetRun(t, repo, fresh.ID).Status), "claimed")
+	checkString(t, "finished status", string(mustGetRun(t, repo, finished.ID).Status), "finished")
+	if got := mustGetRun(t, repo, finished.ID); got.ClaimedAt == nil {
+		t.Errorf("finished run lost its claimed_at, want it preserved")
+	}
+}
+
+// TestRecoverStale_NothingToRecoverReportsZero pins the count when the
+// sweep is a no-op.
+func TestRecoverStale_NothingToRecoverReportsZero(t *testing.T) {
+	repo := newTestRepo(t)
+	queueRunAt(t, repo, "queued", "a1", time.Now().UTC())
+
+	recovered, err := repo.RecoverStale(context.Background(), time.Now().UTC())
+	if err != nil {
+		t.Fatalf("recover stale: %v", err)
+	}
+	if recovered != 0 {
+		t.Errorf("recovered = %d, want 0", recovered)
+	}
+}
+
+// TestCleanExpired_DeletesOnlyTerminalRunsBeforeTheCutoff covers the
+// status filter, the strict finished_at comparison, and the returned
+// row count. Cancelled runs are deliberately not swept.
+func TestCleanExpired_DeletesOnlyTerminalRunsBeforeTheCutoff(t *testing.T) {
+	repo := newTestRepo(t)
+	ctx := context.Background()
+	cutoff := time.Date(2026, 4, 1, 12, 0, 0, 0, time.UTC)
+	old := cutoff.Add(-time.Hour)
+
+	seedTerminal(t, repo, "old-finished", "finished", timePtr(old))
+	seedTerminal(t, repo, "old-failed", "failed", timePtr(old))
+	seedTerminal(t, repo, "at-cutoff", "finished", timePtr(cutoff))
+	seedTerminal(t, repo, "new-finished", "finished", timePtr(cutoff.Add(time.Hour)))
+	seedTerminal(t, repo, "old-cancelled", "cancelled", timePtr(old))
+	seedTerminal(t, repo, "no-finished-at", "finished", nil)
+	queueRunAt(t, repo, "queued", "a9", old)
+
+	deleted, err := repo.CleanExpired(ctx, cutoff)
+	if err != nil {
+		t.Fatalf("clean expired: %v", err)
+	}
+	if deleted != 2 {
+		t.Errorf("deleted = %d, want 2", deleted)
+	}
+
+	survivors := []string{"at-cutoff", "new-finished", "old-cancelled", "no-finished-at", "queued"}
+	for _, id := range survivors {
+		if _, err := repo.GetRunByID(ctx, id); err != nil {
+			t.Errorf("run %q was deleted (%v), want it kept", id, err)
+		}
+	}
+	if n := countRuns(t, repo); n != len(survivors) {
+		t.Errorf("%d rows left, want %d", n, len(survivors))
+	}
+}
+
+// seedTerminal creates a run already in a terminal state with the given
+// finished_at, bypassing the state machine the way the sweeper's
+// fixtures do.
+func seedTerminal(
+	t *testing.T, repo *runssqlite.Repository, id, status string, finishedAt *time.Time,
+) {
+	t.Helper()
+	run := queueRunAt(t, repo, id, "a1", time.Date(2026, 3, 1, 0, 0, 0, 0, time.UTC))
+	setStatus(t, repo, run.ID, status, nil, finishedAt)
+}