diff --git a/apps/backend/internal/runs/repository/sqlite/base_test.go b/apps/backend/internal/runs/repository/sqlite/base_test.go new file mode 100644 index 0000000000..59db04d901 --- /dev/null +++ b/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") + } +} diff --git a/apps/backend/internal/runs/repository/sqlite/cancel_test.go b/apps/backend/internal/runs/repository/sqlite/cancel_test.go new file mode 100644 index 0000000000..dd3b251080 --- /dev/null +++ b/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) + } +} diff --git a/apps/backend/internal/runs/repository/sqlite/errors_test.go b/apps/backend/internal/runs/repository/sqlite/errors_test.go new file mode 100644 index 0000000000..c93811d908 --- /dev/null +++ b/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) +} diff --git a/apps/backend/internal/runs/repository/sqlite/run_events_test.go b/apps/backend/internal/runs/repository/sqlite/run_events_test.go new file mode 100644 index 0000000000..6379261179 --- /dev/null +++ b/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) + } + } +} diff --git a/apps/backend/internal/runs/repository/sqlite/runs_claim_test.go b/apps/backend/internal/runs/repository/sqlite/runs_claim_test.go new file mode 100644 index 0000000000..a3da8e6cb9 --- /dev/null +++ b/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) + } + } +} diff --git a/apps/backend/internal/runs/repository/sqlite/runs_comments_test.go b/apps/backend/internal/runs/repository/sqlite/runs_comments_test.go new file mode 100644 index 0000000000..f3f370e6ef --- /dev/null +++ b/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) + } +} diff --git a/apps/backend/internal/runs/repository/sqlite/runs_crud_test.go b/apps/backend/internal/runs/repository/sqlite/runs_crud_test.go new file mode 100644 index 0000000000..7c1f026bb5 --- /dev/null +++ b/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) + } +} diff --git a/apps/backend/internal/runs/repository/sqlite/runs_list_test.go b/apps/backend/internal/runs/repository/sqlite/runs_list_test.go new file mode 100644 index 0000000000..1c4201aa11 --- /dev/null +++ b/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) +} diff --git a/apps/backend/internal/runs/repository/sqlite/runs_lookup_test.go b/apps/backend/internal/runs/repository/sqlite/runs_lookup_test.go new file mode 100644 index 0000000000..6ebfa28094 --- /dev/null +++ b/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) + } +} diff --git a/apps/backend/internal/runs/repository/sqlite/runs_queue_test.go b/apps/backend/internal/runs/repository/sqlite/runs_queue_test.go new file mode 100644 index 0000000000..a60a5c1812 --- /dev/null +++ b/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) +}