diff --git a/apps/backend/internal/analytics/models/models.go b/apps/backend/internal/analytics/models/models.go index 459236e0d4..c5dfd903b9 100644 --- a/apps/backend/internal/analytics/models/models.go +++ b/apps/backend/internal/analytics/models/models.go @@ -102,10 +102,17 @@ type GitStats struct { // wants a single "effective lines" number (the larger of the two, to avoid // double-counting work that was later committed) computes that itself, the // same way the source plugin's effectiveLines helper does. +// +// LinesAddedCommitted / LinesDeletedCommitted are nil, not 0, for a session +// that predates commit-capture activation (see commit_capture_activated_at in +// kandev_meta) and has no observed commits: capture wasn't running yet for +// that session, so "zero" cannot be distinguished from "unknown" and must not +// be asserted as a real measurement. A session with at least one observed +// commit, or one that started after activation, always gets a real int64. type SessionCodeStats struct { SessionID string `json:"session_id"` - LinesAddedCommitted int64 `json:"lines_added_committed"` - LinesDeletedCommitted int64 `json:"lines_deleted_committed"` + LinesAddedCommitted *int64 `json:"lines_added_committed"` + LinesDeletedCommitted *int64 `json:"lines_deleted_committed"` LinesAddedPeakPending int64 `json:"lines_added_peak_pending"` LinesDeletedPeakPending int64 `json:"lines_deleted_peak_pending"` } diff --git a/apps/backend/internal/analytics/repository/sqlite/commit_capture_activation_integration_test.go b/apps/backend/internal/analytics/repository/sqlite/commit_capture_activation_integration_test.go new file mode 100644 index 0000000000..d912b37cc7 --- /dev/null +++ b/apps/backend/internal/analytics/repository/sqlite/commit_capture_activation_integration_test.go @@ -0,0 +1,140 @@ +package sqlite + +// Cross-package integration coverage for the commit_capture_activated_at +// legacy/NULL comparison in ListSessionCodeStats. The rest of this package's +// tests build task_sessions.started_at and kandev_meta rows by hand (raw +// RFC3339 strings via execOrFatal), which never exercises the actual text +// format task/repository/sqlite writes in production: started_at is bound as +// a Go time.Time through the sqlite3 driver, and commit_capture_activated_at +// is written by migrateSessionCommitsDedupeAndActivation via +// time.Now().UTC().Format(time.RFC3339Nano) — two independent code paths +// that are not guaranteed to agree on a text format. dialect.DateTimeOf +// normalizes both sides before comparing for exactly this reason; these +// tests prove that normalization against the real writers, not a synthetic +// stand-in for them. + +import ( + "context" + "path/filepath" + "testing" + "time" + + "github.com/jmoiron/sqlx" + + "github.com/kandev/kandev/internal/analytics/models" + "github.com/kandev/kandev/internal/db" + taskmodels "github.com/kandev/kandev/internal/task/models" + tasksqlite "github.com/kandev/kandev/internal/task/repository/sqlite" +) + +// newRealSchemaTestDB opens a fresh SQLite file and boots the REAL +// task/repository/sqlite schema on it (creates kandev_meta and runs +// migrateSessionCommitsDedupeAndActivation for real), then attaches an +// analytics Repository to the same connection — mirroring how +// backendapp/storage.go wires both repositories onto one shared pool. +func newRealSchemaTestDB(t *testing.T) (*sqlx.DB, *tasksqlite.Repository, *Repository) { + t.Helper() + tmpDir := t.TempDir() + rawDB, err := db.OpenSQLite(filepath.Join(tmpDir, "activation.db")) + if err != nil { + t.Fatalf("OpenSQLite: %v", err) + } + t.Cleanup(func() { _ = rawDB.Close() }) + sqlxDB := sqlx.NewDb(rawDB, "sqlite3") + + taskRepo, err := tasksqlite.NewWithDB(sqlxDB, sqlxDB, nil) + if err != nil { + t.Fatalf("task repo NewWithDB: %v", err) + } + analyticsRepo, err := NewWithDB(sqlxDB, sqlxDB) + if err != nil { + t.Fatalf("analytics repo NewWithDB: %v", err) + } + return sqlxDB, taskRepo, analyticsRepo +} + +// Regression: a session started well after the real activation marker +// (written by the real migration, not a hand-formatted string) must get a +// real *0, not nil, when it has no commits — proving dialect.DateTimeOf +// correctly normalizes the actual driver-written started_at text against +// the actual RFC3339Nano-formatted kandev_meta value. +func TestListSessionCodeStats_RealSchemaPostActivationSessionReturnsZero(t *testing.T) { + sqlxDB, taskRepo, analyticsRepo := newRealSchemaTestDB(t) + ctx := context.Background() + + var activatedAt string + if err := sqlxDB.Get(&activatedAt, `SELECT value FROM kandev_meta WHERE key = 'commit_capture_activated_at'`); err != nil { + t.Fatalf("read real activation marker: %v", err) + } + if _, err := time.Parse(time.RFC3339Nano, activatedAt); err != nil { + t.Fatalf("activation marker %q is not RFC3339Nano: %v", activatedAt, err) + } + + if err := taskRepo.CreateTask(ctx, &taskmodels.Task{ID: "task-post", Title: "T1"}); err != nil { + t.Fatalf("CreateTask: %v", err) + } + if err := taskRepo.CreateTaskSession(ctx, &taskmodels.TaskSession{ + ID: "sess-post", TaskID: "task-post", StartedAt: time.Now().UTC(), + }); err != nil { + t.Fatalf("CreateTaskSession: %v", err) + } + + results, err := analyticsRepo.ListSessionCodeStats(ctx, models.SessionCodeStatsFilter{}) + if err != nil { + t.Fatalf("ListSessionCodeStats: %v", err) + } + if len(results) != 1 { + t.Fatalf("expected 1 session, got %d: %+v", len(results), results) + } + stat := results[0] + if stat.LinesAddedCommitted == nil || *stat.LinesAddedCommitted != 0 { + t.Fatalf("LinesAddedCommitted = %v, want *0 (real driver-written started_at compared against "+ + "real RFC3339Nano activation marker) — a raw-text format mismatch would misclassify this "+ + "post-activation session as legacy", stat.LinesAddedCommitted) + } + if stat.LinesDeletedCommitted == nil || *stat.LinesDeletedCommitted != 0 { + t.Fatalf("LinesDeletedCommitted = %v, want *0", stat.LinesDeletedCommitted) + } +} + +// Regression companion: a session whose real, driver-written started_at +// predates the real activation marker must get nil, not 0. +func TestListSessionCodeStats_RealSchemaPreActivationSessionReturnsNil(t *testing.T) { + sqlxDB, taskRepo, analyticsRepo := newRealSchemaTestDB(t) + ctx := context.Background() + + if err := taskRepo.CreateTask(ctx, &taskmodels.Task{ID: "task-pre", Title: "T1"}); err != nil { + t.Fatalf("CreateTask: %v", err) + } + // A real driver-written started_at, deliberately long before "now" (and + // therefore before the activation marker the migration just wrote at + // boot time, which is always ~"now"). + legacyStartedAt := time.Date(2020, 1, 1, 0, 0, 0, 0, time.UTC) + if err := taskRepo.CreateTaskSession(ctx, &taskmodels.TaskSession{ + ID: "sess-pre", TaskID: "task-pre", StartedAt: legacyStartedAt, + }); err != nil { + t.Fatalf("CreateTaskSession: %v", err) + } + + var rawStartedAt string + if err := sqlxDB.Get(&rawStartedAt, `SELECT started_at FROM task_sessions WHERE id = 'sess-pre'`); err != nil { + t.Fatalf("read raw started_at: %v", err) + } + t.Logf("real driver-written started_at = %q", rawStartedAt) + + results, err := analyticsRepo.ListSessionCodeStats(ctx, models.SessionCodeStatsFilter{}) + if err != nil { + t.Fatalf("ListSessionCodeStats: %v", err) + } + if len(results) != 1 { + t.Fatalf("expected 1 session, got %d: %+v", len(results), results) + } + stat := results[0] + if stat.LinesAddedCommitted != nil { + t.Fatalf("LinesAddedCommitted = %v, want nil (real driver-written started_at predates the real "+ + "activation marker)", *stat.LinesAddedCommitted) + } + if stat.LinesDeletedCommitted != nil { + t.Fatalf("LinesDeletedCommitted = %v, want nil", *stat.LinesDeletedCommitted) + } +} diff --git a/apps/backend/internal/analytics/repository/sqlite/repository_test.go b/apps/backend/internal/analytics/repository/sqlite/repository_test.go index a227922e9b..7c1ff571bc 100644 --- a/apps/backend/internal/analytics/repository/sqlite/repository_test.go +++ b/apps/backend/internal/analytics/repository/sqlite/repository_test.go @@ -129,10 +129,25 @@ func createTestDB(t *testing.T) *sqlx.DB { files TEXT DEFAULT '{}', created_at TIMESTAMP NOT NULL ); + CREATE TABLE IF NOT EXISTS kandev_meta ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL DEFAULT '' + ); ` if _, err := sqlxDB.Exec(schema); err != nil { t.Fatalf("failed to create schema: %v", err) } + // Seed a commit-capture activation marker far in the past, matching + // migrateSessionCommitsDedupeAndActivation's production kandev_meta row, + // so every test session (started_at ~ time.Now()) is post-activation and + // gets a real 0 for committed lines rather than the legacy-NULL case - + // tests that want the legacy path override this row explicitly. + if _, err := sqlxDB.Exec( + `INSERT INTO kandev_meta (key, value) VALUES (?, ?)`, + commitCaptureActivatedAtMetaKey, "2000-01-01T00:00:00Z", + ); err != nil { + t.Fatalf("failed to seed commit_capture_activated_at: %v", err) + } return sqlxDB } diff --git a/apps/backend/internal/analytics/repository/sqlite/session_code_stats_postgres_test.go b/apps/backend/internal/analytics/repository/sqlite/session_code_stats_postgres_test.go new file mode 100644 index 0000000000..927fc4afa8 --- /dev/null +++ b/apps/backend/internal/analytics/repository/sqlite/session_code_stats_postgres_test.go @@ -0,0 +1,104 @@ +package sqlite + +// Postgres parity coverage for ListSessionCodeStats' commit-capture-activation +// comparison. dialect.NaiveUTCTimestampOf / dialect.DateTimeOf are dialect- +// sensitive by design: task_sessions.started_at is a naive TIMESTAMP column +// (no embedded zone) always written as UTC wall-clock time, and +// kandev_meta.value is text carrying an explicit RFC3339Nano "Z" offset. A +// bare `(started_at)::timestamptz` cast on Postgres reinterprets the naive +// value in the session's `timezone` GUC instead of UTC, silently shifting it +// by the server's offset and misclassifying a post-activation session as +// legacy — this reproduces only against a real, non-UTC-configured Postgres +// server; a SQLite-only test suite cannot catch it. Skips unless +// KANDEV_TEST_POSTGRES_DSN is set; CI runs these in postgres-boot. + +import ( + "context" + "testing" + "time" + + "github.com/kandev/kandev/internal/analytics/models" + taskmodels "github.com/kandev/kandev/internal/task/models" + tasksqlite "github.com/kandev/kandev/internal/task/repository/sqlite" + "github.com/kandev/kandev/internal/testutil" +) + +// TestPostgresListSessionCodeStatsActivationComparison drives both the +// task-repository and analytics-repository writers/readers over the SAME +// Postgres connection (mirroring backendapp/storage.go's shared writer/reader +// pool), so both the real driver-written started_at text and the real +// RFC3339Nano-formatted activation marker are exercised — not a synthetic +// stand-in for either. +func TestPostgresListSessionCodeStatsActivationComparison(t *testing.T) { + db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) + + taskRepo, err := tasksqlite.NewWithDB(db, db, nil) + if err != nil { + t.Fatalf("task repo NewWithDB: %v", err) + } + analyticsRepo, err := NewWithDB(db, db) + if err != nil { + t.Fatalf("analytics repo NewWithDB: %v", err) + } + + var activatedAt string + if err := db.Get(&activatedAt, db.Rebind(`SELECT value FROM kandev_meta WHERE key = ?`), commitCaptureActivatedAtMetaKey); err != nil { + t.Fatalf("read activation marker: %v", err) + } + if _, err := time.Parse(time.RFC3339Nano, activatedAt); err != nil { + t.Fatalf("activation marker %q is not RFC3339Nano: %v", activatedAt, err) + } + + ctx := context.Background() + if err := taskRepo.CreateTask(ctx, &taskmodels.Task{ID: "task-pg-post", Title: "T1"}); err != nil { + t.Fatalf("CreateTask post: %v", err) + } + // Started moments after the activation marker was written (real + // driver-bound time.Time, not a hand-formatted string). + if err := taskRepo.CreateTaskSession(ctx, &taskmodels.TaskSession{ + ID: "sess-pg-post", TaskID: "task-pg-post", StartedAt: time.Now().UTC(), + }); err != nil { + t.Fatalf("CreateTaskSession post: %v", err) + } + + if err := taskRepo.CreateTask(ctx, &taskmodels.Task{ID: "task-pg-pre", Title: "T2"}); err != nil { + t.Fatalf("CreateTask pre: %v", err) + } + if err := taskRepo.CreateTaskSession(ctx, &taskmodels.TaskSession{ + ID: "sess-pg-pre", TaskID: "task-pg-pre", StartedAt: time.Date(2020, 1, 1, 0, 0, 0, 0, time.UTC), + }); err != nil { + t.Fatalf("CreateTaskSession pre: %v", err) + } + + results, err := analyticsRepo.ListSessionCodeStats(ctx, models.SessionCodeStatsFilter{}) + if err != nil { + t.Fatalf("ListSessionCodeStats: %v", err) + } + byID := make(map[string]*models.SessionCodeStats, len(results)) + for _, r := range results { + byID[r.SessionID] = r + } + + post, ok := byID["sess-pg-post"] + if !ok { + t.Fatalf("sess-pg-post missing from results: %+v", results) + } + if post.LinesAddedCommitted == nil || *post.LinesAddedCommitted != 0 { + t.Errorf("post-activation session: LinesAddedCommitted = %v, want *0 — "+ + "a naive-timestamp-vs-session-timezone bug would misclassify this as legacy", post.LinesAddedCommitted) + } + if post.LinesDeletedCommitted == nil || *post.LinesDeletedCommitted != 0 { + t.Errorf("post-activation session: LinesDeletedCommitted = %v, want *0", post.LinesDeletedCommitted) + } + + pre, ok := byID["sess-pg-pre"] + if !ok { + t.Fatalf("sess-pg-pre missing from results: %+v", results) + } + if pre.LinesAddedCommitted != nil { + t.Errorf("pre-activation session: LinesAddedCommitted = %v, want nil", *pre.LinesAddedCommitted) + } + if pre.LinesDeletedCommitted != nil { + t.Errorf("pre-activation session: LinesDeletedCommitted = %v, want nil", *pre.LinesDeletedCommitted) + } +} diff --git a/apps/backend/internal/analytics/repository/sqlite/session_code_stats_test.go b/apps/backend/internal/analytics/repository/sqlite/session_code_stats_test.go index 0bf61e083d..d9525a2061 100644 --- a/apps/backend/internal/analytics/repository/sqlite/session_code_stats_test.go +++ b/apps/backend/internal/analytics/repository/sqlite/session_code_stats_test.go @@ -39,11 +39,11 @@ func TestListSessionCodeStats_CommittedSumsOnly(t *testing.T) { if stat.SessionID != "sess-1" { t.Errorf("SessionID: want sess-1, got %s", stat.SessionID) } - if stat.LinesAddedCommitted != 30 { - t.Errorf("LinesAddedCommitted: want 30, got %d", stat.LinesAddedCommitted) + if stat.LinesAddedCommitted == nil || *stat.LinesAddedCommitted != 30 { + t.Errorf("LinesAddedCommitted: want *30, got %v", stat.LinesAddedCommitted) } - if stat.LinesDeletedCommitted != 5 { - t.Errorf("LinesDeletedCommitted: want 5, got %d", stat.LinesDeletedCommitted) + if stat.LinesDeletedCommitted == nil || *stat.LinesDeletedCommitted != 5 { + t.Errorf("LinesDeletedCommitted: want *5, got %v", stat.LinesDeletedCommitted) } if stat.LinesAddedPeakPending != 0 { t.Errorf("LinesAddedPeakPending: want 0 (no snapshots), got %d", stat.LinesAddedPeakPending) @@ -99,12 +99,14 @@ func TestListSessionCodeStats_PeakPendingWinsOverLatestSnapshot(t *testing.T) { if stat.LinesDeletedPeakPending != 50 { t.Errorf("LinesDeletedPeakPending: want 50 (peak, not latest), got %d", stat.LinesDeletedPeakPending) } - // No commits — committed sums stay at zero and are NOT conflated with pending. - if stat.LinesAddedCommitted != 0 { - t.Errorf("LinesAddedCommitted: want 0, got %d", stat.LinesAddedCommitted) + // No commits, but the session started after commit-capture activation + // (createTestDB seeds the marker at 2000-01-01) — committed sums are a + // real *0, not nil, and are NOT conflated with pending. + if stat.LinesAddedCommitted == nil || *stat.LinesAddedCommitted != 0 { + t.Errorf("LinesAddedCommitted: want *0, got %v", stat.LinesAddedCommitted) } - if stat.LinesDeletedCommitted != 0 { - t.Errorf("LinesDeletedCommitted: want 0, got %d", stat.LinesDeletedCommitted) + if stat.LinesDeletedCommitted == nil || *stat.LinesDeletedCommitted != 0 { + t.Errorf("LinesDeletedCommitted: want *0, got %v", stat.LinesDeletedCommitted) } } @@ -141,14 +143,14 @@ func TestListSessionCodeStats_CommittedAndPendingReportedSeparately(t *testing.T stat := results[0] // Committed and pending must be reported as distinct fields, not summed // (10 committed + 7 pending must NOT collapse into 17 anywhere). - if stat.LinesAddedCommitted != 10 { - t.Errorf("LinesAddedCommitted: want 10, got %d", stat.LinesAddedCommitted) + if stat.LinesAddedCommitted == nil || *stat.LinesAddedCommitted != 10 { + t.Errorf("LinesAddedCommitted: want *10, got %v", stat.LinesAddedCommitted) } if stat.LinesAddedPeakPending != 7 { t.Errorf("LinesAddedPeakPending: want 7, got %d", stat.LinesAddedPeakPending) } - if stat.LinesDeletedCommitted != 4 { - t.Errorf("LinesDeletedCommitted: want 4, got %d", stat.LinesDeletedCommitted) + if stat.LinesDeletedCommitted == nil || *stat.LinesDeletedCommitted != 4 { + t.Errorf("LinesDeletedCommitted: want *4, got %v", stat.LinesDeletedCommitted) } if stat.LinesDeletedPeakPending != 1 { t.Errorf("LinesDeletedPeakPending: want 1, got %d", stat.LinesDeletedPeakPending) @@ -179,7 +181,10 @@ func TestListSessionCodeStats_NoCommitsNoSnapshotsReturnsZeros(t *testing.T) { t.Fatalf("expected 1 session, got %d", len(results)) } stat := results[0] - if stat.LinesAddedCommitted != 0 || stat.LinesDeletedCommitted != 0 || + // The session started after commit-capture activation (createTestDB seeds + // the marker at 2000-01-01), so committed lines are a real *0, not nil. + if stat.LinesAddedCommitted == nil || *stat.LinesAddedCommitted != 0 || + stat.LinesDeletedCommitted == nil || *stat.LinesDeletedCommitted != 0 || stat.LinesAddedPeakPending != 0 || stat.LinesDeletedPeakPending != 0 { t.Errorf("expected all-zero stats for session with no commits/snapshots, got %+v", stat) } @@ -297,3 +302,122 @@ func TestListSessionCodeStats_FiltersByWorkspaceIDs(t *testing.T) { t.Errorf("SessionID: want sess-2, got %s", results[0].SessionID) } } + +// legacyStartedAt predates createTestDB's seeded commit_capture_activated_at +// (2000-01-01T00:00:00Z). +const legacyStartedAt = "1999-01-01T00:00:00Z" + +// Regression: a session that started before commit-capture activation, and +// has no observed commits, must report nil for committed lines — not 0. +// Capture wasn't running for this session, so "zero commits" and "unknown +// because nothing could have been captured yet" are indistinguishable; only +// nil is honest. Before this fix, COALESCE(..., 0) always claimed a real +// zero regardless of the activation marker (see migrateSessionCommitsDedupeAndActivation). +func TestListSessionCodeStats_LegacySessionReturnsNilCommittedLines(t *testing.T) { + dbConn := createTestDB(t) + repo, err := NewWithDB(dbConn, dbConn) + if err != nil { + t.Fatalf("NewWithDB failed: %v", err) + } + + ctx := context.Background() + now := time.Now().UTC() + nowStr := now.Format(time.RFC3339) + + execOrFatal(t, dbConn, `INSERT INTO workspaces (id, name, created_at, updated_at) VALUES ('ws-1', 'Test', ?, ?)`, nowStr, nowStr) + execOrFatal(t, dbConn, `INSERT INTO boards (id, workspace_id, name, created_at, updated_at) VALUES ('board-1', 'ws-1', 'Board', ?, ?)`, nowStr, nowStr) + execOrFatal(t, dbConn, `INSERT INTO tasks (id, workspace_id, board_id, workflow_step_id, title, is_ephemeral, created_at, updated_at) VALUES ('task-1', 'ws-1', 'board-1', '', 'T1', 0, ?, ?)`, nowStr, nowStr) + execOrFatal(t, dbConn, `INSERT INTO task_sessions (id, task_id, agent_profile_id, state, started_at, updated_at) VALUES ('sess-legacy', 'task-1', 'agent-1', 'COMPLETED', ?, ?)`, legacyStartedAt, nowStr) + + results, err := repo.ListSessionCodeStats(ctx, models.SessionCodeStatsFilter{}) + if err != nil { + t.Fatalf("ListSessionCodeStats failed: %v", err) + } + if len(results) != 1 { + t.Fatalf("expected 1 session, got %d", len(results)) + } + stat := results[0] + if stat.LinesAddedCommitted != nil { + t.Errorf("LinesAddedCommitted: want nil (legacy, predates commit-capture activation), got %d", *stat.LinesAddedCommitted) + } + if stat.LinesDeletedCommitted != nil { + t.Errorf("LinesDeletedCommitted: want nil (legacy, predates commit-capture activation), got %d", *stat.LinesDeletedCommitted) + } + // Peak-pending is unrelated to commit-capture activation and stays a real 0. + if stat.LinesAddedPeakPending != 0 || stat.LinesDeletedPeakPending != 0 { + t.Errorf("peak pending: want 0/0, got %d/%d", stat.LinesAddedPeakPending, stat.LinesDeletedPeakPending) + } +} + +// Regression companion: a session that started AFTER activation, with no +// commits, must still report a real *0 — capture was live for its entire +// lifetime, so zero commits is a trustworthy measurement, not an unknown. +func TestListSessionCodeStats_PostActivationSessionWithNoCommitsReturnsZeroNotNil(t *testing.T) { + dbConn := createTestDB(t) + repo, err := NewWithDB(dbConn, dbConn) + if err != nil { + t.Fatalf("NewWithDB failed: %v", err) + } + + ctx := context.Background() + now := time.Now().UTC() + nowStr := now.Format(time.RFC3339) + + execOrFatal(t, dbConn, `INSERT INTO workspaces (id, name, created_at, updated_at) VALUES ('ws-1', 'Test', ?, ?)`, nowStr, nowStr) + execOrFatal(t, dbConn, `INSERT INTO boards (id, workspace_id, name, created_at, updated_at) VALUES ('board-1', 'ws-1', 'Board', ?, ?)`, nowStr, nowStr) + execOrFatal(t, dbConn, `INSERT INTO tasks (id, workspace_id, board_id, workflow_step_id, title, is_ephemeral, created_at, updated_at) VALUES ('task-1', 'ws-1', 'board-1', '', 'T1', 0, ?, ?)`, nowStr, nowStr) + execOrFatal(t, dbConn, `INSERT INTO task_sessions (id, task_id, agent_profile_id, state, started_at, updated_at) VALUES ('sess-post', 'task-1', 'agent-1', 'COMPLETED', ?, ?)`, nowStr, nowStr) + + results, err := repo.ListSessionCodeStats(ctx, models.SessionCodeStatsFilter{}) + if err != nil { + t.Fatalf("ListSessionCodeStats failed: %v", err) + } + if len(results) != 1 { + t.Fatalf("expected 1 session, got %d", len(results)) + } + stat := results[0] + if stat.LinesAddedCommitted == nil || *stat.LinesAddedCommitted != 0 { + t.Errorf("LinesAddedCommitted: want *0 (capture was active, genuinely zero commits), got %v", stat.LinesAddedCommitted) + } + if stat.LinesDeletedCommitted == nil || *stat.LinesDeletedCommitted != 0 { + t.Errorf("LinesDeletedCommitted: want *0, got %v", stat.LinesDeletedCommitted) + } +} + +// Regression companion: a session that predates activation but DOES have an +// observed commit (captured by a later-activated writer, or backfilled) must +// report the real sum — nil applies only when there is truly nothing to +// distinguish from zero, not to every session older than the activation +// marker. +func TestListSessionCodeStats_LegacySessionWithCommitsReturnsRealSum(t *testing.T) { + dbConn := createTestDB(t) + repo, err := NewWithDB(dbConn, dbConn) + if err != nil { + t.Fatalf("NewWithDB failed: %v", err) + } + + ctx := context.Background() + now := time.Now().UTC() + nowStr := now.Format(time.RFC3339) + + execOrFatal(t, dbConn, `INSERT INTO workspaces (id, name, created_at, updated_at) VALUES ('ws-1', 'Test', ?, ?)`, nowStr, nowStr) + execOrFatal(t, dbConn, `INSERT INTO boards (id, workspace_id, name, created_at, updated_at) VALUES ('board-1', 'ws-1', 'Board', ?, ?)`, nowStr, nowStr) + execOrFatal(t, dbConn, `INSERT INTO tasks (id, workspace_id, board_id, workflow_step_id, title, is_ephemeral, created_at, updated_at) VALUES ('task-1', 'ws-1', 'board-1', '', 'T1', 0, ?, ?)`, nowStr, nowStr) + execOrFatal(t, dbConn, `INSERT INTO task_sessions (id, task_id, agent_profile_id, state, started_at, updated_at) VALUES ('sess-legacy-with-commit', 'task-1', 'agent-1', 'COMPLETED', ?, ?)`, legacyStartedAt, nowStr) + execOrFatal(t, dbConn, `INSERT INTO task_session_commits (id, session_id, commit_sha, committed_at, files_changed, insertions, deletions, created_at) VALUES ('c-1', 'sess-legacy-with-commit', 'sha1', ?, 1, 10, 4, ?)`, nowStr, nowStr) + + results, err := repo.ListSessionCodeStats(ctx, models.SessionCodeStatsFilter{}) + if err != nil { + t.Fatalf("ListSessionCodeStats failed: %v", err) + } + if len(results) != 1 { + t.Fatalf("expected 1 session, got %d", len(results)) + } + stat := results[0] + if stat.LinesAddedCommitted == nil || *stat.LinesAddedCommitted != 10 { + t.Errorf("LinesAddedCommitted: want *10 (real observed commit, not nil), got %v", stat.LinesAddedCommitted) + } + if stat.LinesDeletedCommitted == nil || *stat.LinesDeletedCommitted != 4 { + t.Errorf("LinesDeletedCommitted: want *4, got %v", stat.LinesDeletedCommitted) + } +} diff --git a/apps/backend/internal/analytics/repository/sqlite/stats.go b/apps/backend/internal/analytics/repository/sqlite/stats.go index 16af7baef2..20232b3549 100644 --- a/apps/backend/internal/analytics/repository/sqlite/stats.go +++ b/apps/backend/internal/analytics/repository/sqlite/stats.go @@ -611,6 +611,14 @@ func (r *Repository) GetGitStats(ctx context.Context, workspaceID string, start // does not specify one, mirroring GetTaskStats' default page size. const defaultSessionCodeStatsLimit = 500 +// commitCaptureActivatedAtMetaKey mirrors the kandev_meta key +// task/repository/sqlite.migrateSessionCommitsDedupeAndActivation publishes +// (same literal string; this package has its own read-only handle onto the +// same database, and there is no shared cross-package helper for kandev_meta +// reads — every owning package re-embeds the literal query, matching the +// existing idiom in internal/persistence and internal/system/database). +const commitCaptureActivatedAtMetaKey = "commit_capture_activated_at" + // ListSessionCodeStats returns, per session, committed LOC (summed from // task_session_commits) and PEAK pending-diff LOC (the largest single // task_session_git_snapshots snapshot, not the latest — the latest snapshot @@ -619,6 +627,17 @@ const defaultSessionCodeStatsLimit = 500 // used to compute by reading the SQLite file directly (see ADR 0043); the // SQL here mirrors that plugin's sessionsQuery. // +// lines_added_committed / lines_deleted_committed are NULL, not 0, for a +// session with no observed commit rows that started before +// commit_capture_activated_at (kandev_meta, published by +// task/repository/sqlite.migrateSessionCommitsDedupeAndActivation): capture +// wasn't running yet for that session, so zero cannot be told apart from +// unknown. A session with any observed commit, or one that started after +// activation, always gets a real sum (activation absent entirely — should +// not happen once boot has completed successfully — falls through to the +// pre-existing zero behavior via SQL's three-valued NULL comparison logic). +// Peak-pending is unrelated to this marker and is always a real int64. +// // Portability: the committed-sum half of this query is plain SQL and works // unchanged on both drivers. The peak-pending half must walk each snapshot's // `files` JSON object (keyed by file path, see task/models.GitSnapshot.Files) @@ -646,11 +665,28 @@ func (r *Repository) ListSessionCodeStats( // origin; the Host data API is one, so a plugin must not see through it // what the UI deliberately does not show. where += andNotAutomationOriginT + // Compare normalized forms, not the raw text or a naive cast on either + // side. ts.started_at is a naive TIMESTAMP column (no embedded zone) that + // is always written as UTC wall-clock time (time.Now().UTC()) — use + // NaiveUTCTimestampOf, not DateTimeOf, or Postgres reinterprets it in the + // session's timezone GUC and silently shifts it. activation.value is text + // formatted via time.RFC3339Nano, which already carries an explicit "Z" — + // use DateTimeOf, which respects that embedded zone. + startedAtNormalized := dialect.NaiveUTCTimestampOf(driver, "ts.started_at") + activationNormalized := dialect.DateTimeOf(driver, "activation.value") query := fmt.Sprintf(` SELECT ts.id AS session_id, - COALESCE(commit_stats.insertions, 0) AS lines_added_committed, - COALESCE(commit_stats.deletions, 0) AS lines_deleted_committed, + CASE + WHEN commit_stats.session_id IS NOT NULL THEN commit_stats.insertions + WHEN %[1]s < %[2]s THEN NULL + ELSE 0 + END AS lines_added_committed, + CASE + WHEN commit_stats.session_id IS NOT NULL THEN commit_stats.deletions + WHEN %[1]s < %[2]s THEN NULL + ELSE 0 + END AS lines_deleted_committed, COALESCE(peak_stats.peak_additions, 0) AS lines_added_peak_pending, COALESCE(peak_stats.peak_deletions, 0) AS lines_deleted_peak_pending FROM task_sessions ts @@ -662,11 +698,12 @@ func (r *Repository) ListSessionCodeStats( FROM task_session_commits c GROUP BY c.session_id ) commit_stats ON commit_stats.session_id = ts.id - LEFT JOIN (%s) peak_stats ON peak_stats.session_id = ts.id - WHERE %s + LEFT JOIN (%[3]s) peak_stats ON peak_stats.session_id = ts.id + LEFT JOIN kandev_meta activation ON activation.key = '%[4]s' + WHERE %[5]s ORDER BY ts.started_at ASC, ts.id ASC LIMIT ? OFFSET ? - `, peakPendingSnapshotSubquery(driver), where) + `, startedAtNormalized, activationNormalized, peakPendingSnapshotSubquery(driver), commitCaptureActivatedAtMetaKey, where) inQuery, inArgs, err := sqlx.In(query, args...) if err != nil { diff --git a/apps/backend/internal/analytics/service/service_test.go b/apps/backend/internal/analytics/service/service_test.go index cb9b664e70..8cba5b3d93 100644 --- a/apps/backend/internal/analytics/service/service_test.go +++ b/apps/backend/internal/analytics/service/service_test.go @@ -58,9 +58,11 @@ func (f *fakeRepository) ListSessionCodeStats( return f.stats, nil } +func int64Ptr(v int64) *int64 { return &v } + func TestService_ListSessionCodeStats_DelegatesToRepositoryAndReturnsResults(t *testing.T) { want := []*models.SessionCodeStats{ - {SessionID: "sess-1", LinesAddedCommitted: 10, LinesDeletedCommitted: 2, LinesAddedPeakPending: 5, LinesDeletedPeakPending: 1}, + {SessionID: "sess-1", LinesAddedCommitted: int64Ptr(10), LinesDeletedCommitted: int64Ptr(2), LinesAddedPeakPending: 5, LinesDeletedPeakPending: 1}, } repo := &fakeRepository{stats: want} svc := New(repo) diff --git a/apps/backend/internal/db/dialect/dialect_test.go b/apps/backend/internal/db/dialect/dialect_test.go index 8dfabec0ed..7c8c91faa8 100644 --- a/apps/backend/internal/db/dialect/dialect_test.go +++ b/apps/backend/internal/db/dialect/dialect_test.go @@ -102,6 +102,28 @@ func TestDateOf(t *testing.T) { } } +func TestDateTimeOf(t *testing.T) { + got := DateTimeOf(SQLite3, "activation.value") + if got != "datetime(activation.value)" { + t.Errorf("sqlite: got %q", got) + } + got = DateTimeOf(PGX, "activation.value") + if got != "(activation.value)::timestamptz" { + t.Errorf("pgx: got %q", got) + } +} + +func TestNaiveUTCTimestampOf(t *testing.T) { + got := NaiveUTCTimestampOf(SQLite3, "ts.started_at") + if got != "datetime(ts.started_at)" { + t.Errorf("sqlite: got %q", got) + } + got = NaiveUTCTimestampOf(PGX, "ts.started_at") + if got != "(ts.started_at AT TIME ZONE 'UTC')" { + t.Errorf("pgx: got %q", got) + } +} + func TestNow(t *testing.T) { if Now(SQLite3) != "datetime('now')" { t.Errorf("sqlite: got %q", Now(SQLite3)) diff --git a/apps/backend/internal/db/dialect/time.go b/apps/backend/internal/db/dialect/time.go index 3b56db4666..af9a1bbe12 100644 --- a/apps/backend/internal/db/dialect/time.go +++ b/apps/backend/internal/db/dialect/time.go @@ -24,6 +24,57 @@ func DateOf(driver, expr string) string { return fmt.Sprintf("date(%s)", expr) } +// DateTimeOf normalizes a timestamp expression that already carries explicit +// zone information — a `timestamptz` column, or text with an explicit +// ISO8601 offset/"Z" such as a value formatted via time.RFC3339Nano — into a +// canonical, directly comparable form. On Postgres this must NOT be used for +// a naive `timestamp` (without time zone) column: casting a naive value +// straight to `timestamptz` interprets it in the session's `timezone` GUC, +// not UTC, silently shifting it by the server's offset even though this +// codebase always writes such columns via time.Now().UTC(). Use +// NaiveUTCTimestampOf for that case instead. +// +// On SQLite this distinction doesn't matter (datetime() normalizes either +// input the same way, and SQLite has no session-timezone concept), but the +// two functions are still named separately so a Postgres-unaware caller +// can't pick the wrong one by accident. +// +// SQLite: datetime(expr) +// Postgres: (expr)::timestamptz +func DateTimeOf(driver, expr string) string { + if IsPostgres(driver) { + return fmt.Sprintf("(%s)::timestamptz", expr) + } + return fmt.Sprintf("datetime(%s)", expr) +} + +// NaiveUTCTimestampOf normalizes a naive timestamp expression — one with NO +// embedded zone information, but KNOWN to always hold UTC wall-clock values +// (every `TIMESTAMP`-typed column in this codebase, written via +// time.Now().UTC()) — into the same comparable form DateTimeOf produces for +// explicitly-zoned values. Do not use this for an expression that already +// carries zone info (a `timestamptz` column, or RFC3339 text with an +// offset/"Z") — use DateTimeOf for those; applying `AT TIME ZONE 'UTC'` to an +// already-zoned value converts it a second time and silently corrupts it. +// +// The bug this exists to prevent: on Postgres, a bare `(expr)::timestamptz` +// cast on a naive column interprets the stored wall-clock text as being in +// the session's `timezone` GUC (e.g. "Asia/Singapore", +08:00) rather than +// UTC, shifting the value by the server's offset before comparison — a +// session that started milliseconds after a UTC activation marker can come +// out looking hours "earlier" than it. `AT TIME ZONE 'UTC'` tells Postgres +// explicitly what zone the naive value is already in, which a bare cast +// cannot express. +// +// SQLite: datetime(expr) +// Postgres: (expr AT TIME ZONE 'UTC') +func NaiveUTCTimestampOf(driver, expr string) string { + if IsPostgres(driver) { + return fmt.Sprintf("(%s AT TIME ZONE 'UTC')", expr) + } + return fmt.Sprintf("datetime(%s)", expr) +} + // Now returns the SQL expression for the current timestamp. // // SQLite: datetime('now') diff --git a/apps/backend/internal/orchestrator/commit_capture_failure_test.go b/apps/backend/internal/orchestrator/commit_capture_failure_test.go new file mode 100644 index 0000000000..edbccf1988 --- /dev/null +++ b/apps/backend/internal/orchestrator/commit_capture_failure_test.go @@ -0,0 +1,133 @@ +package orchestrator + +import ( + "context" + "expvar" + "testing" + + client "github.com/kandev/kandev/internal/agent/runtime/agentctl" +) + +// commitCaptureFetchFailedCount reads the current commit_capture_fetch_failed +// count for trigger, treating an unset key as 0. +func commitCaptureFetchFailedCount(trigger commitCaptureTrigger) int64 { + v := commitCaptureFetchFailed.Get(string(trigger)) + if v == nil { + return 0 + } + iv, ok := v.(*expvar.Int) + if !ok { + return 0 + } + return iv.Value() +} + +// Regression: captureCommitsForTrigger silently ignored a result-level +// GetGitLog failure (HTTP 200 but Success == false) - it only checked the +// transport err and a nil result, so a git command failure inside agentctl +// (e.g. corrupted worktree, git binary missing) produced zero commits and +// zero observability. This undercuts the writer-health goal recorded +// alongside commitCaptureFailed/commitCaptureObserved: a trigger silently +// going dark should be visible. +func TestCaptureCommitsForTrigger_RecordsTotalFetchFailure(t *testing.T) { + ctx := context.Background() + testRepo := setupTestRepo(t) + seedSession(t, testRepo, "t-fetch-fail", "s-fetch-fail", "step1") + + agentMgr := &mockAgentManager{ + getGitLogFunc: func(context.Context, string, string, int, string) (*client.GitLogResult, error) { + return &client.GitLogResult{Success: false, Error: "git log failed: fatal: bad object"}, nil + }, + } + svc := createTestServiceWithAgent(testRepo, newMockStepGetter(), newMockTaskRepo(), agentMgr) + + before := commitCaptureFetchFailedCount(commitCaptureTriggerSweep) + svc.captureCommitsForTrigger(ctx, "s-fetch-fail", "base-sha", "", commitCaptureTriggerSweep) + after := commitCaptureFetchFailedCount(commitCaptureTriggerSweep) + + if after != before+1 { + t.Fatalf("commit_capture_fetch_failed[sweep] = %d, want %d (a total-failure GetGitLog result must be recorded)", after, before+1) + } + + commits, err := testRepo.GetSessionCommits(ctx, "s-fetch-fail") + if err != nil { + t.Fatalf("GetSessionCommits: %v", err) + } + if len(commits) != 0 { + t.Fatalf("commit rows = %d, want 0 (nothing to persist on total failure)", len(commits)) + } +} + +// Regression: a multi-repo GetGitLog response can report overall Success == +// true (at least one repo succeeded) while PerRepoErrors names the repos +// that failed - captureCommitsForTrigger persisted the successful commits +// but never recorded or logged the per-repo failures, so a repo silently +// going uncaptured inside an otherwise-succeeding multi-repo session was +// invisible. +func TestCaptureCommitsForTrigger_RecordsPartialMultiRepoFailure(t *testing.T) { + ctx := context.Background() + testRepo := setupTestRepo(t) + seedSession(t, testRepo, "t-partial-fail", "s-partial-fail", "step1") + + agentMgr := &mockAgentManager{ + getGitLogFunc: func(context.Context, string, string, int, string) (*client.GitLogResult, error) { + return &client.GitLogResult{ + Success: true, + Commits: []*client.GitCommitInfo{ + {CommitSHA: "ok-sha", CommitMessage: "feat: from the healthy repo", CommittedAt: "2026-04-02T03:04:05Z", Insertions: 2}, + }, + PerRepoErrors: []client.GitLogRepoError{ + {RepositoryName: "flaky-repo", Error: "git log failed: repository not found"}, + }, + }, nil + }, + } + svc := createTestServiceWithAgent(testRepo, newMockStepGetter(), newMockTaskRepo(), agentMgr) + + before := commitCaptureFetchFailedCount(commitCaptureTriggerLive) + svc.captureCommitsForTrigger(ctx, "s-partial-fail", "base-sha", "", commitCaptureTriggerLive) + after := commitCaptureFetchFailedCount(commitCaptureTriggerLive) + + if after != before+1 { + t.Fatalf("commit_capture_fetch_failed[live] = %d, want %d (a per-repo failure inside a successful multi-repo result must be recorded)", after, before+1) + } + + commits, err := testRepo.GetSessionCommits(ctx, "s-partial-fail") + if err != nil { + t.Fatalf("GetSessionCommits: %v", err) + } + if len(commits) != 1 || commits[0].CommitSHA != "ok-sha" { + t.Fatalf("commits = %+v, want exactly [ok-sha] (the healthy repo's commit must still persist despite the other repo's failure)", commits) + } +} + +// Regression: a multi-repo total failure (every repo failed) reports +// Success == false AND populates PerRepoErrors for every repo - the two +// signals must not be double-counted into commit_capture_fetch_failed. +func TestCaptureCommitsForTrigger_TotalMultiRepoFailureCountsOncePerRepo(t *testing.T) { + ctx := context.Background() + testRepo := setupTestRepo(t) + seedSession(t, testRepo, "t-total-multi-fail", "s-total-multi-fail", "step1") + + agentMgr := &mockAgentManager{ + getGitLogFunc: func(context.Context, string, string, int, string) (*client.GitLogResult, error) { + return &client.GitLogResult{ + Success: false, + Error: "git log failed for all 2 repositories", + PerRepoErrors: []client.GitLogRepoError{ + {RepositoryName: "repo-a", Error: "git log failed: repository not found"}, + {RepositoryName: "repo-b", Error: "git log failed: permission denied"}, + }, + }, nil + }, + } + svc := createTestServiceWithAgent(testRepo, newMockStepGetter(), newMockTaskRepo(), agentMgr) + + before := commitCaptureFetchFailedCount(commitCaptureTriggerArchive) + svc.captureCommitsForTrigger(ctx, "s-total-multi-fail", "base-sha", "", commitCaptureTriggerArchive) + after := commitCaptureFetchFailedCount(commitCaptureTriggerArchive) + + if after != before+2 { + t.Fatalf("commit_capture_fetch_failed[archive] = %d, want %d (one increment per failed repo, not one more for the overall Success=false)", after, before+2) + } +} diff --git a/apps/backend/internal/orchestrator/commit_capture_metrics.go b/apps/backend/internal/orchestrator/commit_capture_metrics.go new file mode 100644 index 0000000000..5021f6bf7a --- /dev/null +++ b/apps/backend/internal/orchestrator/commit_capture_metrics.go @@ -0,0 +1,59 @@ +package orchestrator + +import "expvar" + +// expvar maps published at package init under /debug/vars (dev profile, see +// apps/backend/AGENTS.md Observability), mirroring the office scheduler's +// routing_* counters (internal/office/scheduler/metrics_vars.go). Each +// observed commit increments "observed" plus exactly one of +// inserted/duplicate/failed, broken down by which write path saw it - +// commitCaptureTriggerLive (handleGitCommitCreated), ...Sweep (the +// agent-completed reconcile pass), or ...Archive (archive capture, +// unchanged as the final reconciliation pass). A trigger that stops +// incrementing while the others keep moving is the writer-health signal: +// one path silently stopped without the whole feature going dark. +var ( + commitCaptureObserved = expvar.NewMap("commit_capture_observed") + commitCaptureInserted = expvar.NewMap("commit_capture_inserted") + commitCaptureDuplicate = expvar.NewMap("commit_capture_duplicate") + commitCaptureFailed = expvar.NewMap("commit_capture_failed") + + // commitCaptureFetchFailed counts GetGitLog-level failures - either a + // total result-level failure (Success == false) or an individual + // per-repo failure inside a multi-repo response's PerRepoErrors - broken + // down by trigger. These happen before any commit is even observed, so + // they are accounted separately from commitCaptureFailed (a known + // commit's DB-write failure, paired with commitCaptureObserved via the + // observed = inserted + duplicate + failed identity); folding fetch + // failures into that counter would break the identity. + commitCaptureFetchFailed = expvar.NewMap("commit_capture_fetch_failed") +) + +// commitCaptureTrigger identifies which write path observed a commit. +type commitCaptureTrigger string + +const ( + commitCaptureTriggerLive commitCaptureTrigger = "live" + commitCaptureTriggerSweep commitCaptureTrigger = "sweep" + commitCaptureTriggerArchive commitCaptureTrigger = "archive" +) + +func recordCommitCaptureObserved(trigger commitCaptureTrigger) { + commitCaptureObserved.Add(string(trigger), 1) +} + +func recordCommitCaptureInserted(trigger commitCaptureTrigger) { + commitCaptureInserted.Add(string(trigger), 1) +} + +func recordCommitCaptureDuplicate(trigger commitCaptureTrigger) { + commitCaptureDuplicate.Add(string(trigger), 1) +} + +func recordCommitCaptureFailed(trigger commitCaptureTrigger) { + commitCaptureFailed.Add(string(trigger), 1) +} + +func recordCommitCaptureFetchFailed(trigger commitCaptureTrigger) { + commitCaptureFetchFailed.Add(string(trigger), 1) +} diff --git a/apps/backend/internal/orchestrator/event_handlers_agent.go b/apps/backend/internal/orchestrator/event_handlers_agent.go index 8e735552db..df7db7c27f 100644 --- a/apps/backend/internal/orchestrator/event_handlers_agent.go +++ b/apps/backend/internal/orchestrator/event_handlers_agent.go @@ -1006,6 +1006,19 @@ func (s *Service) handleAgentCompleted(ctx context.Context, data watcher.AgentEv return } + // Reconcile task_session_commits before acquiring the per-session + // cancel-in-flight mutex below. This must still run before + // handleAgentCompletedLocked's rotated-execution/terminal-state guards + // (see captureSessionCommitsSweep's doc for why - GetGitLog is resolved + // by session ID, not execution ID, and a session's worktree is shared + // across executions), but it does not need this mutex's exclusivity: the + // sweep is read-mostly, best-effort, and idempotent on the write side + // (ON CONFLICT DO NOTHING). Running it here, before the lock, keeps + // Stop/Cancel/Delete on this session from blocking behind up to 10s of + // git I/O - the mutex below is needed by ~20 other call sites across + // internal/orchestrator/ for exactly those operations. + s.captureSessionCommitsSweep(context.WithoutCancel(ctx), data.SessionID) + // Completion owns workflow advancement only while serialized with every // cancel/interrupt decision for this session. If coordinator stop won while // the event waited, the guarded state reload below observes CANCELLED and @@ -1049,6 +1062,13 @@ func (s *Service) handleAgentCompletedLocked(ctx context.Context, data watcher.A return } + // task_session_commits reconciliation (captureSessionCommitsSweep) runs in + // the caller, handleAgentCompleted, before this function's rotated- + // execution/terminal-state guards below are even reached - and, for a + // session-scoped event, before the per-session cancel-in-flight mutex is + // acquired at all (see the comment at that call site for why the sweep + // does not need that mutex's exclusivity). + // Skip transition logic when this event is the side-effect of a deliberate // stop (e.g. a workflow profile-switch calling completeAndStopSession). Two // signals identify that case: diff --git a/apps/backend/internal/orchestrator/event_handlers_agent_commit_sweep_test.go b/apps/backend/internal/orchestrator/event_handlers_agent_commit_sweep_test.go new file mode 100644 index 0000000000..b421d07150 --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_agent_commit_sweep_test.go @@ -0,0 +1,394 @@ +package orchestrator + +import ( + "context" + "testing" + "time" + + client "github.com/kandev/kandev/internal/agent/runtime/agentctl" + "github.com/kandev/kandev/internal/orchestrator/watcher" + "github.com/kandev/kandev/internal/task/models" +) + +// Regression: the live commit-created event can miss a commit (e.g. pushed +// just before agentctl's next poll tick, see filterLocalCommits' upstream +// filter). captureSessionCommitsSweep is the per-turn reconcile pass that +// runs while the agent process is still alive - unlike archive capture, +// whose GetGitLog call usually finds it already gone - and must persist any +// commit the live path missed. +func TestCaptureSessionCommitsSweep_FillsCommitLivePathMissed(t *testing.T) { + ctx := context.Background() + testRepo := setupTestRepo(t) + seedSession(t, testRepo, "t-sweep", "s-sweep", "step1") + + session, err := testRepo.GetTaskSession(ctx, "s-sweep") + if err != nil { + t.Fatalf("load session: %v", err) + } + session.BaseCommitSHA = "base-sha" + if err := testRepo.UpdateTaskSession(ctx, session); err != nil { + t.Fatalf("set base commit: %v", err) + } + + agentMgr := &mockAgentManager{ + getGitLogFunc: func(_ context.Context, sessionID, baseCommit string, _ int, _ string) (*client.GitLogResult, error) { + if sessionID != "s-sweep" || baseCommit != "base-sha" { + t.Fatalf("GetGitLog called with (%q, %q), want (s-sweep, base-sha)", sessionID, baseCommit) + } + return &client.GitLogResult{ + Success: true, + Commits: []*client.GitCommitInfo{ + {CommitSHA: "missed-sha", CommitMessage: "feat: missed by live path", CommittedAt: "2026-04-02T03:04:05Z", Insertions: 5}, + }, + }, nil + }, + } + svc := createTestServiceWithAgent(testRepo, newMockStepGetter(), newMockTaskRepo(), agentMgr) + + svc.captureSessionCommitsSweep(ctx, "s-sweep") + + commits, err := testRepo.GetSessionCommits(ctx, "s-sweep") + if err != nil { + t.Fatalf("GetSessionCommits: %v", err) + } + if len(commits) != 1 { + t.Fatalf("commit rows = %d, want 1", len(commits)) + } + if commits[0].CommitSHA != "missed-sha" { + t.Errorf("CommitSHA = %q, want missed-sha", commits[0].CommitSHA) + } +} + +// Regression: when the agent process is gone (GetGitLog returns a nil +// result with a nil error, matching the documented "agent not running" +// contract), the sweep must not error or persist anything - archive capture +// remains the final safety net. +func TestCaptureSessionCommitsSweep_AgentNotRunningIsNoop(t *testing.T) { + ctx := context.Background() + testRepo := setupTestRepo(t) + seedSession(t, testRepo, "t-sweep-gone", "s-sweep-gone", "step1") + + session, err := testRepo.GetTaskSession(ctx, "s-sweep-gone") + if err != nil { + t.Fatalf("load session: %v", err) + } + session.BaseCommitSHA = "base-sha" + if err := testRepo.UpdateTaskSession(ctx, session); err != nil { + t.Fatalf("set base commit: %v", err) + } + + agentMgr := &mockAgentManager{ + getGitLogFunc: func(context.Context, string, string, int, string) (*client.GitLogResult, error) { + return nil, nil + }, + } + svc := createTestServiceWithAgent(testRepo, newMockStepGetter(), newMockTaskRepo(), agentMgr) + + svc.captureSessionCommitsSweep(ctx, "s-sweep-gone") + + commits, err := testRepo.GetSessionCommits(ctx, "s-sweep-gone") + if err != nil { + t.Fatalf("GetSessionCommits: %v", err) + } + if len(commits) != 0 { + t.Fatalf("commit rows = %d, want 0", len(commits)) + } +} + +// Regression: captureSessionCommitsSweep runs inside handleAgentCompletedLocked +// while it holds the per-session cancel-in-flight mutex (see +// acquireCancelInFlightGuard) - the same mutex stopTaskSessionForCoordinator +// and DeleteSession must acquire to serve a user's Stop/Cancel/Delete for +// this session. GetGitLog is a real `git log --shortstat` shellout through +// agentctl, bounded only by the agentctl HTTP client's 60s default timeout +// when passed an undecorated context. The archive-capture call site +// (service_tasks.go's CaptureArchiveSnapshot) deliberately wraps the same +// GetGitLog call in a 10s context.WithTimeout for exactly this reason +// ("prevent blocking the archive operation if agentctl is stuck") - the +// sweep must do the same so a stuck agentctl cannot block Stop/Cancel/Delete +// for up to a minute on every turn completion. +func TestCaptureSessionCommitsSweep_BoundsGetGitLogWithTimeout(t *testing.T) { + ctx := context.Background() + testRepo := setupTestRepo(t) + seedSession(t, testRepo, "t-sweep-bound", "s-sweep-bound", "step1") + + session, err := testRepo.GetTaskSession(ctx, "s-sweep-bound") + if err != nil { + t.Fatalf("load session: %v", err) + } + session.BaseCommitSHA = "base-sha" + if err := testRepo.UpdateTaskSession(ctx, session); err != nil { + t.Fatalf("set base commit: %v", err) + } + + var sawDeadline bool + agentMgr := &mockAgentManager{ + getGitLogFunc: func(callCtx context.Context, _, _ string, _ int, _ string) (*client.GitLogResult, error) { + deadline, ok := callCtx.Deadline() + sawDeadline = ok + if ok && time.Until(deadline) > 11*time.Second { + t.Errorf("GetGitLog context deadline is %s out, want <= ~10s (matching the archive-capture bound)", + time.Until(deadline)) + } + return &client.GitLogResult{Success: true}, nil + }, + } + svc := createTestServiceWithAgent(testRepo, newMockStepGetter(), newMockTaskRepo(), agentMgr) + + svc.captureSessionCommitsSweep(ctx, "s-sweep-bound") + + if !sawDeadline { + t.Fatal("captureSessionCommitsSweep called GetGitLog with an unbounded context (no deadline) - " + + "it runs inside the per-session cancel mutex and must not be able to block Stop/Cancel/Delete " + + "for up to the agentctl client's full 60s timeout") + } +} + +// Regression: handleAgentCompletedLocked must run captureSessionCommitsSweep +// for every completed turn - including one that ends with a pending +// clarification. The pending-clarification branch returns early (deferring +// the workflow transition, captureGitStatusSnapshot, and +// finalizeAutomationRun to a later turn), but the agent process is still +// alive for THIS turn and any commit made during it would otherwise never be +// captured by the live event path alone (see +// TestCaptureSessionCommitsSweep_FillsCommitLivePathMissed). Skipping the +// sweep here silently drops that turn's commits from task_session_commits. +func TestHandleAgentCompleted_CommitSweepRunsEvenWhileClarificationPending(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "t-sweep-clarify", "s-sweep-clarify", "step1") + setSessionExecID(t, repo, "s-sweep-clarify", "exec-1") + + session, err := repo.GetTaskSession(ctx, "s-sweep-clarify") + if err != nil { + t.Fatalf("load session: %v", err) + } + session.BaseCommitSHA = "base-sha" + session.State = models.TaskSessionStateRunning + if err := repo.UpdateTaskSession(ctx, session); err != nil { + t.Fatalf("set base commit and session state: %v", err) + } + + now := time.Now().UTC() + if err := repo.CreateTurn(ctx, &models.Turn{ + ID: "turn-sweep-clarify", TaskSessionID: "s-sweep-clarify", TaskID: "t-sweep-clarify", StartedAt: now, + }); err != nil { + t.Fatalf("create turn: %v", err) + } + if err := repo.CreateMessage(ctx, &models.Message{ + ID: "clarify-sweep-1", TaskSessionID: "s-sweep-clarify", TaskID: "t-sweep-clarify", TurnID: "turn-sweep-clarify", + AuthorType: models.MessageAuthorAgent, Type: "clarification_request", Content: "Which approach?", + CreatedAt: now, + Metadata: map[string]interface{}{"pending_id": "pending-1", "question_id": "q1", "status": "pending"}, + }); err != nil { + t.Fatalf("create pending clarification message: %v", err) + } + + gitLogCalls := 0 + agentMgr := &mockAgentManager{ + repoForExecutionLookup: repo, + getGitLogFunc: func(_ context.Context, sessionID, baseCommit string, _ int, _ string) (*client.GitLogResult, error) { + gitLogCalls++ + if sessionID != "s-sweep-clarify" || baseCommit != "base-sha" { + t.Fatalf("GetGitLog called with (%q, %q), want (s-sweep-clarify, base-sha)", sessionID, baseCommit) + } + return &client.GitLogResult{ + Success: true, + Commits: []*client.GitCommitInfo{ + {CommitSHA: "mid-turn-sha", CommitMessage: "feat: made before the clarification question", CommittedAt: "2026-04-02T03:04:05Z", Insertions: 3}, + }, + }, nil + }, + } + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr) + svc.turnService = &repoBackedTurnService{repo: repo} + + svc.handleAgentCompleted(ctx, watcher.AgentEventData{ + TaskID: "t-sweep-clarify", SessionID: "s-sweep-clarify", AgentExecutionID: "exec-1", + }) + + if gitLogCalls != 1 { + t.Fatalf("GetGitLog calls = %d, want 1 (sweep must run even when a turn ends with pending clarification)", gitLogCalls) + } + commits, err := repo.GetSessionCommits(ctx, "s-sweep-clarify") + if err != nil { + t.Fatalf("GetSessionCommits: %v", err) + } + if len(commits) != 1 || commits[0].CommitSHA != "mid-turn-sha" { + t.Fatalf("commits = %+v, want exactly [mid-turn-sha]", commits) + } +} + +// Regression: handleAgentCompletedLocked's rotated-execution guard (a +// profile switch already replaced this session's live execution before this +// stale agent.completed event arrived) must not skip captureSessionCommitsSweep. +// GetGitLog is resolved by session ID, not execution ID, and a session's +// worktree is shared across its executions (see internal/worktree/Manager), +// so a live GetGitLog call issued through the NEW execution still reads the +// commits the OLD execution made before rotation - the exact commits this +// stale event's own turn would otherwise never get another chance to record. +func TestHandleAgentCompleted_CommitSweepRunsWhenExecutionRotated(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "t-sweep-rotate", "s-sweep-rotate", "step1") + seedExecutorRunning(t, repo, "s-sweep-rotate", "t-sweep-rotate", "exec-new") + + session, err := repo.GetTaskSession(ctx, "s-sweep-rotate") + if err != nil { + t.Fatalf("load session: %v", err) + } + session.BaseCommitSHA = "base-sha" + if err := repo.UpdateTaskSession(ctx, session); err != nil { + t.Fatalf("set base commit: %v", err) + } + + gitLogCalls := 0 + agentMgr := &mockAgentManager{ + repoForExecutionLookup: repo, + getGitLogFunc: func(_ context.Context, sessionID, baseCommit string, _ int, _ string) (*client.GitLogResult, error) { + gitLogCalls++ + if sessionID != "s-sweep-rotate" || baseCommit != "base-sha" { + t.Fatalf("GetGitLog called with (%q, %q), want (s-sweep-rotate, base-sha)", sessionID, baseCommit) + } + return &client.GitLogResult{ + Success: true, + Commits: []*client.GitCommitInfo{ + {CommitSHA: "pre-rotation-sha", CommitMessage: "feat: made by the old execution", CommittedAt: "2026-04-02T03:04:05Z", Insertions: 4}, + }, + }, nil + }, + } + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr) + + // The event's AgentExecutionID ("exec-old") differs from the seeded live + // execution ("exec-new"), so this fires the rotated-execution guard. + svc.handleAgentCompleted(ctx, watcher.AgentEventData{ + TaskID: "t-sweep-rotate", SessionID: "s-sweep-rotate", AgentExecutionID: "exec-old", + }) + + if gitLogCalls != 1 { + t.Fatalf("GetGitLog calls = %d, want 1 (sweep must run even when the rotated-execution guard fires)", gitLogCalls) + } + commits, err := repo.GetSessionCommits(ctx, "s-sweep-rotate") + if err != nil { + t.Fatalf("GetSessionCommits: %v", err) + } + if len(commits) != 1 || commits[0].CommitSHA != "pre-rotation-sha" { + t.Fatalf("commits = %+v, want exactly [pre-rotation-sha]", commits) + } +} + +// Regression: handleAgentCompletedLocked's terminal-session-state guard (the +// session was already deliberately stopped via completeAndStopSession before +// this stale agent.completed event arrived) must not skip +// captureSessionCommitsSweep either. In this scenario the underlying agent +// process is very likely already torn down, so the sweep's GetGitLog call is +// expected to be attempted and to safely no-op (see +// TestCaptureSessionCommitsSweep_AgentNotRunningIsNoop) - the point of this +// test is that the sweep is reached at all, not skipped by the same early +// return that skips the workflow-transition logic. +func TestHandleAgentCompleted_CommitSweepRunsWhenSessionAlreadyTerminal(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "t-sweep-terminal", "s-sweep-terminal", "step1") + + session, err := repo.GetTaskSession(ctx, "s-sweep-terminal") + if err != nil { + t.Fatalf("load session: %v", err) + } + session.BaseCommitSHA = "base-sha" + session.State = models.TaskSessionStateCompleted + if err := repo.UpdateTaskSession(ctx, session); err != nil { + t.Fatalf("set base commit and terminal state: %v", err) + } + + gitLogCalls := 0 + agentMgr := &mockAgentManager{ + getGitLogFunc: func(_ context.Context, sessionID, baseCommit string, _ int, _ string) (*client.GitLogResult, error) { + gitLogCalls++ + if sessionID != "s-sweep-terminal" || baseCommit != "base-sha" { + t.Fatalf("GetGitLog called with (%q, %q), want (s-sweep-terminal, base-sha)", sessionID, baseCommit) + } + // The old execution is torn down by the time this stale event + // arrives - matches the documented "agent not running" contract. + return nil, nil + }, + } + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr) + + svc.handleAgentCompleted(ctx, watcher.AgentEventData{ + TaskID: "t-sweep-terminal", SessionID: "s-sweep-terminal", AgentExecutionID: "exec-old", + }) + + if gitLogCalls != 1 { + t.Fatalf("GetGitLog calls = %d, want 1 (sweep must be attempted even when the terminal-state guard fires)", gitLogCalls) + } +} + +// Regression: captureSessionCommitsSweep must not run while +// handleAgentCompleted holds the per-session cancel-in-flight mutex +// (acquireCancelInFlightGuard) - the same mutex ~20 other call sites across +// internal/orchestrator/ need to serve a user's Stop/Cancel/Delete for this +// session. GetGitLog is a real git shellout through agentctl that can take +// up to the sweep's 10s bound (see +// TestCaptureSessionCommitsSweep_BoundsGetGitLogWithTimeout); holding the +// mutex for that long blocks those operations. This test blocks GetGitLog +// mid-flight and asserts a concurrent caller can still acquire the same +// per-session mutex immediately, rather than waiting behind the sweep. +func TestHandleAgentCompleted_SweepDoesNotBlockCancelInFlightMutex(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "t-sweep-nolock", "s-sweep-nolock", "step1") + + session, err := repo.GetTaskSession(ctx, "s-sweep-nolock") + if err != nil { + t.Fatalf("load session: %v", err) + } + session.BaseCommitSHA = "base-sha" + if err := repo.UpdateTaskSession(ctx, session); err != nil { + t.Fatalf("set base commit: %v", err) + } + + gitLogStarted := make(chan struct{}) + releaseGitLog := make(chan struct{}) + agentMgr := &mockAgentManager{ + getGitLogFunc: func(context.Context, string, string, int, string) (*client.GitLogResult, error) { + close(gitLogStarted) + <-releaseGitLog + return &client.GitLogResult{Success: true}, nil + }, + } + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr) + + done := make(chan struct{}) + go func() { + defer close(done) + svc.handleAgentCompleted(ctx, watcher.AgentEventData{ + TaskID: "t-sweep-nolock", SessionID: "s-sweep-nolock", AgentExecutionID: "exec-1", + }) + }() + + select { + case <-gitLogStarted: + case <-time.After(2 * time.Second): + t.Fatal("GetGitLog was never called") + } + + lock, release := svc.acquireCancelInFlightGuard("s-sweep-nolock") + defer release() + acquired := make(chan struct{}) + go func() { + lock.Lock() + close(acquired) + }() + select { + case <-acquired: + lock.Unlock() + case <-time.After(200 * time.Millisecond): + t.Fatal("acquireCancelInFlightGuard's mutex was held by the in-flight commit sweep - " + + "Stop/Cancel/Delete on this session would block for up to the sweep's full duration") + } + + close(releaseGitLog) + <-done +} diff --git a/apps/backend/internal/orchestrator/event_handlers_git.go b/apps/backend/internal/orchestrator/event_handlers_git.go index 59aae285b1..98efedfbdb 100644 --- a/apps/backend/internal/orchestrator/event_handlers_git.go +++ b/apps/backend/internal/orchestrator/event_handlers_git.go @@ -740,8 +740,11 @@ func pickRejectOption(options []map[string]interface{}) string { return "" } -// handleGitCommitCreated handles git commit events by forwarding them to the frontend. -// In the live model, commits are not persisted to DB - they're only captured at archive time. +// handleGitCommitCreated persists the commit (see persistSessionCommit) and +// forwards it to the frontend. This is the primary write path for +// task_session_commits: it fires on every commit agentctl observes, unlike +// archive capture which only ran once per task and needed the agent process +// still alive to succeed. func (s *Service) handleGitCommitCreated(ctx context.Context, data watcher.GitEventData) { if data.Commit == nil { s.logger.Debug("missing commit data for git commit event", @@ -753,6 +756,18 @@ func (s *Service) handleGitCommitCreated(ctx context.Context, data watcher.GitEv zap.String("task_id", data.TaskID), zap.String("commit_sha", data.Commit.CommitSHA)) + s.persistSessionCommit(ctx, data.SessionID, commitCaptureTriggerLive, &models.SessionCommit{ + CommitSHA: data.Commit.CommitSHA, + ParentSHA: data.Commit.ParentSHA, + AuthorName: data.Commit.AuthorName, + AuthorEmail: data.Commit.AuthorEmail, + CommitMessage: data.Commit.Message, + CommittedAt: parseCommitTime(data.Commit.CommittedAt), + FilesChanged: data.Commit.FilesChanged, + Insertions: data.Commit.Insertions, + Deletions: data.Commit.Deletions, + }) + // Forward commit_created event to WebSocket subject for frontend real-time updates if s.eventBus != nil { event := bus.NewEvent(events.GitEvent, "orchestrator", &lifecycle.GitEventPayload{ diff --git a/apps/backend/internal/orchestrator/event_handlers_git_commit_test.go b/apps/backend/internal/orchestrator/event_handlers_git_commit_test.go new file mode 100644 index 0000000000..97306606f8 --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_git_commit_test.go @@ -0,0 +1,107 @@ +package orchestrator + +import ( + "context" + "testing" + + "github.com/kandev/kandev/internal/agent/runtime/lifecycle" + "github.com/kandev/kandev/internal/orchestrator/watcher" +) + +// Regression: task_session_commits was written only by archive capture, which +// needed the agent process still running and fired at most once per task - +// 13 rows against 74 sessions in production. handleGitCommitCreated is now +// the primary write path: it must persist every observed commit, not merely +// forward it to the WebSocket. +func TestHandleGitCommitCreated_PersistsCommit(t *testing.T) { + ctx := context.Background() + testRepo := setupTestRepo(t) + seedSession(t, testRepo, "t-commit-live", "s-commit-live", "step1") + + svc := createTestService(testRepo, newMockStepGetter(), newMockTaskRepo()) + + svc.handleGitCommitCreated(ctx, watcher.GitEventData{ + TaskID: "t-commit-live", + SessionID: "s-commit-live", + Commit: &lifecycle.GitCommitData{ + CommitSHA: "abc123", + ParentSHA: "def456", + Message: "feat: live capture", + AuthorName: "Kandev Agent", + AuthorEmail: "agent@kandev.local", + FilesChanged: 2, + Insertions: 10, + Deletions: 1, + CommittedAt: "2026-04-02T03:04:05Z", + RepositoryName: "myrepo", + }, + }) + + commits, err := testRepo.GetSessionCommits(ctx, "s-commit-live") + if err != nil { + t.Fatalf("GetSessionCommits: %v", err) + } + if len(commits) != 1 { + t.Fatalf("commit rows = %d, want 1", len(commits)) + } + got := commits[0] + if got.CommitSHA != "abc123" { + t.Errorf("CommitSHA = %q, want abc123", got.CommitSHA) + } + if got.Insertions != 10 || got.Deletions != 1 || got.FilesChanged != 2 { + t.Errorf("stats = (%d, %d, %d), want (2, 10, 1)", got.FilesChanged, got.Insertions, got.Deletions) + } +} + +// Regression: the same commit can now be observed by more than one trigger +// (live event, per-turn sweep, archive). A second observation of the same +// (session_id, commit_sha) must not duplicate the row. +func TestHandleGitCommitCreated_DuplicateEventDoesNotDuplicateRow(t *testing.T) { + ctx := context.Background() + testRepo := setupTestRepo(t) + seedSession(t, testRepo, "t-commit-dup", "s-commit-dup", "step1") + + svc := createTestService(testRepo, newMockStepGetter(), newMockTaskRepo()) + + event := watcher.GitEventData{ + TaskID: "t-commit-dup", + SessionID: "s-commit-dup", + Commit: &lifecycle.GitCommitData{ + CommitSHA: "same-sha", + Message: "feat: observed twice", + CommittedAt: "2026-04-02T03:04:05Z", + }, + } + svc.handleGitCommitCreated(ctx, event) + svc.handleGitCommitCreated(ctx, event) + + commits, err := testRepo.GetSessionCommits(ctx, "s-commit-dup") + if err != nil { + t.Fatalf("GetSessionCommits: %v", err) + } + if len(commits) != 1 { + t.Fatalf("commit rows after duplicate event = %d, want 1", len(commits)) + } +} + +// Regression: handleGitCommitCreated must not panic or block when +// data.Commit is nil (e.g. a malformed upstream event). +func TestHandleGitCommitCreated_NilCommitIsNoop(t *testing.T) { + ctx := context.Background() + testRepo := setupTestRepo(t) + seedSession(t, testRepo, "t-commit-nil", "s-commit-nil", "step1") + + svc := createTestService(testRepo, newMockStepGetter(), newMockTaskRepo()) + svc.handleGitCommitCreated(ctx, watcher.GitEventData{ + TaskID: "t-commit-nil", + SessionID: "s-commit-nil", + }) + + commits, err := testRepo.GetSessionCommits(ctx, "s-commit-nil") + if err != nil { + t.Fatalf("GetSessionCommits: %v", err) + } + if len(commits) != 0 { + t.Fatalf("commit rows = %d, want 0", len(commits)) + } +} diff --git a/apps/backend/internal/orchestrator/event_handlers_test.go b/apps/backend/internal/orchestrator/event_handlers_test.go index a708ddbdb0..8840a7fddb 100644 --- a/apps/backend/internal/orchestrator/event_handlers_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_test.go @@ -244,6 +244,10 @@ func (m *mockTaskRepo) UpdateTaskStateIfSessionState( type mockAgentManager struct { isPassthrough bool isAgentRunning bool + // getGitLogFunc, when non-nil, overrides GetGitLog. Lets tests model a + // commit reconcile sweep (or archive capture) observing new commits, or + // simulate the agent process being gone (nil, nil). + getGitLogFunc func(ctx context.Context, sessionID, baseCommit string, limit int, targetBranch string) (*client.GitLogResult, error) // isAgentRunningFn, when non-nil, overrides isAgentRunning for // IsAgentRunningForSession. Lets tests model state changes mid-sequence // (e.g. stream disconnect between PromptAgent call and queue write). @@ -653,7 +657,10 @@ func (m *mockAgentManager) GetExecutionIDForSession(ctx context.Context, session } return "", fmt.Errorf("no execution found") } -func (m *mockAgentManager) GetGitLog(_ context.Context, _, _ string, _ int, _ string) (*client.GitLogResult, error) { +func (m *mockAgentManager) GetGitLog(ctx context.Context, sessionID, baseCommit string, limit int, targetBranch string) (*client.GitLogResult, error) { + if m.getGitLogFunc != nil { + return m.getGitLogFunc(ctx, sessionID, baseCommit, limit, targetBranch) + } return nil, nil } func (m *mockAgentManager) GetCumulativeDiff(_ context.Context, _, _ string) (*client.CumulativeDiffResult, error) { diff --git a/apps/backend/internal/orchestrator/executor/executor_mocks_test.go b/apps/backend/internal/orchestrator/executor/executor_mocks_test.go index a76214bce4..f1d5ff0414 100644 --- a/apps/backend/internal/orchestrator/executor/executor_mocks_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_mocks_test.go @@ -954,8 +954,8 @@ func (m *mockRepository) GetGitSnapshotsBySession(ctx context.Context, sessionID } // Session Commit operations -func (m *mockRepository) CreateSessionCommit(ctx context.Context, commit *models.SessionCommit) error { - return nil +func (m *mockRepository) CreateSessionCommit(ctx context.Context, commit *models.SessionCommit) (bool, error) { + return true, nil } func (m *mockRepository) GetSessionCommits(ctx context.Context, sessionID string) ([]*models.SessionCommit, error) { return nil, nil diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 6cda06e065..6a729a30f5 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -288,7 +288,7 @@ type sessionExecutorStore interface { CreateGitSnapshot(ctx context.Context, snapshot *models.GitSnapshot) error DeleteLiveMonitorSnapshots(ctx context.Context, sessionID string) error UpsertLatestLiveGitSnapshot(ctx context.Context, snapshot *models.GitSnapshot) error - CreateSessionCommit(ctx context.Context, commit *models.SessionCommit) error + CreateSessionCommit(ctx context.Context, commit *models.SessionCommit) (bool, error) GetSessionCommits(ctx context.Context, sessionID string) ([]*models.SessionCommit, error) DeleteSessionCommit(ctx context.Context, id string) error // Session listing + delete diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index 186e5d1174..c470c4185c 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -2995,7 +2995,7 @@ func (s *Service) CaptureArchiveSnapshot(ctx context.Context, sessionID string) } // Capture commits - baseBranch can be used for merge-base even if baseCommit is empty - if !s.captureArchiveCommits(ctx, sessionID, baseCommit, baseBranch) { + if !s.captureCommitsForTrigger(ctx, sessionID, baseCommit, baseBranch, commitCaptureTriggerArchive) { // Agent not running, skip diff capture as well return nil } @@ -3043,28 +3043,107 @@ func (s *Service) resolveArchiveBaseCommitAndBranch(ctx context.Context, session return "", baseBranch, nil } -// captureArchiveCommits fetches and saves commits from baseCommit to HEAD. -// If targetBranch is provided, uses dynamic merge-base for accurate filtering. -// Returns false if the agent is not running (caller should skip remaining capture). -func (s *Service) captureArchiveCommits(ctx context.Context, sessionID, baseCommit, targetBranch string) bool { +// captureCommitsForTrigger fetches commits from baseCommit to HEAD and +// persists them tagged with trigger (see persistSessionCommit). If +// targetBranch is provided, uses dynamic merge-base for accurate filtering. +// Returns false if the agent is not running (caller should skip remaining +// capture that also needs it, e.g. archive's diff capture). +func (s *Service) captureCommitsForTrigger(ctx context.Context, sessionID, baseCommit, targetBranch string, trigger commitCaptureTrigger) bool { logResult, err := s.agentManager.GetGitLog(ctx, sessionID, baseCommit, 0, targetBranch) // 0 = no limit if err != nil { - s.logger.Warn("failed to capture git log for archive", + s.logger.Warn("failed to capture git log", zap.String("session_id", sessionID), + zap.String("trigger", string(trigger)), zap.Error(err)) return true // Continue with diff capture even if log fails } if logResult == nil { - s.logger.Debug("agent not running, skipping archive snapshot capture", - zap.String("session_id", sessionID)) + s.logger.Debug("agent not running, skipping commit capture", + zap.String("session_id", sessionID), + zap.String("trigger", string(trigger))) return false } + s.recordCommitCaptureFetchFailures(sessionID, trigger, logResult) if logResult.Success && len(logResult.Commits) > 0 { - s.saveArchiveCommits(ctx, sessionID, logResult.Commits) + s.persistCommits(ctx, sessionID, trigger, logResult.Commits) } return true } +// recordCommitCaptureFetchFailures makes a GetGitLog-level failure +// observable instead of silently dropping it. logResult can fail two ways +// that are otherwise indistinguishable from "nothing to capture this time": +// a total result-level failure (logResult.Success == false, single-repo git +// command failure or every repo failed in a multi-repo fan-out), or a +// partial multi-repo failure (Success == true because at least one repo +// succeeded, but PerRepoErrors names the ones that didn't). A multi-repo +// total failure sets BOTH Success == false AND populates PerRepoErrors for +// every repo (see agentctl/server/api/git.go's mergeGitLogResults), so this +// records one failure per PerRepoErrors entry when present, and only falls +// back to counting the top-level Success == false once when PerRepoErrors is +// empty (the single-repo case) - never both, to avoid double-counting. +func (s *Service) recordCommitCaptureFetchFailures(sessionID string, trigger commitCaptureTrigger, logResult *client.GitLogResult) { + if len(logResult.PerRepoErrors) > 0 { + for _, repoErr := range logResult.PerRepoErrors { + recordCommitCaptureFetchFailed(trigger) + s.logger.Warn("git log capture failed for repository", + zap.String("session_id", sessionID), + zap.String("trigger", string(trigger)), + zap.String("repository_name", repoErr.RepositoryName), + zap.String("error", repoErr.Error)) + } + return + } + if !logResult.Success { + recordCommitCaptureFetchFailed(trigger) + s.logger.Warn("git log capture reported failure", + zap.String("session_id", sessionID), + zap.String("trigger", string(trigger)), + zap.String("error", logResult.Error)) + } +} + +// captureSessionCommitsSweep reconciles task_session_commits against +// agentctl while the agent process is still alive. The live event-driven +// path (handleGitCommitCreated) can miss a commit that was pushed just +// before agentctl's next poll tick (see filterLocalCommits' upstream-commit +// filter), and this sweep is the only trigger guaranteed to run with the +// agent still running - unlike archive capture, whose GetGitLog call +// usually finds the agent process already gone. Cost is one +// `git log --shortstat base..HEAD`, the same command agentctl's own poller +// already runs on every tick. Best-effort: errors are logged and swallowed, +// matching captureGitStatusSnapshot alongside which this is called. +// +// Called from handleAgentCompletedLocked while it holds the per-session +// cancel-in-flight mutex (acquireCancelInFlightGuard) - the same mutex +// stopTaskSessionForCoordinator and DeleteSession need to serve a user's +// Stop/Cancel/Delete for this session. GetGitLog is a real git shellout +// through agentctl with no bound of its own beyond the agentctl HTTP +// client's 60s default timeout, so an undecorated ctx here could hold that +// mutex - and therefore Stop/Cancel/Delete - for up to a minute on every +// single turn completion. Bound it the same way the pre-existing archive +// capture call site already does (service_tasks.go's CaptureArchiveSnapshot: +// "Use a bounded timeout to prevent blocking the archive operation if +// agentctl is stuck"). A timeout here falls through the existing +// warn-and-continue path in captureCommitsForTrigger; archive capture +// remains the reconciliation pass, so nothing is permanently lost. +func (s *Service) captureSessionCommitsSweep(ctx context.Context, sessionID string) { + sweepCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + + baseCommit, baseBranch, err := s.resolveArchiveBaseCommitAndBranch(sweepCtx, sessionID) + if err != nil { + s.logger.Debug("failed to resolve base commit for commit sweep", + zap.String("session_id", sessionID), + zap.Error(err)) + return + } + if baseCommit == "" && baseBranch == "" { + return + } + s.captureCommitsForTrigger(sweepCtx, sessionID, baseCommit, baseBranch, commitCaptureTriggerSweep) +} + // captureGitStatusSnapshot fetches the current (cached) git status from agentctl // and saves it as a DB snapshot. Called when a session's execution completes so // the status is available when clients subscribe later. @@ -3217,11 +3296,13 @@ func parseCommitTime(s string) time.Time { return t.UTC() } -// saveArchiveCommits persists commits to the database for archive purposes. -func (s *Service) saveArchiveCommits(ctx context.Context, sessionID string, commits []*client.GitCommitInfo) { +// persistCommits persists a batch of commits fetched via GetGitLog, tagged +// with trigger. Shared by archive capture and the per-turn reconcile sweep +// (captureSessionCommitsSweep) - see persistSessionCommit for the +// idempotent single-commit write and writer-health counters. +func (s *Service) persistCommits(ctx context.Context, sessionID string, trigger commitCaptureTrigger, commits []*client.GitCommitInfo) { for _, commit := range commits { - if err := s.repo.CreateSessionCommit(ctx, &models.SessionCommit{ - SessionID: sessionID, + s.persistSessionCommit(ctx, sessionID, trigger, &models.SessionCommit{ CommitSHA: commit.CommitSHA, ParentSHA: commit.ParentSHA, AuthorName: commit.AuthorName, @@ -3231,18 +3312,45 @@ func (s *Service) saveArchiveCommits(ctx context.Context, sessionID string, comm FilesChanged: commit.FilesChanged, Insertions: commit.Insertions, Deletions: commit.Deletions, - }); err != nil { - s.logger.Warn("failed to save commit for archive", - zap.String("session_id", sessionID), - zap.String("commit_sha", commit.CommitSHA), - zap.Error(err)) - } + }) } - s.logger.Debug("saved archive commits", + s.logger.Debug("persisted commits", zap.String("session_id", sessionID), + zap.String("trigger", string(trigger)), zap.Int("count", len(commits))) } +// persistSessionCommit is the single write path shared by the live commit +// event (handleGitCommitCreated), the per-turn reconcile sweep +// (captureSessionCommitsSweep), and archive capture (saveArchiveCommits). It +// records writer-health counters (commit_capture_*, see +// commit_capture_metrics.go) so a trigger that silently stops firing is +// observable, and treats CreateSessionCommit reporting a duplicate +// (session_id, commit_sha) as a normal outcome, not a failure - the same +// underlying git commit is expected to be observed by more than one trigger. +func (s *Service) persistSessionCommit(ctx context.Context, sessionID string, trigger commitCaptureTrigger, commit *models.SessionCommit) { + if s.repo == nil { + return + } + recordCommitCaptureObserved(trigger) + commit.SessionID = sessionID + inserted, err := s.repo.CreateSessionCommit(ctx, commit) + if err != nil { + recordCommitCaptureFailed(trigger) + s.logger.Warn("failed to persist session commit", + zap.String("session_id", sessionID), + zap.String("commit_sha", commit.CommitSHA), + zap.String("trigger", string(trigger)), + zap.Error(err)) + return + } + if inserted { + recordCommitCaptureInserted(trigger) + } else { + recordCommitCaptureDuplicate(trigger) + } +} + // PromptTask sends a follow-up prompt to a running agent for a task session. // If planMode is true, a plan mode prefix is prepended to the prompt. // Attachments (images) are passed through to the agent if provided. diff --git a/apps/backend/internal/plugins/host_data_mappers.go b/apps/backend/internal/plugins/host_data_mappers.go index 8d7714708d..a117e5cf1a 100644 --- a/apps/backend/internal/plugins/host_data_mappers.go +++ b/apps/backend/internal/plugins/host_data_mappers.go @@ -225,12 +225,22 @@ func messageModelToDTO(m *taskmodels.Message) pluginsdk.Message { } } +// sessionCodeStatsModelToDTO converts the internal analytics model - which +// represents "capture wasn't active yet for this session" as a nil +// LinesAddedCommitted/LinesDeletedCommitted pair - into the public plugin +// SDK's DTO, which represents the same fact as CommittedLinesAvailable == +// false with both fields reading 0 (see pluginsdk.SessionCodeStats' doc for +// why the SDK doesn't use pointers here). func sessionCodeStatsModelToDTO(s *analyticsmodels.SessionCodeStats) pluginsdk.SessionCodeStats { - return pluginsdk.SessionCodeStats{ + dto := pluginsdk.SessionCodeStats{ SessionID: s.SessionID, - LinesAddedCommitted: s.LinesAddedCommitted, - LinesDeletedCommitted: s.LinesDeletedCommitted, LinesAddedPeakPending: s.LinesAddedPeakPending, LinesDeletedPeakPending: s.LinesDeletedPeakPending, } + if s.LinesAddedCommitted != nil && s.LinesDeletedCommitted != nil { + dto.LinesAddedCommitted = *s.LinesAddedCommitted + dto.LinesDeletedCommitted = *s.LinesDeletedCommitted + dto.CommittedLinesAvailable = true + } + return dto } diff --git a/apps/backend/internal/plugins/host_data_test.go b/apps/backend/internal/plugins/host_data_test.go index e2cc60e74c..2ad79b1764 100644 --- a/apps/backend/internal/plugins/host_data_test.go +++ b/apps/backend/internal/plugins/host_data_test.go @@ -19,6 +19,8 @@ import ( "google.golang.org/grpc/status" ) +func int64Ptr(v int64) *int64 { return &v } + // ── fakes for the narrow Host data API interfaces ─────────────────────── type fakeTaskDataSource struct { @@ -755,7 +757,7 @@ func TestPluginHost_Sessions_PaginatesBeforeResolvingACPSessionID(t *testing.T) func TestPluginHost_SessionsCodeStats_DelegatesToAnalyticsService(t *testing.T) { d := newTestDataHost(manifest.Capabilities{APIRead: []string{"sessions"}}) d.codeStats.stats = []*analyticsmodels.SessionCodeStats{ - {SessionID: "session-1", LinesAddedCommitted: 10, LinesDeletedCommitted: 2, LinesAddedPeakPending: 5, LinesDeletedPeakPending: 1}, + {SessionID: "session-1", LinesAddedCommitted: int64Ptr(10), LinesDeletedCommitted: int64Ptr(2), LinesAddedPeakPending: 5, LinesDeletedPeakPending: 1}, } filter := pluginsdk.SessionFilter{TaskIDs: []string{"task-1"}, WorkspaceIDs: []string{"ws-1"}, States: []string{"RUNNING"}} @@ -780,7 +782,8 @@ func TestPluginHost_SessionsCodeStats_DelegatesToAnalyticsService(t *testing.T) if d.codeStats.lastFilter.Limit != 11 { t.Errorf("filter.Limit = %d, want 11 (requested 10 + 1 probe row)", d.codeStats.lastFilter.Limit) } - if len(stats) != 1 || stats[0].SessionID != "session-1" || stats[0].LinesAddedCommitted != 10 { + if len(stats) != 1 || stats[0].SessionID != "session-1" || + !stats[0].CommittedLinesAvailable || stats[0].LinesAddedCommitted != 10 { t.Fatalf("CodeStats() = %+v, want session-1 passed through unchanged", stats) } if info == nil || info.HasMore { diff --git a/apps/backend/internal/plugins/host_data_wire_test.go b/apps/backend/internal/plugins/host_data_wire_test.go index d41cf96b61..3bf8d1bd50 100644 --- a/apps/backend/internal/plugins/host_data_wire_test.go +++ b/apps/backend/internal/plugins/host_data_wire_test.go @@ -105,9 +105,12 @@ func seedSessionsWireFixture(d *testDataHost, n int) { if i == 0 { sessions[i].Metadata = map[string]any{"acp": map[string]any{"session_id": "acp-session-0"}} } + linesAdded := int64(10 * (i + 1)) + linesDeleted := int64(i + 1) stats[i] = &analyticsmodels.SessionCodeStats{ - SessionID: id, - LinesAddedCommitted: int64(10 * (i + 1)), + SessionID: id, + LinesAddedCommitted: &linesAdded, + LinesDeletedCommitted: &linesDeleted, } } d.tasks.sessionsByTask = map[string][]*taskmodels.TaskSession{"task-1": sessions} @@ -156,6 +159,7 @@ func TestPluginHostData_Wire_SessionsRoundTrip(t *testing.T) { require.NoError(t, err) require.Len(t, stats, 1) require.Equal(t, "session-0", stats[0].SessionID) + require.True(t, stats[0].CommittedLinesAvailable) require.Equal(t, int64(10), stats[0].LinesAddedCommitted) } diff --git a/apps/backend/internal/task/handlers/process_handlers_test.go b/apps/backend/internal/task/handlers/process_handlers_test.go index d1930e3e71..fb0260772c 100644 --- a/apps/backend/internal/task/handlers/process_handlers_test.go +++ b/apps/backend/internal/task/handlers/process_handlers_test.go @@ -532,8 +532,8 @@ func (m *mockRepository) GetGitSnapshotsBySession(ctx context.Context, sessionID } // Session Commit operations -func (m *mockRepository) CreateSessionCommit(ctx context.Context, commit *models.SessionCommit) error { - return nil +func (m *mockRepository) CreateSessionCommit(ctx context.Context, commit *models.SessionCommit) (bool, error) { + return true, nil } func (m *mockRepository) GetSessionCommits(ctx context.Context, sessionID string) ([]*models.SessionCommit, error) { return nil, nil diff --git a/apps/backend/internal/task/repository/interface.go b/apps/backend/internal/task/repository/interface.go index 4e74abff93..c8309b9018 100644 --- a/apps/backend/internal/task/repository/interface.go +++ b/apps/backend/internal/task/repository/interface.go @@ -310,7 +310,7 @@ type GitSnapshotRepository interface { GetLatestGitSnapshotsBySessionIDs(ctx context.Context, sessionIDs []string) (map[string]*models.GitSnapshot, error) GetFirstGitSnapshot(ctx context.Context, sessionID string) (*models.GitSnapshot, error) GetGitSnapshotsBySession(ctx context.Context, sessionID string, limit int) ([]*models.GitSnapshot, error) - CreateSessionCommit(ctx context.Context, commit *models.SessionCommit) error + CreateSessionCommit(ctx context.Context, commit *models.SessionCommit) (bool, error) GetSessionCommits(ctx context.Context, sessionID string) ([]*models.SessionCommit, error) GetLatestSessionCommit(ctx context.Context, sessionID string) (*models.SessionCommit, error) DeleteSessionCommit(ctx context.Context, id string) error diff --git a/apps/backend/internal/task/repository/sqlite/base_migrations.go b/apps/backend/internal/task/repository/sqlite/base_migrations.go index e0b111aa60..de36e4ac58 100644 --- a/apps/backend/internal/task/repository/sqlite/base_migrations.go +++ b/apps/backend/internal/task/repository/sqlite/base_migrations.go @@ -233,6 +233,16 @@ func (r *Repository) runMigrations() error { r.migrate.Apply("task_session_messages.updated_at.backfill", `UPDATE task_session_messages SET updated_at = created_at WHERE updated_at IS NULL`) r.migrate.Apply("idx_messages_session_updated", `CREATE INDEX IF NOT EXISTS idx_messages_session_updated ON task_session_messages(task_session_id, updated_at)`) + // task_session_commits gains a uniqueness constraint before its writer + // starts firing from more than just archive capture (CreateSessionCommit + // was previously a plain INSERT). Must dedupe existing duplicates first: + // CREATE UNIQUE INDEX fails on a duplicate pair, and MigrateLogger.Apply + // swallows non-"already exists" errors, so an unhandled duplicate would + // silently leave both the index and every future ON CONFLICT missing. + if err := r.migrateSessionCommitsDedupeAndActivation(); err != nil { + return err + } + // Backfill the per-session cost/token columns. Runs after the gated // task_sessions rebuilds above so it repairs legacy DBs whose schema can no // longer trigger a rebuild (see migrateSessionsAddCostColumns). @@ -680,6 +690,94 @@ func (r *Repository) backfillExecutorsRunningFromTaskSessions() error { return nil } +// commitCaptureActivatedAtMetaKey is the kandev_meta key published by +// migrateSessionCommitsDedupeAndActivation. +const commitCaptureActivatedAtMetaKey = "commit_capture_activated_at" + +// migrateSessionCommitsDedupeAndActivation enforces uniqueness on +// task_session_commits(session_id, commit_sha) and publishes the point in +// time commit capture started firing from more than just archive capture, so +// downstream readers (the Rill extract, ListSessionCodeStats) can tell +// "capture wasn't running yet" apart from "session made zero commits" - +// both previously read as a plain 0. +// +// Dedup keeps the earliest-observed row per (session_id, commit_sha), ties +// broken by id: task_session_commits is an append-only observation ledger +// (a rebase/squash adds new SHAs, it never retroactively changes which row +// was first seen), so "earliest seen" is the row future replays preserve. +// +// Rebase/squash decision: immutable observation history, not the final +// branch object set. A rebase that drops a previously-observed commit SHA +// from reachable history leaves that row in place - it is not deleted or +// reconciled against the current branch. Same tradeoff the task brief +// already accepts for summed commit diffstats counting churn and reverts +// (task_session_git_snapshots deltas are the net-branch-growth answer; +// commit rows are the observation-history answer, published side by side, +// not merged into one number). Live capture makes this more visible than +// the old archive-only design (which only ever ran GetGitLog once, at +// archive time, so it naturally reflected whatever was reachable from HEAD +// at that instant) - continuous capture can now persist a commit that a +// later rebase makes unreachable. Accepted rather than reconciled: pruning +// on every rebase/force-push would require diffing against live git state +// on every sweep, and "what got captured" is itself useful observability +// (e.g. abandoned work), not just noise. +// +// Must run before CreateSessionCommit starts firing from more than archive - +// CREATE UNIQUE INDEX fails on an existing duplicate pair, and +// MigrateLogger.Apply swallows non-"already exists" errors, so an unhandled +// duplicate would silently leave both the index and every future +// ON CONFLICT missing. That is exactly why these two statements, unlike most +// migrations in this file, do NOT go through r.migrate.Apply: the writer's +// ON CONFLICT (session_id, commit_sha) target hard-requires this index to +// exist, so a failure here must abort boot (propagated below) rather than +// leave every future commit insert failing silently forever. +func (r *Repository) migrateSessionCommitsDedupeAndActivation() error { + if _, err := r.db.Exec(` + DELETE FROM task_session_commits + WHERE id NOT IN ( + SELECT id FROM ( + SELECT id, + ROW_NUMBER() OVER ( + PARTITION BY session_id, commit_sha + ORDER BY created_at ASC, id ASC + ) AS rn + FROM task_session_commits + ) ranked + WHERE rn = 1 + ) + `); err != nil { + return fmt.Errorf("dedupe task_session_commits: %w", err) + } + if _, err := r.db.Exec( + `CREATE UNIQUE INDEX IF NOT EXISTS uniq_session_commits_session_sha ON task_session_commits(session_id, commit_sha)`, + ); err != nil { + return fmt.Errorf("create uniq_session_commits_session_sha: %w", err) + } + if r.log != nil { + r.log.Info("migration applied", zap.String("name", "task_session_commits.dedupe_and_unique_index")) + } + + // kandev_meta already exists by the time repository migrations run in + // production (persistence.Provide creates it before opening any + // repository), but repo-level tests build a bare DB via NewWithDB where + // it does not, so recreate it defensively. + if _, err := r.db.Exec(` + CREATE TABLE IF NOT EXISTS kandev_meta ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL DEFAULT '' + )`); err != nil { + return fmt.Errorf("ensure kandev_meta: %w", err) + } + now := time.Now().UTC().Format(time.RFC3339Nano) + if _, err := r.db.Exec(r.db.Rebind(` + INSERT INTO kandev_meta (key, value) VALUES (?, ?) + ON CONFLICT(key) DO NOTHING + `), commitCaptureActivatedAtMetaKey, now); err != nil { + return fmt.Errorf("write %s: %w", commitCaptureActivatedAtMetaKey, err) + } + return nil +} + // migrateSessionsRemoveAgentExecutionID drops the agent_execution_id and // container_id columns from task_sessions. After this migration, executors_running // is the single source of truth for both fields — no more denormalization. diff --git a/apps/backend/internal/task/repository/sqlite/git_snapshots.go b/apps/backend/internal/task/repository/sqlite/git_snapshots.go index 6ef466c5d1..eaac48121d 100644 --- a/apps/backend/internal/task/repository/sqlite/git_snapshots.go +++ b/apps/backend/internal/task/repository/sqlite/git_snapshots.go @@ -332,7 +332,20 @@ func (r *Repository) GetGitSnapshotsBySession(ctx context.Context, sessionID str } // CreateSessionCommit inserts a new commit record into the database. -func (r *Repository) CreateSessionCommit(ctx context.Context, commit *models.SessionCommit) error { +// Idempotent on (session_id, commit_sha): a commit already observed for this +// session is silently skipped rather than duplicated, since it can now be +// reported from more than one trigger (live commit events, the per-turn +// reconcile sweep, and archive capture) for the same underlying git commit. +// The returned bool reports whether a new row was actually inserted, so +// callers can distinguish a fresh observation from a re-observed duplicate +// for writer-health counters. +// +// pre_commit_snapshot_id / post_commit_snapshot_id are always written empty +// (commit.PreCommitSnapshotID / PostCommitSnapshotID are never populated by +// any caller). No writer for either column exists anywhere in this codebase; +// they do not link a commit row to the task_session_git_snapshots chain. +// Treat them as reserved/unpopulated, not as live data. +func (r *Repository) CreateSessionCommit(ctx context.Context, commit *models.SessionCommit) (bool, error) { if commit.ID == "" { commit.ID = uuid.New().String() } @@ -340,18 +353,27 @@ func (r *Repository) CreateSessionCommit(ctx context.Context, commit *models.Ses commit.CreatedAt = time.Now().UTC() } - _, err := r.db.ExecContext(ctx, r.db.Rebind(` + result, err := r.db.ExecContext(ctx, r.db.Rebind(` INSERT INTO task_session_commits ( id, session_id, commit_sha, parent_sha, author_name, author_email, commit_message, committed_at, pre_commit_snapshot_id, post_commit_snapshot_id, files_changed, insertions, deletions, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (session_id, commit_sha) DO NOTHING `), commit.ID, commit.SessionID, commit.CommitSHA, commit.ParentSHA, commit.AuthorName, commit.AuthorEmail, commit.CommitMessage, commit.CommittedAt, commit.PreCommitSnapshotID, commit.PostCommitSnapshotID, commit.FilesChanged, commit.Insertions, commit.Deletions, commit.CreatedAt) + if err != nil { + return false, err + } - return err + // RowsAffected returns 0 on a conflict-skipped row in both SQLite and Postgres. + rows, err := result.RowsAffected() + if err != nil { + return false, err + } + return rows > 0, nil } // GetSessionCommits retrieves all commits for a session, ordered by committed_at descending. diff --git a/apps/backend/internal/task/repository/sqlite/git_snapshots_postgres_test.go b/apps/backend/internal/task/repository/sqlite/git_snapshots_postgres_test.go index fb6dc10172..a42c673e67 100644 --- a/apps/backend/internal/task/repository/sqlite/git_snapshots_postgres_test.go +++ b/apps/backend/internal/task/repository/sqlite/git_snapshots_postgres_test.go @@ -141,16 +141,39 @@ func TestPostgresSessionCommitLifecycle(t *testing.T) { FilesChanged: 9, Insertions: 123, Deletions: 45, CreatedAt: time.Date(2026, 4, 2, 3, 4, 6, 0, time.UTC), } - if err := repo.CreateSessionCommit(ctx, want); err != nil { + inserted, err := repo.CreateSessionCommit(ctx, want) + if err != nil { t.Fatalf("CreateSessionCommit: %v", err) } + if !inserted { + t.Error("CreateSessionCommit(want) inserted = false, want true (first observation)") + } got, err := repo.GetLatestSessionCommit(ctx, "session-commit-pg") if err != nil { t.Fatalf("GetLatestSessionCommit: %v", err) } assertSessionCommitEqual(t, got, want) - if err := repo.CreateSessionCommit(ctx, &models.SessionCommit{ + // The unique index on (session_id, commit_sha) must hold on Postgres too + // (ADR-0027 parity): the same commit observed by a second trigger (live + // event, sweep, archive) is a no-op, not a duplicate row or an error. + dupInserted, err := repo.CreateSessionCommit(ctx, &models.SessionCommit{ + ID: "commit-pg-duplicate", SessionID: "session-commit-pg", CommitSHA: want.CommitSHA, + CommittedAt: want.CommittedAt, + }) + if err != nil { + t.Fatalf("CreateSessionCommit(duplicate): %v", err) + } + if dupInserted { + t.Error("CreateSessionCommit(duplicate) inserted = true, want false (ON CONFLICT DO NOTHING)") + } + if got := countRows(t, repo, + `SELECT COUNT(1) FROM task_session_commits WHERE session_id = ? AND commit_sha = ?`, + "session-commit-pg", want.CommitSHA); got != 1 { + t.Errorf("rows for %s after duplicate insert = %d, want 1", want.CommitSHA, got) + } + + if _, err := repo.CreateSessionCommit(ctx, &models.SessionCommit{ ID: "commit-pg-older", SessionID: "session-commit-pg", CommitSHA: "older", CommittedAt: want.CommittedAt.Add(-time.Hour), }); err != nil { @@ -176,6 +199,98 @@ func TestPostgresSessionCommitLifecycle(t *testing.T) { } } +// TestPostgresSessionCommitsDedupeAndActivationMigration mirrors +// TestSessionCommitsDedupeAndActivationMigration (session_commits_migration_test.go) +// on Postgres: migrateSessionCommitsDedupeAndActivation is dialect-sensitive +// (ROW_NUMBER() OVER (PARTITION BY ...), CREATE UNIQUE INDEX IF NOT EXISTS, +// ON CONFLICT(key) DO NOTHING), and the SQLite test asserts index presence +// through sqlite_master, which does not exist on Postgres — this asserts the +// same end state through pg_indexes instead (ADR-0027 parity). +func TestPostgresSessionCommitsDedupeAndActivationMigration(t *testing.T) { + db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) + repo, err := NewWithDB(db, db, nil) + if err != nil { + t.Fatalf("init postgres schema: %v", err) + } + ctx := context.Background() + seedPostgresTask(t, repo, "task-commits-migration-pg") + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "session-commits-migration-pg", TaskID: "task-commits-migration-pg", + State: models.TaskSessionStateWaitingForInput, + }); err != nil { + t.Fatalf("CreateTaskSession: %v", err) + } + + // Simulate a legacy database: no unique index yet, and a duplicate + // (session_id, commit_sha) pair that a plain INSERT (pre-ON CONFLICT) + // could have produced, e.g. via a re-run archive capture. + if _, err := repo.db.Exec(`DROP INDEX IF EXISTS uniq_session_commits_session_sha`); err != nil { + t.Fatalf("drop unique index to simulate legacy schema: %v", err) + } + + earlier := time.Date(2026, 4, 1, 0, 0, 0, 0, time.UTC) + later := time.Date(2026, 4, 2, 0, 0, 0, 0, time.UTC) + insertRawCommit(t, repo, "commit-earliest-pg", "session-commits-migration-pg", "dup-sha-pg", earlier) + insertRawCommit(t, repo, "commit-latest-pg", "session-commits-migration-pg", "dup-sha-pg", later) + insertRawCommit(t, repo, "commit-unique-pg", "session-commits-migration-pg", "solo-sha-pg", later) + + if got := countRows(t, repo, + `SELECT COUNT(1) FROM task_session_commits WHERE session_id = ?`, + "session-commits-migration-pg"); got != 3 { + t.Fatalf("pre-migration commit rows = %d, want 3", got) + } + + if err := repo.runMigrations(); err != nil { + t.Fatalf("replay migrations: %v", err) + } + + var indexName string + if err := repo.db.Get(&indexName, ` + SELECT indexname FROM pg_indexes WHERE indexname = 'uniq_session_commits_session_sha' + `); err != nil { + t.Fatalf("uniq_session_commits_session_sha index is missing: %v", err) + } + + if got := countRows(t, repo, + `SELECT COUNT(1) FROM task_session_commits WHERE session_id = ? AND commit_sha = ?`, + "session-commits-migration-pg", "dup-sha-pg"); got != 1 { + t.Fatalf("duplicate (session_id, commit_sha) rows after migration = %d, want 1", got) + } + + var survivorID string + if err := repo.db.Get(&survivorID, repo.db.Rebind(` + SELECT id FROM task_session_commits WHERE session_id = ? AND commit_sha = ? + `), "session-commits-migration-pg", "dup-sha-pg"); err != nil { + t.Fatalf("select dedupe survivor: %v", err) + } + if survivorID != "commit-earliest-pg" { + t.Errorf("dedupe kept %q, want commit-earliest-pg (earliest created_at)", survivorID) + } + + activatedAt := readCommitCaptureActivatedAt(t, repo) + if activatedAt == "" { + t.Fatal("commit_capture_activated_at was not published") + } + if _, err := time.Parse(time.RFC3339Nano, activatedAt); err != nil { + t.Errorf("commit_capture_activated_at = %q, not RFC3339Nano: %v", activatedAt, err) + } + + // Replay: must not error, must not duplicate the index, and must not + // move the activation marker (legacy rows must stay pinned to the + // original activation instant, not silently drift on every boot). + if err := repo.runMigrations(); err != nil { + t.Fatalf("replay migrations twice: %v", err) + } + if got := readCommitCaptureActivatedAt(t, repo); got != activatedAt { + t.Errorf("commit_capture_activated_at changed on replay: %q -> %q", activatedAt, got) + } + if got := countRows(t, repo, + `SELECT COUNT(1) FROM task_session_commits WHERE session_id = ?`, + "session-commits-migration-pg"); got != 2 { + t.Fatalf("commit rows after second replay = %d, want 2 (no re-duplication)", got) + } +} + func TestPostgresEnvironmentAndExecutorProfileLifecycle(t *testing.T) { db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) repo, err := NewWithDB(db, db, nil) diff --git a/apps/backend/internal/task/repository/sqlite/git_snapshots_test.go b/apps/backend/internal/task/repository/sqlite/git_snapshots_test.go index 307f00f6bc..56a9274f7a 100644 --- a/apps/backend/internal/task/repository/sqlite/git_snapshots_test.go +++ b/apps/backend/internal/task/repository/sqlite/git_snapshots_test.go @@ -665,7 +665,7 @@ func TestGitSnapshotsAndCommitsCascadeOnSessionDelete(t *testing.T) { }); err != nil { t.Fatalf("CreateGitSnapshot: %v", err) } - if err := repo.CreateSessionCommit(ctx, &models.SessionCommit{ + if _, err := repo.CreateSessionCommit(ctx, &models.SessionCommit{ ID: "cascade-commit", SessionID: "session-cascade", CommitSHA: "abc123", CommittedAt: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC), }); err != nil { @@ -707,7 +707,7 @@ func TestCreateSessionCommitRoundTripsEveryField(t *testing.T) { Deletions: 45, CreatedAt: time.Date(2026, 4, 2, 3, 4, 6, 0, time.UTC), } - if err := repo.CreateSessionCommit(ctx, want); err != nil { + if _, err := repo.CreateSessionCommit(ctx, want); err != nil { t.Fatalf("CreateSessionCommit: %v", err) } @@ -779,7 +779,7 @@ func TestCreateSessionCommitDefaultsIDAndCreatedAt(t *testing.T) { SessionID: "session-commit-defaults", CommitSHA: "deadbeef", CommittedAt: time.Date(2026, 4, 2, 3, 4, 5, 0, time.UTC), } - if err := repo.CreateSessionCommit(ctx, commit); err != nil { + if _, err := repo.CreateSessionCommit(ctx, commit); err != nil { t.Fatalf("CreateSessionCommit: %v", err) } if commit.ID == "" { @@ -797,6 +797,57 @@ func TestCreateSessionCommitDefaultsIDAndCreatedAt(t *testing.T) { } } +// Regression: the same underlying git commit can now be observed by more +// than one trigger (live event, per-turn sweep, archive capture). A second +// CreateSessionCommit for the same (session_id, commit_sha) must be a no-op +// - not a duplicate row, and not an error - and must preserve the original +// observation's created_at rather than overwriting it with the replay time. +func TestCreateSessionCommitIsIdempotentOnSessionAndSHA(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + seedSessionForGit(t, repo, "task-commit-idempotent", "session-commit-idempotent") + + first := &models.SessionCommit{ + ID: "commit-first", SessionID: "session-commit-idempotent", CommitSHA: "shared-sha", + CommitMessage: "feat: first observation", + CommittedAt: time.Date(2026, 4, 2, 3, 4, 5, 0, time.UTC), + } + inserted, err := repo.CreateSessionCommit(ctx, first) + if err != nil { + t.Fatalf("CreateSessionCommit(first): %v", err) + } + if !inserted { + t.Error("CreateSessionCommit(first) inserted = false, want true") + } + + second := &models.SessionCommit{ + ID: "commit-second", SessionID: "session-commit-idempotent", CommitSHA: "shared-sha", + CommitMessage: "feat: re-observed by a different trigger", + CommittedAt: time.Date(2026, 4, 2, 4, 0, 0, 0, time.UTC), + } + inserted, err = repo.CreateSessionCommit(ctx, second) + if err != nil { + t.Fatalf("CreateSessionCommit(second): %v", err) + } + if inserted { + t.Error("CreateSessionCommit(second) inserted = true, want false (ON CONFLICT DO NOTHING)") + } + + listed, err := repo.GetSessionCommits(ctx, "session-commit-idempotent") + if err != nil { + t.Fatalf("GetSessionCommits: %v", err) + } + if len(listed) != 1 { + t.Fatalf("commit rows = %d, want 1", len(listed)) + } + if listed[0].ID != "commit-first" { + t.Errorf("surviving row = %q, want commit-first (the original observation)", listed[0].ID) + } + if listed[0].CommitMessage != first.CommitMessage { + t.Errorf("CommitMessage = %q, want the first observation's message unchanged", listed[0].CommitMessage) + } +} + func TestGetSessionCommitsOrdersByCommittedAtDescending(t *testing.T) { repo := newRepoForSessionTests(t) ctx := context.Background() @@ -805,14 +856,14 @@ func TestGetSessionCommitsOrdersByCommittedAtDescending(t *testing.T) { base := time.Date(2026, 4, 10, 0, 0, 0, 0, time.UTC) for i, id := range []string{"c1", "c2", "c3"} { - if err := repo.CreateSessionCommit(ctx, &models.SessionCommit{ + if _, err := repo.CreateSessionCommit(ctx, &models.SessionCommit{ ID: id, SessionID: "session-commit-order", CommitSHA: id, CommittedAt: base.Add(time.Duration(i) * time.Hour), }); err != nil { t.Fatalf("CreateSessionCommit(%s): %v", id, err) } } - if err := repo.CreateSessionCommit(ctx, &models.SessionCommit{ + if _, err := repo.CreateSessionCommit(ctx, &models.SessionCommit{ ID: "noise", SessionID: "session-commit-noise", CommitSHA: "noise", CommittedAt: base.Add(10 * time.Hour), }); err != nil { @@ -849,7 +900,7 @@ func TestDeleteSessionCommitRemovesOnlyTheTargetRow(t *testing.T) { base := time.Date(2026, 4, 20, 0, 0, 0, 0, time.UTC) for i, id := range []string{"keep-1", "drop-me", "keep-2"} { - if err := repo.CreateSessionCommit(ctx, &models.SessionCommit{ + if _, err := repo.CreateSessionCommit(ctx, &models.SessionCommit{ ID: id, SessionID: "session-commit-del", CommitSHA: id, CommittedAt: base.Add(time.Duration(i) * time.Hour), }); err != nil { @@ -916,7 +967,7 @@ func TestGitSnapshotMethodsPropagateBackendErrors(t *testing.T) { if _, err := repo.GetLatestGitSnapshotsBySessionIDs(ctx, []string{"session-git-broken"}); err == nil { t.Error("GetLatestGitSnapshotsBySessionIDs on a closed database returned nil") } - if err := repo.CreateSessionCommit(ctx, &models.SessionCommit{ + if _, err := repo.CreateSessionCommit(ctx, &models.SessionCommit{ ID: "commit-broken", SessionID: "session-git-broken", CommitSHA: "x", CommittedAt: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC), }); err == nil { diff --git a/apps/backend/internal/task/repository/sqlite/session_commits_migration_test.go b/apps/backend/internal/task/repository/sqlite/session_commits_migration_test.go new file mode 100644 index 0000000000..466134a08f --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/session_commits_migration_test.go @@ -0,0 +1,222 @@ +package sqlite + +import ( + "context" + "testing" + "time" +) + +// TestSessionCommitsDedupeAndActivationMigration mirrors +// TestTaskExternalIDMigrationAddsColumnsAndIndex: it reproduces the +// pre-migration state a legacy database can be in (duplicate +// (session_id, commit_sha) rows, no unique index, no activation marker), +// replays the migration twice, and asserts the end state is correct and +// idempotent. See migrateSessionCommitsDedupeAndActivation. +func TestSessionCommitsDedupeAndActivationMigration(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + seedSessionForGit(t, repo, "task-commits-migration", "session-commits-migration") + + // Simulate a legacy database: no unique index yet, and a duplicate + // (session_id, commit_sha) pair that a plain INSERT (pre-ON CONFLICT) + // could have produced, e.g. via a re-run archive capture. + if _, err := repo.db.Exec(`DROP INDEX IF EXISTS uniq_session_commits_session_sha`); err != nil { + t.Fatalf("drop unique index to simulate legacy schema: %v", err) + } + + earlier := time.Date(2026, 4, 1, 0, 0, 0, 0, time.UTC) + later := time.Date(2026, 4, 2, 0, 0, 0, 0, time.UTC) + insertRawCommit(t, repo, "commit-earliest", "session-commits-migration", "dup-sha", earlier) + insertRawCommit(t, repo, "commit-latest", "session-commits-migration", "dup-sha", later) + insertRawCommit(t, repo, "commit-unique", "session-commits-migration", "solo-sha", later) + + if got := countRows(t, repo, + `SELECT COUNT(1) FROM task_session_commits WHERE session_id = ?`, "session-commits-migration"); got != 3 { + t.Fatalf("pre-migration commit rows = %d, want 3", got) + } + + if err := repo.runMigrations(); err != nil { + t.Fatalf("replay migrations: %v", err) + } + + var indexName string + if err := repo.db.Get(&indexName, ` + SELECT name FROM sqlite_master WHERE type = 'index' AND name = 'uniq_session_commits_session_sha' + `); err != nil { + t.Fatalf("uniq_session_commits_session_sha index is missing: %v", err) + } + + if got := countRows(t, repo, + `SELECT COUNT(1) FROM task_session_commits WHERE session_id = ? AND commit_sha = ?`, + "session-commits-migration", "dup-sha"); got != 1 { + t.Fatalf("duplicate (session_id, commit_sha) rows after migration = %d, want 1", got) + } + + survivor, err := repo.GetLatestSessionCommit(ctx, "session-commits-migration") + if err != nil { + t.Fatalf("GetLatestSessionCommit: %v", err) + } + if survivor.ID != "commit-unique" { + // commit-unique has the later committed_at among the two distinct + // SHAs, so it should sort first; the dedupe survivor for dup-sha is + // checked directly below. + t.Errorf("GetLatestSessionCommit = %q, want commit-unique", survivor.ID) + } + + var survivorID string + if err := repo.db.Get(&survivorID, repo.db.Rebind(` + SELECT id FROM task_session_commits WHERE session_id = ? AND commit_sha = ? + `), "session-commits-migration", "dup-sha"); err != nil { + t.Fatalf("select dedupe survivor: %v", err) + } + if survivorID != "commit-earliest" { + t.Errorf("dedupe kept %q, want commit-earliest (earliest created_at)", survivorID) + } + + activatedAt := readCommitCaptureActivatedAt(t, repo) + if activatedAt == "" { + t.Fatal("commit_capture_activated_at was not published") + } + if _, err := time.Parse(time.RFC3339Nano, activatedAt); err != nil { + t.Errorf("commit_capture_activated_at = %q, not RFC3339Nano: %v", activatedAt, err) + } + + // Replay: must not error, must not duplicate the index, and must not + // move the activation marker (legacy rows must stay pinned to the + // original activation instant, not silently drift on every boot). + if err := repo.runMigrations(); err != nil { + t.Fatalf("replay migrations twice: %v", err) + } + if got := readCommitCaptureActivatedAt(t, repo); got != activatedAt { + t.Errorf("commit_capture_activated_at changed on replay: %q -> %q", activatedAt, got) + } + if got := countRows(t, repo, + `SELECT COUNT(1) FROM task_session_commits WHERE session_id = ?`, "session-commits-migration"); got != 2 { + t.Fatalf("commit rows after second replay = %d, want 2 (no re-duplication)", got) + } +} + +// Regression: initSchema() (the real boot path — persistence.Provide calls +// this, not runMigrations() directly) must not crash on an existing database +// that already has duplicate (session_id, commit_sha) rows. Before this +// test, initGitSchema() (base_schema.go) created +// uniq_session_commits_session_sha unconditionally as part of its +// unconditional, error-propagating schema-init Exec — which runs BEFORE +// runMigrations() in initSchema()'s step list, i.e. before +// migrateSessionCommitsDedupeAndActivation ever gets a chance to dedupe. +// CREATE UNIQUE INDEX over pre-existing duplicate data fails outright, and +// initGitSchema returns that error directly (unlike migrate.Apply, which +// swallows it), so initSchema() — and therefore process boot — would abort. +// TestSessionCommitsDedupeAndActivationMigration alone cannot catch this: it +// calls runMigrations() directly, skipping initGitSchema entirely. +func TestInitSchemaSurvivesLegacyDuplicateCommitRows(t *testing.T) { + repo := newRepoForSessionTests(t) + seedSessionForGit(t, repo, "task-boot-dup", "session-boot-dup") + + // seedSessionForGit's bare CreateTask leaves workspace_id unset. On a + // real legacy database every existing task already went through + // ensureDefaultWorkspace's backfill on a prior boot; simulate that here + // so this test isolates the task_session_commits ordering bug under + // test rather than tripping the unrelated (pre-existing, out of scope) + // ensureDefaultWorkspace behavior for a same-boot, workflow-less task. + var defaultWorkspaceID string + if err := repo.db.Get(&defaultWorkspaceID, `SELECT id FROM workspaces ORDER BY created_at LIMIT 1`); err != nil { + t.Fatalf("look up default workspace: %v", err) + } + if _, err := repo.db.Exec(repo.db.Rebind(`UPDATE tasks SET workspace_id = ? WHERE id = ?`), + defaultWorkspaceID, "task-boot-dup"); err != nil { + t.Fatalf("backfill task workspace_id: %v", err) + } + + // Simulate the pre-deploy production state this migration exists to + // clean up: unique index absent, duplicate (session_id, commit_sha) rows + // already present (e.g. from a pre-idempotency-fix repeated archive + // capture). + if _, err := repo.db.Exec(`DROP INDEX IF EXISTS uniq_session_commits_session_sha`); err != nil { + t.Fatalf("drop index to simulate legacy schema: %v", err) + } + now := time.Now().UTC() + insertRawCommit(t, repo, "dup-a", "session-boot-dup", "dup-sha", now) + insertRawCommit(t, repo, "dup-b", "session-boot-dup", "dup-sha", now.Add(time.Minute)) + + // Simulate the next process boot on this same (legacy, duplicate-laden) + // database. + if err := repo.initSchema(); err != nil { + t.Fatalf("initSchema() on a legacy DB with pre-existing duplicate commit rows must not error: %v", err) + } + + if got := countRows(t, repo, + `SELECT COUNT(1) FROM task_session_commits WHERE session_id = ? AND commit_sha = ?`, + "session-boot-dup", "dup-sha"); got != 1 { + t.Errorf("rows for dup-sha after initSchema() = %d, want 1 (deduped)", got) + } + + var indexName string + if err := repo.db.Get(&indexName, ` + SELECT name FROM sqlite_master WHERE type = 'index' AND name = 'uniq_session_commits_session_sha' + `); err != nil { + t.Fatalf("uniq_session_commits_session_sha index is missing after initSchema(): %v", err) + } +} + +// Regression: migrateSessionCommitsDedupeAndActivation must not swallow a +// failure in the dedupe DELETE or the CREATE UNIQUE INDEX step. Both used to +// run through r.migrate.Apply, which logs a WARN and returns void on any +// error that isn't "already exists" - so a failure there left the function +// returning nil, publishing the activation marker, and leaving +// CreateSessionCommit's `ON CONFLICT (session_id, commit_sha)` writer +// permanently broken (no matching unique constraint) with nothing visibly +// wrong at boot. Dropping the table before calling the migration function +// directly reproduces a failure in that first step without needing to +// fabricate a real duplicate-data edge case DELETE/CREATE INDEX IF NOT +// EXISTS can't otherwise hit. +func TestMigrateSessionCommitsDedupeAndActivationPropagatesDedupeOrIndexFailure(t *testing.T) { + repo := newRepoForSessionTests(t) + + // newRepoForSessionTests already ran this migration once successfully + // (via NewWithDB -> initSchema -> runMigrations), so the activation + // marker already exists with a real timestamp. Capture it, then prove a + // second, failing call does not touch it - the failure must be surfaced + // as an error, not silently absorbed and treated as another successful + // (re-)activation. + activatedAtBefore := readCommitCaptureActivatedAt(t, repo) + + if _, err := repo.db.Exec(`DROP TABLE task_session_commits`); err != nil { + t.Fatalf("drop table to simulate a failed dedupe/index step: %v", err) + } + + if err := repo.migrateSessionCommitsDedupeAndActivation(); err == nil { + t.Fatal("migrateSessionCommitsDedupeAndActivation returned nil after the dedupe/index step failed - " + + "the unique index the ON CONFLICT write path depends on may now be silently missing") + } + + if got := readCommitCaptureActivatedAt(t, repo); got != activatedAtBefore { + t.Errorf("commit_capture_activated_at changed to %q despite the dedupe/index step failing, want unchanged %q", + got, activatedAtBefore) + } +} + +// insertRawCommit writes a task_session_commits row bypassing +// CreateSessionCommit, so a test can construct pre-migration states +// (duplicates) the production write path would now refuse. +func insertRawCommit(t *testing.T, repo *Repository, id, sessionID, commitSHA string, committedAt time.Time) { + t.Helper() + if _, err := repo.db.Exec(repo.db.Rebind(` + INSERT INTO task_session_commits (id, session_id, commit_sha, committed_at, created_at) + VALUES (?, ?, ?, ?, ?) + `), id, sessionID, commitSHA, committedAt, committedAt); err != nil { + t.Fatalf("insertRawCommit(%s): %v", id, err) + } +} + +func readCommitCaptureActivatedAt(t *testing.T, repo *Repository) string { + t.Helper() + var value string + err := repo.db.Get(&value, repo.db.Rebind(` + SELECT value FROM kandev_meta WHERE key = ? + `), commitCaptureActivatedAtMetaKey) + if err != nil { + t.Fatalf("read commit_capture_activated_at: %v", err) + } + return value +} diff --git a/apps/backend/internal/task/service/service_turns_coverage_test.go b/apps/backend/internal/task/service/service_turns_coverage_test.go index 333555ee48..ff3dd3712a 100644 --- a/apps/backend/internal/task/service/service_turns_coverage_test.go +++ b/apps/backend/internal/task/service/service_turns_coverage_test.go @@ -273,7 +273,7 @@ func TestGitSnapshotAndCommitReads(t *testing.T) { t.Fatalf("latest snapshot = %q, want %q", gotLatest.ID, latest.ID) } - if err := repo.CreateSessionCommit(ctx, &models.SessionCommit{ + if _, err := repo.CreateSessionCommit(ctx, &models.SessionCommit{ ID: "commit-1", SessionID: "sess-msg", CommitSHA: "sha-1", CommitMessage: "first", CommittedAt: time.Now().UTC(), }); err != nil { diff --git a/apps/backend/pkg/pluginsdk/data_types.go b/apps/backend/pkg/pluginsdk/data_types.go index 1299d488c8..9d519aa10f 100644 --- a/apps/backend/pkg/pluginsdk/data_types.go +++ b/apps/backend/pkg/pluginsdk/data_types.go @@ -690,12 +690,25 @@ func sessionFilterFromProto(p *pluginv1.SessionFilter) SessionFilter { // SessionCodeStats is the Go-native mirror of // kandev.plugin.v1.SessionCodeStats — a computed per-session code-change // summary, never the raw commit/snapshot rows. +// +// CommittedLinesAvailable is false (with LinesAddedCommitted/ +// LinesDeletedCommitted reading 0) for a session that predates +// commit-capture activation and has no observed commits — capture wasn't +// running yet, so that zero cannot be told apart from a real zero-change +// session. This is a separate bool rather than making the existing fields +// pointers: they are already-shipped public SDK fields (ADR 0043's +// additive-only DTO contract), and changing their Go type from int64 to +// *int64 would be a source-compatibility break for any plugin rebuilding +// against a new SDK version. See internal/analytics/models.SessionCodeStats +// for the full contract (that internal type is not part of the public +// plugin surface and keeps its own *int64 representation). type SessionCodeStats struct { SessionID string LinesAddedCommitted int64 LinesDeletedCommitted int64 LinesAddedPeakPending int64 LinesDeletedPeakPending int64 + CommittedLinesAvailable bool } func (s SessionCodeStats) toProto() *pluginv1.SessionCodeStats { @@ -705,6 +718,7 @@ func (s SessionCodeStats) toProto() *pluginv1.SessionCodeStats { LinesDeletedCommitted: s.LinesDeletedCommitted, LinesAddedPeakPending: s.LinesAddedPeakPending, LinesDeletedPeakPending: s.LinesDeletedPeakPending, + CommittedLinesAvailable: s.CommittedLinesAvailable, } } @@ -718,6 +732,7 @@ func sessionCodeStatsFromProto(p *pluginv1.SessionCodeStats) SessionCodeStats { LinesDeletedCommitted: p.GetLinesDeletedCommitted(), LinesAddedPeakPending: p.GetLinesAddedPeakPending(), LinesDeletedPeakPending: p.GetLinesDeletedPeakPending(), + CommittedLinesAvailable: p.GetCommittedLinesAvailable(), } } diff --git a/apps/backend/pkg/pluginsdk/data_types_test.go b/apps/backend/pkg/pluginsdk/data_types_test.go index 49de952ba6..f0b69569fb 100644 --- a/apps/backend/pkg/pluginsdk/data_types_test.go +++ b/apps/backend/pkg/pluginsdk/data_types_test.go @@ -252,11 +252,35 @@ func TestSessionCodeStatsProtoRoundTrip(t *testing.T) { LinesDeletedCommitted: 40, LinesAddedPeakPending: 15, LinesDeletedPeakPending: 3, + CommittedLinesAvailable: true, } proto := stats.toProto() require.Equal(t, stats, sessionCodeStatsFromProto(proto)) } +// A session that predates commit-capture activation reports +// CommittedLinesAvailable == false, not a real measurement, for committed +// lines — the wire contract must round-trip that unavailability exactly, +// not silently present it as a real zero-change session. LinesAddedCommitted/ +// LinesDeletedCommitted stay plain int64 (not pointers) because they are +// already-shipped public SDK fields (ADR 0043's additive-only DTO contract); +// see SessionCodeStats' doc comment. +func TestSessionCodeStatsProtoRoundTrip_CommittedLinesUnavailable(t *testing.T) { + stats := SessionCodeStats{ + SessionID: "session-legacy", + LinesAddedCommitted: 0, + LinesDeletedCommitted: 0, + LinesAddedPeakPending: 7, + LinesDeletedPeakPending: 1, + CommittedLinesAvailable: false, + } + proto := stats.toProto() + if proto.GetCommittedLinesAvailable() { + t.Errorf("proto.CommittedLinesAvailable = true, want false") + } + require.Equal(t, stats, sessionCodeStatsFromProto(proto)) +} + func TestTasksSliceProtoRoundTrip_EmptyIsNil(t *testing.T) { tasks, err := tasksFromProto(nil) require.NoError(t, err) diff --git a/apps/backend/pkg/pluginsdk/host_data_test.go b/apps/backend/pkg/pluginsdk/host_data_test.go index 0986d3ddf4..9ee6289b71 100644 --- a/apps/backend/pkg/pluginsdk/host_data_test.go +++ b/apps/backend/pkg/pluginsdk/host_data_test.go @@ -250,7 +250,7 @@ func TestHostData_TaskWritesAndMessage(t *testing.T) { func TestHostData_SessionsListAndCodeStats(t *testing.T) { impl := &dataRecordingHost{ sessions: []Session{{ID: "session-1", TaskID: "task-1", State: "running"}}, - codeStats: []SessionCodeStats{{SessionID: "session-1", LinesAddedCommitted: 10}}, + codeStats: []SessionCodeStats{{SessionID: "session-1", LinesAddedCommitted: 10, CommittedLinesAvailable: true}}, } host := dialHostOverBufconn(t, impl) diff --git a/apps/backend/proto/kandev/plugin/v1/plugin.pb.go b/apps/backend/proto/kandev/plugin/v1/plugin.pb.go index 004bfa90f0..33b8d572ca 100644 --- a/apps/backend/proto/kandev/plugin/v1/plugin.pb.go +++ b/apps/backend/proto/kandev/plugin/v1/plugin.pb.go @@ -4316,6 +4316,17 @@ type SessionCodeStats struct { LinesDeletedCommitted int64 `protobuf:"varint,3,opt,name=lines_deleted_committed,json=linesDeletedCommitted,proto3" json:"lines_deleted_committed,omitempty"` // SUM of commit deletions LinesAddedPeakPending int64 `protobuf:"varint,4,opt,name=lines_added_peak_pending,json=linesAddedPeakPending,proto3" json:"lines_added_peak_pending,omitempty"` // peak uncommitted diff additions across snapshots LinesDeletedPeakPending int64 `protobuf:"varint,5,opt,name=lines_deleted_peak_pending,json=linesDeletedPeakPending,proto3" json:"lines_deleted_peak_pending,omitempty"` + // False (with lines_added_committed/lines_deleted_committed reading 0) for + // a session that predates commit-capture activation and has no observed + // commits — capture wasn't running yet, so a plugin must not treat that + // zero as "zero lines changed". Kept as a separate additive field, rather + // than making the existing fields 2/3 `optional`, because fields 2/3 are + // an already-shipped part of this DTO's public contract (see ADR 0043: + // "DTO fields are additive-only thereafter") — plugin authors read them as + // plain int64 in the generated Go SDK, and switching that Go field type to + // *int64 is a source-level break for anyone rebuilding against a new SDK, + // even though the wire-level `optional` change itself is protobuf-safe. + CommittedLinesAvailable bool `protobuf:"varint,6,opt,name=committed_lines_available,json=committedLinesAvailable,proto3" json:"committed_lines_available,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -4385,6 +4396,13 @@ func (x *SessionCodeStats) GetLinesDeletedPeakPending() int64 { return 0 } +func (x *SessionCodeStats) GetCommittedLinesAvailable() bool { + if x != nil { + return x.CommittedLinesAvailable + } + return false +} + type ListSessionCodeStatsRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Filter *SessionFilter `protobuf:"bytes,1,opt,name=filter,proto3" json:"filter,omitempty"` @@ -6117,14 +6135,15 @@ const file_kandev_plugin_v1_plugin_proto_rawDesc = "" + "\x04page\x18\x02 \x01(\v2\x16.kandev.plugin.v1.PageR\x04page\"\x86\x01\n" + "\x14ListSessionsResponse\x125\n" + "\bsessions\x18\x01 \x03(\v2\x19.kandev.plugin.v1.SessionR\bsessions\x127\n" + - "\tpage_info\x18\x02 \x01(\v2\x1a.kandev.plugin.v1.PageInfoR\bpageInfo\"\x93\x02\n" + + "\tpage_info\x18\x02 \x01(\v2\x1a.kandev.plugin.v1.PageInfoR\bpageInfo\"\xcf\x02\n" + "\x10SessionCodeStats\x12\x1d\n" + "\n" + "session_id\x18\x01 \x01(\tR\tsessionId\x122\n" + "\x15lines_added_committed\x18\x02 \x01(\x03R\x13linesAddedCommitted\x126\n" + "\x17lines_deleted_committed\x18\x03 \x01(\x03R\x15linesDeletedCommitted\x127\n" + "\x18lines_added_peak_pending\x18\x04 \x01(\x03R\x15linesAddedPeakPending\x12;\n" + - "\x1alines_deleted_peak_pending\x18\x05 \x01(\x03R\x17linesDeletedPeakPending\"\x82\x01\n" + + "\x1alines_deleted_peak_pending\x18\x05 \x01(\x03R\x17linesDeletedPeakPending\x12:\n" + + "\x19committed_lines_available\x18\x06 \x01(\bR\x17committedLinesAvailable\"\x82\x01\n" + "\x1bListSessionCodeStatsRequest\x127\n" + "\x06filter\x18\x01 \x01(\v2\x1f.kandev.plugin.v1.SessionFilterR\x06filter\x12*\n" + "\x04page\x18\x02 \x01(\v2\x16.kandev.plugin.v1.PageR\x04page\"\x91\x01\n" + diff --git a/apps/backend/proto/kandev/plugin/v1/plugin.proto b/apps/backend/proto/kandev/plugin/v1/plugin.proto index 6164e26869..76dc655e75 100644 --- a/apps/backend/proto/kandev/plugin/v1/plugin.proto +++ b/apps/backend/proto/kandev/plugin/v1/plugin.proto @@ -530,6 +530,17 @@ message SessionCodeStats { int64 lines_deleted_committed = 3; // SUM of commit deletions int64 lines_added_peak_pending = 4; // peak uncommitted diff additions across snapshots int64 lines_deleted_peak_pending = 5; + // False (with lines_added_committed/lines_deleted_committed reading 0) for + // a session that predates commit-capture activation and has no observed + // commits — capture wasn't running yet, so a plugin must not treat that + // zero as "zero lines changed". Kept as a separate additive field, rather + // than making the existing fields 2/3 `optional`, because fields 2/3 are + // an already-shipped part of this DTO's public contract (see ADR 0043: + // "DTO fields are additive-only thereafter") — plugin authors read them as + // plain int64 in the generated Go SDK, and switching that Go field type to + // *int64 is a source-level break for anyone rebuilding against a new SDK, + // even though the wire-level `optional` change itself is protobuf-safe. + bool committed_lines_available = 6; } message ListSessionCodeStatsRequest { SessionFilter filter = 1;