diff --git a/internal/database/compact_timestamp_benchmark_test.go b/internal/database/compact_timestamp_benchmark_test.go new file mode 100644 index 00000000..5998c9da --- /dev/null +++ b/internal/database/compact_timestamp_benchmark_test.go @@ -0,0 +1,318 @@ +package database_test + +import ( + "context" + "fmt" + "path/filepath" + "reflect" + "runtime" + "testing" + "time" + + "github.com/omarluq/librecode/internal/database" +) + +const ( + retainedHeapElements = 1_000_000 + amd64Architecture = "amd64" +) + +type compactTimestampLayoutPair struct { + Name string `json:"name"` + CurrentBytes uintptr `json:"current_bytes"` + CompactBytes uintptr `json:"compact_bytes"` + SavedBytes uintptr `json:"saved_bytes"` +} + +// BenchmarkSessionRepositoryLargeTranscript scans production repository pages, +// including message-part hydration, against the deterministic indexed fixture. +// Fixture construction is outside the timer. Run with -count=10 for the Phase 0 +// sample set. +func BenchmarkSessionRepositoryLargeTranscript(b *testing.B) { + fixtureRows := compactTimestampRowCount(b) + path := filepath.Join(b.TempDir(), "compact-timestamp-transcript.db") + connection := generateCompactTimestampFixture(b, path, fixtureRows) + b.Cleanup(func() { + if err := connection.Close(); err != nil { + b.Errorf("close transcript benchmark fixture: %v", err) + } + }) + + repositories, err := database.NewRepositories(connection) + if err != nil { + b.Fatalf("construct benchmark repositories: %v", err) + } + + for _, pageSize := range []int{256, 4096} { + b.Run(fmt.Sprintf("tail_%d", pageSize), func(b *testing.B) { + if fixtureRows < pageSize { + b.Skipf("fixture has %d rows; tail page requires %d", fixtureRows, pageSize) + } + + benchmarkTranscriptPage(b, func(ctx context.Context) ([]database.SessionMessageEntity, error) { + return repositories.Sessions.TranscriptMessageTail(ctx, compactTimestampSessionID, pageSize) + }, pageSize) + }) + + cursorIndex := fixtureRows / 2 + cursor := time.Date(2025, 1, 1, 0, 0, 0, 1, time.UTC). + Add(time.Duration(cursorIndex) * 100 * time.Millisecond) + cursorID := fmt.Sprintf("01910000-0000-7000-8000-%012x", cursorIndex+1) + + b.Run(fmt.Sprintf("cursor_%d", pageSize), func(b *testing.B) { + // RFC3339Nano omits the fractional part for whole seconds, and + // SQLite orders TEXT lexically, so the whole-second row at the + // cursor's second sorts after a fractional cursor string. Count + // older rows with the same lexical semantics production uses. + availableRows := cursorIndex + if cursorIndex%10 != 0 { + availableRows-- + } + + if availableRows < pageSize { + b.Skipf("cursor has %d older rows; page requires %d", availableRows, pageSize) + } + + benchmarkTranscriptPage(b, func(ctx context.Context) ([]database.SessionMessageEntity, error) { + return repositories.Sessions.TranscriptMessagesBefore( + ctx, compactTimestampSessionID, cursor, cursorID, pageSize, + ) + }, pageSize) + }) + } +} + +func benchmarkTranscriptPage( + b *testing.B, + load func(context.Context) ([]database.SessionMessageEntity, error), + pageSize int, +) { + b.Helper() + b.ReportAllocs() + b.ResetTimer() + + var messageSink []database.SessionMessageEntity + + for b.Loop() { + messages, err := load(context.Background()) + if err != nil { + b.Fatalf("scan transcript page: %v", err) + } + + if len(messages) != pageSize { + b.Fatalf("transcript page length = %d, want %d", len(messages), pageSize) + } + + messageSink = messages + } + + runtime.KeepAlive(messageSink) + b.ReportMetric(float64(pageSize), "messages/op") +} + +// BenchmarkCompactTimestampLayouts reports raw amd64 struct bytes. The compact +// layouts are test-only field-for-field proposals; they do not alter production. +func BenchmarkCompactTimestampLayouts(b *testing.B) { + if runtime.GOARCH != amd64Architecture { + b.Skip("raw layout baseline is specified for amd64") + } + + var layoutSink compactTimestampLayoutPair + + for _, layout := range compactTimestampLayouts() { + b.Run(layout.Name, func(b *testing.B) { + for b.Loop() { + layoutSink = layout + } + + runtime.KeepAlive(layoutSink) + b.ReportMetric(float64(layout.CurrentBytes), "current-B/entity") + b.ReportMetric(float64(layout.CompactBytes), "compact-B/entity") + b.ReportMetric(float64(layout.SavedBytes), "saved-B/entity") + }) + } +} + +func compactTimestampLayouts() []compactTimestampLayoutPair { + if runtime.GOARCH != amd64Architecture { + return nil + } + + return []compactTimestampLayoutPair{ + newLayoutPair("MessageEntity", reflect.TypeFor[database.MessageEntity](), compactMessageType()), + newLayoutPair( + "SessionMessageEntity", + reflect.TypeFor[database.SessionMessageEntity](), + compactSessionMessageType(), + ), + newLayoutPair("SessionEntity", reflect.TypeFor[database.SessionEntity](), compactSessionType()), + newLayoutPair("EntryEntity", reflect.TypeFor[database.EntryEntity](), compactEntryType()), + newLayoutPair("TaskEntity", reflect.TypeFor[database.TaskEntity](), compactTaskType()), + } +} + +func newLayoutPair(name string, currentType, compactType reflect.Type) compactTimestampLayoutPair { + currentBytes := currentType.Size() + compactBytes := compactType.Size() + + return compactTimestampLayoutPair{ + Name: name, + CurrentBytes: currentBytes, + CompactBytes: compactBytes, + SavedBytes: currentBytes - compactBytes, + } +} + +func compactMessageType() reflect.Type { + return reflect.StructOf([]reflect.StructField{ + layoutField[uint32]("Timestamp"), + layoutField[database.Role]("Role"), + layoutField[string]("Content"), + layoutField[string]("Provider"), + layoutField[string]("Model"), + layoutField[[]database.MessagePartEntity]("Parts"), + }) +} + +func compactSessionMessageType() reflect.Type { + return reflect.StructOf([]reflect.StructField{ + layoutField[uint32]("CreatedAt"), + layoutField[string]("SessionID"), + layoutField[string]("EntryID"), + layoutField[string]("Sender"), + layoutField[database.Role]("Role"), + layoutField[string]("Content"), + layoutField[string]("Provider"), + layoutField[string]("Model"), + layoutField[[]database.MessagePartEntity]("Parts"), + }) +} + +func compactSessionType() reflect.Type { + return reflect.StructOf([]reflect.StructField{ + layoutField[uint32]("CreatedAt"), + layoutField[uint32]("UpdatedAt"), + layoutField[string]("ID"), + layoutField[string]("CWD"), + layoutField[string]("Name"), + layoutField[string]("ParentSession"), + }) +} + +func compactEntryType() reflect.Type { + return reflect.StructOf([]reflect.StructField{ + layoutField[uint32]("CreatedAt"), + layoutField[*string]("ParentID"), + layoutField[string]("ToolStatus"), + layoutField[string]("SessionID"), + layoutField[string]("ToolArgsJSON"), + layoutField[string]("CustomType"), + layoutField[string]("DataJSON"), + layoutField[string]("ID"), + layoutField[string]("Summary"), + layoutField[string]("ToolName"), + layoutField[database.EntryType]("Type"), + layoutField[string]("BranchFromEntryID"), + layoutField[string]("CompactionFirstKeptEntryID"), + layoutFieldOf("Message", compactMessageType()), + layoutField[int]("CompactionTokensBefore"), + layoutField[int]("TokenEstimate"), + layoutField[bool]("Display"), + layoutField[bool]("ModelFacing"), + }) +} + +func compactTaskType() reflect.Type { + optionalTimestampType := reflect.StructOf([]reflect.StructField{ + layoutField[uint32]("Value"), + layoutField[bool]("Valid"), + }) + + return reflect.StructOf([]reflect.StructField{ + layoutField[uint32]("CreatedAt"), + layoutFieldOf("StartedAt", optionalTimestampType), + layoutFieldOf("FinishedAt", optionalTimestampType), + layoutField[uint32]("UpdatedAt"), + layoutFieldOf("LeaseExpiresAt", optionalTimestampType), + layoutField[string]("ID"), + layoutField[string]("Kind"), + layoutField[string]("ParentTaskID"), + layoutField[string]("OwnerSessionID"), + layoutField[string]("ConcurrencyKey"), + layoutField[string]("LeaseOwner"), + layoutField[database.TaskState]("State"), + layoutField[string]("Result"), + layoutField[string]("ErrorCode"), + layoutField[string]("ErrorMessage"), + }) +} + +func layoutField[T any](name string) reflect.StructField { + return layoutFieldOf(name, reflect.TypeFor[T]()) +} + +func layoutFieldOf(name string, fieldType reflect.Type) reflect.StructField { + return reflect.StructField{Name: name, Type: fieldType} +} + +// BenchmarkCompactTimestampRetainedHeap follows Hatchet's million-element +// method: force GC before and after allocation, read retained heap, and keep the +// backing array live with runtime.KeepAlive. Run this benchmark in isolation. +func BenchmarkCompactTimestampRetainedHeap(b *testing.B) { + if runtime.GOARCH != amd64Architecture { + b.Skip("retained heap baseline accompanies the amd64 raw layout baseline") + } + + b.Run("EntryEntity_current", func(b *testing.B) { + entryType := reflect.TypeFor[database.EntryEntity]() + measureRetainedHeap(b, retainedHeapElements, entryType.Size(), func() any { + values := make([]database.EntryEntity, retainedHeapElements) + stamp := time.Unix(1_735_689_600, 123_000_000).UTC() + + for index := range values { + values[index].CreatedAt = stamp + values[index].Message.Timestamp = stamp + } + + return values + }) + }) + + b.Run("EntryEntity_compact", func(b *testing.B) { + entryType := compactEntryType() + measureRetainedHeap(b, retainedHeapElements, entryType.Size(), func() any { + return reflect.MakeSlice(reflect.SliceOf(entryType), retainedHeapElements, retainedHeapElements).Interface() + }) + }) +} + +func measureRetainedHeap(b *testing.B, count int, rawBytes uintptr, allocate func() any) { + b.Helper() + b.StopTimer() + runtime.GC() + + var before, after runtime.MemStats + runtime.ReadMemStats(&before) + + values := allocate() + + runtime.GC() + runtime.ReadMemStats(&after) + + retained := uint64(0) + if after.HeapAlloc > before.HeapAlloc { + retained = after.HeapAlloc - before.HeapAlloc + } + + b.StartTimer() + + for b.Loop() { + runtime.KeepAlive(values) + } + + b.StopTimer() + runtime.KeepAlive(values) + b.ReportMetric(float64(rawBytes), "raw-B/entity") + b.ReportMetric(float64(retained)/float64(count), "retained-B/entity") + b.ReportMetric(float64(retained)/(1024*1024), "retained-MiB") +} diff --git a/internal/database/compact_timestamp_fixture_test.go b/internal/database/compact_timestamp_fixture_test.go new file mode 100644 index 00000000..e82cbf18 --- /dev/null +++ b/internal/database/compact_timestamp_fixture_test.go @@ -0,0 +1,976 @@ +package database_test + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "math" + "os" + "path/filepath" + "runtime" + "slices" + "strconv" + "strings" + "syscall" + "testing" + "time" + + "github.com/omarluq/librecode/internal/database" + _ "modernc.org/sqlite" // Register the production SQLite driver for the baseline harness. +) + +const ( + compactTimestampArtifactPathEnv = "LIBRECODE_COMPACT_TIMESTAMP_BASELINE_PATH" + compactTimestampRowsEnv = "LIBRECODE_DATABASE_BENCH_ROWS" + compactTimestampFixtureRows = 1_000_000 + compactTimestampFixtureBatch = 10_000 + compactTimestampSessionID = "01900000-0000-7000-8000-000000000001" + fixtureStartTimestamp = "2025-01-01T00:00:00Z" +) + +type baselineQueryPlanSpec struct { + Name string + SQL string + Args []any +} + +type baselinePageUsage struct { + Name string `json:"name"` + Kind string `json:"kind"` + Bytes int64 `json:"bytes"` + Pages int64 `json:"pages"` + PayloadBytes int64 `json:"payload_bytes"` + UnusedBytes int64 `json:"unused_bytes"` +} + +type baselineStorage struct { + Objects []baselinePageUsage `json:"objects"` + FileBytes int64 `json:"file_bytes"` + PageSize int64 `json:"page_size"` + PageCount int64 `json:"page_count"` + FreeListPages int64 `json:"free_list_pages"` + ActiveDBStatPages int64 `json:"active_dbstat_pages"` + HeapAllocBytes uint64 `json:"process_heap_alloc_bytes"` + RSSBytes uint64 `json:"process_rss_bytes"` +} + +type baselineCollisionStats struct { + EqualTimestampGroups int64 `json:"equal_timestamp_groups"` + RowsInEqualGroups int64 `json:"rows_in_equal_groups"` + MaxGroupLength int64 `json:"max_group_length"` +} + +type baselineIndex struct { + Name string `json:"name"` + TableName string `json:"table_name"` + SQL string `json:"sql"` +} + +type compactTimestampBaselineReport struct { + QueryPlans map[string][]string `json:"query_plans"` + Indexes []baselineIndex `json:"indexes"` + TimestampIndexes []string `json:"timestamp_indexes"` + Layouts []compactTimestampLayoutPair `json:"amd64_layouts,omitempty"` + GeneratedAtUTC string `json:"generated_at_utc"` + GOARCH string `json:"goarch"` + GoVersion string `json:"go_version"` + SessionID string `json:"session_id"` + TimestampStorage string `json:"timestamp_storage"` + BeforeVacuum baselineStorage `json:"before_vacuum"` + AfterVacuum baselineStorage `json:"after_vacuum"` + Collisions baselineCollisionStats `json:"same_second_collisions"` + FixtureRows int `json:"fixture_rows"` +} + +type baselineMeasurements struct { + Plans map[string][]string + Indexes []baselineIndex + TimestampIndexes []string + Storage baselineStorage + Collisions baselineCollisionStats +} + +// TestCompactTimestampBaselineContracts cheaply freezes the schema and query-plan +// inputs used by the expensive artifact generator. It never creates the million-row fixture. +func TestCompactTimestampBaselineContracts(t *testing.T) { + t.Parallel() + + connection := openCompactTimestampDatabase(t, ":memory:") + t.Cleanup(func() { + if err := connection.Close(); err != nil { + t.Errorf("close baseline schema: %v", err) + } + }) + + gotIndexes, err := timestampIndexInventory(t.Context(), connection) + if err != nil { + t.Fatalf("inventory timestamp indexes: %v", err) + } + + wantIndexes := expectedTimestampIndexes() + if !slices.Equal(gotIndexes, wantIndexes) { + t.Fatalf("timestamp index inventory changed\n got: %v\nwant: %v", gotIndexes, wantIndexes) + } + + plans, err := collectBaselineQueryPlans(t.Context(), connection) + if err != nil { + t.Fatalf("collect query plans: %v", err) + } + + for _, name := range expectedBaselinePlanNames() { + if len(plans[name]) == 0 { + t.Errorf("query plan %q was not collected", name) + } + } + + usage, err := collectDBStat(t.Context(), connection) + if err != nil { + t.Fatalf("collect dbstat: %v", err) + } + + if len(usage) == 0 { + t.Fatal("dbstat returned no schema objects") + } + + assertProductionTranscriptIndexes(t, connection) +} + +// TestGenerateCompactTimestampBaseline is an explicit, provider-free artifact +// generator. Set LIBRECODE_COMPACT_TIMESTAMP_BASELINE_PATH to the desired .db +// path. It creates the database and a sibling .json report. The row count is one +// million by default; LIBRECODE_DATABASE_BENCH_ROWS exists only for smoke runs. +func TestGenerateCompactTimestampBaseline(t *testing.T) { + t.Parallel() + + path := cleanArtifactPath(t) + fixtureRows := compactTimestampRowCount(t) + connection := generateCompactTimestampFixture(t, path, fixtureRows) + + defer func() { + if err := connection.Close(); err != nil { + t.Errorf("close generated baseline: %v", err) + } + }() + + before := collectBaselineMeasurements(t, connection, path) + + _, vacuumErr := connection.ExecContext(t.Context(), "VACUUM") + if vacuumErr != nil { + t.Fatalf("vacuum baseline fixture: %v", vacuumErr) + } + + after, err := measureBaselineStorage(t.Context(), connection, path) + if err != nil { + t.Fatalf("measure storage after VACUUM: %v", err) + } + + report := compactTimestampBaselineReport{ + QueryPlans: before.Plans, + GeneratedAtUTC: time.Now().UTC().Format(time.RFC3339), + GOARCH: runtime.GOARCH, + GoVersion: runtime.Version(), + SessionID: compactTimestampSessionID, + TimestampStorage: "RFC3339Nano TEXT", + BeforeVacuum: before.Storage, + AfterVacuum: after, + Indexes: before.Indexes, + TimestampIndexes: before.TimestampIndexes, + Layouts: compactTimestampLayouts(), + Collisions: before.Collisions, + FixtureRows: fixtureRows, + } + + writeBaselineReport(t, path, &report) + logBaselineReport(t, path, &report) +} + +func cleanArtifactPath(tb testing.TB) string { + tb.Helper() + + path := strings.TrimSpace(os.Getenv(compactTimestampArtifactPathEnv)) + if path == "" { + tb.Skip("set " + compactTimestampArtifactPathEnv + " to generate the expensive baseline fixture") + } + + absolutePath, err := filepath.Abs(path) + if err != nil { + tb.Fatalf("resolve baseline artifact path: %v", err) + } + + return absolutePath +} + +func collectBaselineMeasurements(tb testing.TB, connection *sql.DB, path string) baselineMeasurements { + tb.Helper() + + storage, err := measureBaselineStorage(tb.Context(), connection, path) + if err != nil { + tb.Fatalf("measure storage before VACUUM: %v", err) + } + + collisions, err := measureSameSecondCollisions(tb.Context(), connection) + if err != nil { + tb.Fatalf("measure timestamp collisions: %v", err) + } + + plans, err := collectBaselineQueryPlans(tb.Context(), connection) + if err != nil { + tb.Fatalf("collect query plans: %v", err) + } + + indexes, err := indexInventory(tb.Context(), connection) + if err != nil { + tb.Fatalf("inventory indexes: %v", err) + } + + timestampIndexes, err := timestampIndexInventory(tb.Context(), connection) + if err != nil { + tb.Fatalf("inventory timestamp indexes: %v", err) + } + + return baselineMeasurements{ + Plans: plans, + Indexes: indexes, + TimestampIndexes: timestampIndexes, + Storage: storage, + Collisions: collisions, + } +} + +func writeBaselineReport(tb testing.TB, path string, report *compactTimestampBaselineReport) { + tb.Helper() + + reportBytes, err := json.MarshalIndent(report, "", " ") + if err != nil { + tb.Fatalf("encode baseline report: %v", err) + } + + reportPath := filepath.Clean(path + ".json") + if err := os.WriteFile(reportPath, append(reportBytes, '\n'), 0o600); err != nil { + tb.Fatalf("write baseline report: %v", err) + } +} + +func logBaselineReport(tb testing.TB, path string, report *compactTimestampBaselineReport) { + tb.Helper() + + tb.Logf("generated %d rows: database=%s report=%s", report.FixtureRows, path, path+".json") + tb.Logf( + "file bytes before=%d after=%d; active pages before=%d after=%d", + report.BeforeVacuum.FileBytes, + report.AfterVacuum.FileBytes, + report.BeforeVacuum.ActiveDBStatPages, + report.AfterVacuum.ActiveDBStatPages, + ) + tb.Logf( + "same-second groups=%d rows-in-groups=%d max-group=%d", + report.Collisions.EqualTimestampGroups, + report.Collisions.RowsInEqualGroups, + report.Collisions.MaxGroupLength, + ) +} + +func compactTimestampRowCount(tb testing.TB) int { + tb.Helper() + + value := strings.TrimSpace(os.Getenv(compactTimestampRowsEnv)) + if value == "" { + return compactTimestampFixtureRows + } + + fixtureRows, err := strconv.Atoi(value) + if err != nil || fixtureRows <= 0 { + tb.Fatalf("%s must be a positive integer, got %q", compactTimestampRowsEnv, value) + } + + return fixtureRows +} + +func openCompactTimestampDatabase(tb testing.TB, path string) *sql.DB { + tb.Helper() + + connection, err := sql.Open("sqlite", path) + if err != nil { + tb.Fatalf("open baseline SQLite database: %v", err) + } + + connection.SetMaxOpenConns(1) + + if err := database.Migrate(context.Background(), connection); err != nil { + closeDatabaseAfterFailure(tb, connection) + tb.Fatalf("migrate baseline SQLite database: %v", err) + } + + return connection +} + +func generateCompactTimestampFixture(tb testing.TB, path string, rowCount int) *sql.DB { + tb.Helper() + + cleanPath, err := filepath.Abs(path) + if err != nil { + tb.Fatalf("resolve fixture path: %v", err) + } + + prepareCompactTimestampFixturePath(tb, cleanPath) + + connection := openCompactTimestampDatabase(tb, cleanPath) + applyCompactTimestampFixtureSettings(tb, connection) + insertCompactTimestampFixtureSession(tb, connection) + + for start := 0; start < rowCount; start += compactTimestampFixtureBatch { + end := min(start+compactTimestampFixtureBatch, rowCount) + insertCompactTimestampFixtureBatch(tb, connection, start, end) + } + + _, analyzeErr := connection.ExecContext(context.Background(), "ANALYZE") + if analyzeErr != nil { + closeDatabaseAfterFailure(tb, connection) + tb.Fatalf("analyze fixture: %v", analyzeErr) + } + + return connection +} + +func prepareCompactTimestampFixturePath(tb testing.TB, path string) { + tb.Helper() + + if err := os.MkdirAll(filepath.Dir(path), 0o750); err != nil { + tb.Fatalf("create fixture directory: %v", err) + } + + for _, candidate := range []string{path, path + "-wal", path + "-shm"} { + if err := os.Remove(candidate); err != nil && !os.IsNotExist(err) { + tb.Fatalf("remove old fixture %s: %v", candidate, err) + } + } +} + +func applyCompactTimestampFixtureSettings(tb testing.TB, connection *sql.DB) { + tb.Helper() + + pragmas := []string{ + "PRAGMA journal_mode=DELETE", + "PRAGMA synchronous=OFF", + "PRAGMA foreign_keys=OFF", + "PRAGMA temp_store=MEMORY", + "PRAGMA cache_size=-65536", + } + + for _, pragma := range pragmas { + if _, err := connection.ExecContext(context.Background(), pragma); err != nil { + closeDatabaseAfterFailure(tb, connection) + tb.Fatalf("apply fixture setting %q: %v", pragma, err) + } + } +} + +func insertCompactTimestampFixtureSession(tb testing.TB, connection *sql.DB) { + tb.Helper() + + const insertSession = `INSERT INTO sessions +(id,cwd,name,parent_session_id,created_at,updated_at) VALUES(?,?,?,NULL,?,?)` + + _, err := connection.ExecContext( + context.Background(), + insertSession, + compactTimestampSessionID, + "/fixture", + "compact timestamp baseline", + fixtureStartTimestamp, + "2025-01-02T00:00:00Z", + ) + if err != nil { + closeDatabaseAfterFailure(tb, connection) + tb.Fatalf("insert fixture session: %v", err) + } +} + +func closeDatabaseAfterFailure(tb testing.TB, connection *sql.DB) { + tb.Helper() + + if err := connection.Close(); err != nil { + tb.Errorf("close database after fixture failure: %v", err) + } +} + +func insertCompactTimestampFixtureBatch(tb testing.TB, connection *sql.DB, start, end int) { + tb.Helper() + + transaction, err := connection.BeginTx(context.Background(), nil) + if err != nil { + tb.Fatalf("begin fixture batch %d:%d: %v", start, end, err) + } + + committed := false + defer func() { + if committed { + return + } + + rollbackErr := transaction.Rollback() + if rollbackErr != nil && !errors.Is(rollbackErr, sql.ErrTxDone) { + tb.Errorf("roll back fixture batch %d:%d: %v", start, end, rollbackErr) + } + }() + + insertCompactTimestampEntries(tb, transaction, start, end) + insertCompactTimestampMessages(tb, transaction, start, end) + + if err := transaction.Commit(); err != nil { + tb.Fatalf("commit fixture batch %d:%d: %v", start, end, err) + } + + committed = true +} + +func insertCompactTimestampEntries(tb testing.TB, transaction *sql.Tx, start, end int) { + tb.Helper() + + const insertEntries = `WITH digits(n) AS (VALUES(0),(1),(2),(3),(4),(5),(6),(7),(8),(9)), +seq(n) AS ( + SELECT ? + a.n + 10*b.n + 100*c.n + 1000*d.n + FROM digits a CROSS JOIN digits b CROSS JOIN digits c CROSS JOIN digits d +) +INSERT INTO session_entries ( + id,session_id,parent_id,entry_type,custom_type,data_json,summary,created_at, + tool_name,tool_status,tool_args_json,token_estimate,model_facing,display, + compaction_first_kept_entry_id,compaction_tokens_before,branch_from_entry_id,operation_id) +SELECT printf('01910000-0000-7000-8000-%012x',n+1), ?, NULL, 'message', '', '{}', '', + strftime('%Y-%m-%dT%H:%M:%S','2025-01-01T00:00:00Z',printf('+%.1f seconds',n/10.0)) || + CASE WHEN n%10 = 0 THEN 'Z' ELSE '.'||printf('%d',n%10)||'Z' END, + '', '', '', 32, 1, 1, '', 0, '', '' FROM seq WHERE n < ?` + + _, err := transaction.ExecContext( + context.Background(), + insertEntries, + start, + compactTimestampSessionID, + end, + ) + if err != nil { + tb.Fatalf("insert fixture entries %d:%d: %v", start, end, err) + } +} + +func insertCompactTimestampMessages(tb testing.TB, transaction *sql.Tx, start, end int) { + tb.Helper() + + firstID := fmt.Sprintf("01910000-0000-7000-8000-%012x", start+1) + lastID := fmt.Sprintf("01910000-0000-7000-8000-%012x", end) + + const insertMessages = `INSERT INTO session_messages(entry_id,role,provider,model) +SELECT id,CASE WHEN substr(id,-1,1) IN ('0','2','4','6','8','a','c','e') THEN 'user' ELSE 'assistant' END, +CASE WHEN substr(id,-1,1) IN ('0','2','4','6','8','a','c','e') THEN '' ELSE 'fixture' END, +CASE WHEN substr(id,-1,1) IN ('0','2','4','6','8','a','c','e') THEN '' ELSE 'fixture-model' END +FROM session_entries WHERE id BETWEEN ? AND ? ORDER BY id` + + if _, err := transaction.ExecContext(context.Background(), insertMessages, firstID, lastID); err != nil { + tb.Fatalf("insert fixture messages %d:%d: %v", start, end, err) + } + + const insertParts = `INSERT INTO session_message_parts +(entry_id,sequence,type,text,mime_type,name,width,height,data) +SELECT id,0,'text','deterministic provider-free transcript payload for entry ' || id || + ' used to measure normal repository hydration allocations','','',0,0,NULL +FROM session_entries WHERE id BETWEEN ? AND ? ORDER BY id` + + if _, err := transaction.ExecContext(context.Background(), insertParts, firstID, lastID); err != nil { + tb.Fatalf("insert fixture message parts %d:%d: %v", start, end, err) + } +} + +func measureBaselineStorage(ctx context.Context, connection *sql.DB, path string) (baselineStorage, error) { + objects, err := collectDBStat(ctx, connection) + if err != nil { + return baselineStorage{}, err + } + + pageSize, err := queryPragmaInt64(ctx, connection, "PRAGMA page_size") + if err != nil { + return baselineStorage{}, err + } + + pageCount, err := queryPragmaInt64(ctx, connection, "PRAGMA page_count") + if err != nil { + return baselineStorage{}, err + } + + freeList, err := queryPragmaInt64(ctx, connection, "PRAGMA freelist_count") + if err != nil { + return baselineStorage{}, err + } + + info, err := os.Stat(filepath.Clean(path)) + if err != nil { + return baselineStorage{}, fmt.Errorf("stat database: %w", err) + } + + var memory runtime.MemStats + runtime.ReadMemStats(&memory) + + activePages := int64(0) + for _, object := range objects { + activePages += object.Pages + } + + return baselineStorage{ + Objects: objects, + FileBytes: info.Size(), + PageSize: pageSize, + PageCount: pageCount, + FreeListPages: freeList, + ActiveDBStatPages: activePages, + HeapAllocBytes: memory.HeapAlloc, + RSSBytes: processRSSBytes(), + }, nil +} + +func queryPragmaInt64(ctx context.Context, connection *sql.DB, query string) (int64, error) { + var value int64 + + if err := connection.QueryRowContext(ctx, query).Scan(&value); err != nil { + return 0, fmt.Errorf("query %s: %w", query, err) + } + + return value, nil +} + +func collectDBStat(ctx context.Context, connection *sql.DB) (_ []baselinePageUsage, returnErr error) { + const query = `SELECT d.name,COALESCE(s.type,CASE WHEN d.name='sqlite_schema' THEN 'table' ELSE 'internal' END), + sum(d.pgsize),count(*),sum(d.payload),sum(d.unused) +FROM dbstat AS d LEFT JOIN sqlite_schema AS s ON s.name=d.name +GROUP BY d.name,COALESCE(s.type,CASE WHEN d.name='sqlite_schema' THEN 'table' ELSE 'internal' END) +ORDER BY d.name` + + rows, err := connection.QueryContext(ctx, query) + if err != nil { + return nil, fmt.Errorf("query SQLite dbstat (driver must include SQLITE_ENABLE_DBSTAT_VTAB): %w", err) + } + + defer func() { returnErr = closeRows(rows, "close dbstat rows", returnErr) }() + + usage := make([]baselinePageUsage, 0) + + for rows.Next() { + var item baselinePageUsage + + err := rows.Scan( + &item.Name, + &item.Kind, + &item.Bytes, + &item.Pages, + &item.PayloadBytes, + &item.UnusedBytes, + ) + if err != nil { + return nil, fmt.Errorf("scan dbstat: %w", err) + } + + usage = append(usage, item) + } + + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate dbstat: %w", err) + } + + return usage, nil +} + +func measureSameSecondCollisions( + ctx context.Context, + connection *sql.DB, +) (baselineCollisionStats, error) { + const query = `WITH second_groups AS ( + SELECT substr(created_at,1,19) AS unix_second,count(*) AS group_length + FROM session_entries GROUP BY substr(created_at,1,19) +) +SELECT COALESCE(sum(CASE WHEN group_length>1 THEN 1 ELSE 0 END),0), + COALESCE(sum(CASE WHEN group_length>1 THEN group_length ELSE 0 END),0), + COALESCE(max(group_length),0) FROM second_groups` + + var result baselineCollisionStats + + err := connection.QueryRowContext(ctx, query).Scan( + &result.EqualTimestampGroups, + &result.RowsInEqualGroups, + &result.MaxGroupLength, + ) + if err != nil { + return baselineCollisionStats{}, fmt.Errorf("query same-second collisions: %w", err) + } + + return result, nil +} + +func collectBaselineQueryPlans(ctx context.Context, connection *sql.DB) (map[string][]string, error) { + specifications := append(baselineQueryPlans(), executionBaselineQueryPlans()...) + plans := make(map[string][]string, len(specifications)) + + for _, specification := range specifications { + plan, err := collectBaselineQueryPlan(ctx, connection, specification) + if err != nil { + return nil, err + } + + plans[specification.Name] = plan + } + + return plans, nil +} + +func collectBaselineQueryPlan( + ctx context.Context, + connection *sql.DB, + specification baselineQueryPlanSpec, +) (_ []string, returnErr error) { + rows, err := connection.QueryContext( + ctx, + "EXPLAIN QUERY PLAN "+specification.SQL, + specification.Args..., + ) + if err != nil { + return nil, fmt.Errorf("explain %s: %w", specification.Name, err) + } + + defer func() { returnErr = closeRows(rows, "close explain "+specification.Name, returnErr) }() + + plan := make([]string, 0) + + for rows.Next() { + var ( + selectID int + parentID int + unusedID int + detail string + ) + + if err := rows.Scan(&selectID, &parentID, &unusedID, &detail); err != nil { + return nil, fmt.Errorf("scan explain %s: %w", specification.Name, err) + } + + plan = append(plan, fmt.Sprintf("%d|%d|%d|%s", selectID, parentID, unusedID, detail)) + } + + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate explain %s: %w", specification.Name, err) + } + + return plan, nil +} + +func indexInventory(ctx context.Context, connection *sql.DB) (_ []baselineIndex, returnErr error) { + const query = `SELECT name,tbl_name,COALESCE(sql,'') +FROM sqlite_schema WHERE type='index' ORDER BY name` + + rows, err := connection.QueryContext(ctx, query) + if err != nil { + return nil, fmt.Errorf("query index inventory: %w", err) + } + + defer func() { returnErr = closeRows(rows, "close index inventory", returnErr) }() + + indexes := make([]baselineIndex, 0) + + for rows.Next() { + var index baselineIndex + + if err := rows.Scan(&index.Name, &index.TableName, &index.SQL); err != nil { + return nil, fmt.Errorf("scan index inventory: %w", err) + } + + indexes = append(indexes, index) + } + + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate index inventory: %w", err) + } + + return indexes, nil +} + +func timestampIndexInventory(ctx context.Context, connection *sql.DB) (_ []string, returnErr error) { + const query = `SELECT name FROM sqlite_schema WHERE type='index' AND sql IS NOT NULL AND ( + lower(sql) LIKE '%created_at%' OR lower(sql) LIKE '%updated_at%' OR + lower(sql) LIKE '%finished_at%' OR lower(sql) LIKE '%lease_expires_at%' OR + lower(sql) LIKE '%consumed_at%') ORDER BY name` + + rows, err := connection.QueryContext(ctx, query) + if err != nil { + return nil, fmt.Errorf("query timestamp index inventory: %w", err) + } + + defer func() { returnErr = closeRows(rows, "close timestamp index inventory", returnErr) }() + + indexes := make([]string, 0) + + for rows.Next() { + var name string + + if err := rows.Scan(&name); err != nil { + return nil, fmt.Errorf("scan timestamp index inventory: %w", err) + } + + indexes = append(indexes, name) + } + + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate timestamp index inventory: %w", err) + } + + return indexes, nil +} + +func closeRows(rows *sql.Rows, message string, returnErr error) error { + closeErr := rows.Close() + if closeErr != nil { + return errors.Join(returnErr, fmt.Errorf("%s: %w", message, closeErr)) + } + + return returnErr +} + +func assertProductionTranscriptIndexes(t *testing.T, connection *sql.DB) { + t.Helper() + + for name, wantColumns := range expectedTranscriptIndexes() { + columns, err := indexColumns(t.Context(), connection, name) + if err != nil { + t.Fatalf("inspect transcript index %s: %v", name, err) + } + + if got := strings.Join(columns, ","); got != wantColumns { + t.Errorf("transcript index %s columns = %q, want %q", name, got, wantColumns) + } + } + + const cursorIndexSQL = `SELECT sql FROM sqlite_schema +WHERE type='index' AND name='idx_session_entries_transcript_cursor'` + + var cursorSQL string + + if err := connection.QueryRowContext(t.Context(), cursorIndexSQL).Scan(&cursorSQL); err != nil { + t.Fatalf("read transcript cursor index SQL: %v", err) + } + + if !strings.Contains(strings.ToLower(cursorSQL), "where display=1") { + t.Errorf("transcript cursor index is not the production partial index: %s", cursorSQL) + } +} + +func indexColumns(ctx context.Context, connection *sql.DB, name string) (_ []string, returnErr error) { + rows, err := connection.QueryContext( + ctx, + `SELECT name FROM pragma_index_info(?) ORDER BY seqno`, + name, + ) + if err != nil { + return nil, fmt.Errorf("query index columns: %w", err) + } + + defer func() { returnErr = closeRows(rows, "close index columns", returnErr) }() + + columns := make([]string, 0) + + for rows.Next() { + var column string + + if err := rows.Scan(&column); err != nil { + return nil, fmt.Errorf("scan index columns: %w", err) + } + + columns = append(columns, column) + } + + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate index columns: %w", err) + } + + return columns, nil +} + +func processRSSBytes() uint64 { + contents, err := os.ReadFile("/proc/self/statm") + if err == nil { + fields := strings.Fields(string(contents)) + if len(fields) >= 2 { + residentPages, parseErr := strconv.ParseUint(fields[1], 10, 64) + pageSize := os.Getpagesize() + + if parseErr == nil && pageSize >= 0 { + return residentPages * uint64(pageSize) + } + } + } + + var usage syscall.Rusage + + if err := syscall.Getrusage(syscall.RUSAGE_SELF, &usage); err != nil || usage.Maxrss < 0 { + return 0 + } + + if runtime.GOOS == "darwin" { + return uint64(usage.Maxrss) + } + + if usage.Maxrss > math.MaxInt64/1024 { + return math.MaxUint64 + } + + return uint64(usage.Maxrss * 1024) +} + +func expectedTimestampIndexes() []string { + return []string{ + "idx_events_completion_repair", + "idx_session_completion_global_pending", + "idx_session_entries_model_facing", + "idx_session_entries_session_created_id", + "idx_session_entries_tool_name", + "idx_session_entries_transcript_cursor", + "idx_sessions_cwd_parent_updated", + "idx_sessions_parent_updated", + "idx_tasks_completion_repair", + "idx_tasks_kind_state_created", + "idx_tasks_owner_state_updated", + "idx_tasks_owner_updated", + "idx_tasks_parent_updated", + "idx_tasks_recoverable_leases", + "idx_tasks_state_created", + } +} + +func expectedBaselinePlanNames() []string { + return []string{ + "newest_session", + "transcript_tail", + "transcript_cursor_pagination", + "current_leaf", + "queued_task_claim", + "lease_recovery", + "completion_repair", + } +} + +func expectedTranscriptIndexes() map[string]string { + return map[string]string{ + "idx_session_entries_session_created_id": "session_id,created_at,id", + "idx_session_entries_transcript_cursor": "session_id,created_at,id", + } +} + +func baselineQueryPlans() []baselineQueryPlanSpec { + plans := baselineSessionQueryPlans() + + return append(plans, baselineTaskQueryPlans()...) +} + +func baselineSessionQueryPlans() []baselineQueryPlanSpec { + return []baselineQueryPlanSpec{ + { + Name: "newest_session", + SQL: `SELECT id, cwd, name, parent_session_id, created_at, updated_at +FROM sessions +WHERE cwd = ? AND parent_session_id IS NULL +ORDER BY updated_at DESC, id DESC +LIMIT 1`, + Args: []any{"/fixture"}, + }, + { + Name: "transcript_tail", + SQL: `SELECT entry_id, session_id, custom_type, role, provider, model, created_at FROM ( +SELECT e.id AS entry_id, e.session_id, e.custom_type, m.role, m.provider, m.model, e.created_at +FROM session_entries AS e INDEXED BY idx_session_entries_transcript_cursor +JOIN session_messages AS m ON m.entry_id = e.id +WHERE e.session_id = ? AND e.display = 1 ORDER BY e.created_at DESC, e.id DESC LIMIT ?) +ORDER BY created_at ASC, entry_id ASC`, + Args: []any{compactTimestampSessionID, 256}, + }, + { + Name: "transcript_cursor_pagination", + SQL: `SELECT entry_id, session_id, custom_type, role, provider, model, created_at FROM ( +SELECT e.id AS entry_id, e.session_id, e.custom_type, m.role, m.provider, m.model, e.created_at +FROM session_entries AS e INDEXED BY idx_session_entries_transcript_cursor +JOIN session_messages AS m ON m.entry_id = e.id +WHERE e.session_id = ? AND e.display = 1 AND (e.created_at < ? OR (e.created_at = ? AND e.id < ?)) +ORDER BY e.created_at DESC, e.id DESC LIMIT ?) ORDER BY created_at ASC, entry_id ASC`, + Args: []any{ + compactTimestampSessionID, + "2025-01-01T01:00:00Z", + "2025-01-01T01:00:00Z", + "01910000-0000-7000-8000-000000008ca0", + 256, + }, + }, + } +} + +func executionBaselineQueryPlans() []baselineQueryPlanSpec { + return []baselineQueryPlanSpec{ + { + Name: "current_leaf", + SQL: `SELECT id, session_id, parent_id, entry_type, custom_type, data_json, summary, created_at, +tool_name, tool_status, tool_args_json, token_estimate, model_facing, display, +compaction_first_kept_entry_id, compaction_tokens_before, branch_from_entry_id +FROM session_entries +WHERE session_id = ? +ORDER BY created_at DESC, id DESC +LIMIT 1`, + Args: []any{compactTimestampSessionID}, + }, + } +} + +func baselineTaskQueryPlans() []baselineQueryPlanSpec { + return []baselineQueryPlanSpec{ + { + Name: "queued_task_claim", + SQL: `UPDATE tasks SET state = ?, started_at = COALESCE(started_at, ?), finished_at = NULL, +updated_at = ?, lease_owner = ?, lease_expires_at = ?, result = '', error_code = '', error_message = '' +WHERE id = ? AND state = ?`, + Args: []any{ + database.TaskRunning, + fixtureStartTimestamp, + fixtureStartTimestamp, + "worker", + "2025-01-01T00:01:00Z", + "01920000-0000-7000-8000-000000000001", + database.TaskQueued, + }, + }, + { + Name: "lease_recovery", + SQL: `SELECT id FROM tasks WHERE kind = ? AND state IN (?, ?) +AND (lease_expires_at IS NULL OR lease_expires_at <= ?) ORDER BY created_at, id LIMIT ?`, + Args: []any{ + database.TaskKindAgent, + database.TaskRunning, + database.TaskCanceling, + fixtureStartTimestamp, + 100, + }, + }, + { + Name: "completion_repair", + SQL: `SELECT e.id AS event_id,t.id AS task_id,e.kind AS event_kind,t.kind AS task_kind, + t.owner_session_id AS owner,t.state,t.result,t.error_code,t.error_message,e.created_at +FROM tasks t INDEXED BY idx_tasks_completion_repair +JOIN task_events te ON te.task_id=t.id + AND te.event_id=(SELECT candidate.event_id FROM task_events candidate + JOIN events candidate_event ON candidate_event.id=candidate.event_id + WHERE candidate.task_id=t.id + AND candidate_event.kind=CASE WHEN t.kind='agent' THEN 'task_'||t.state ELSE 'workflow_'||t.state END + ORDER BY candidate.sequence DESC LIMIT 1) +JOIN events e ON e.id=te.event_id +WHERE t.kind IN ('agent','workflow') + AND t.state IN ('succeeded','failed','canceled','interrupted') + AND (t.kind!='agent' OR NOT EXISTS (SELECT 1 FROM workflow_agent_tasks wa WHERE wa.agent_task_id=t.id)) + AND NOT EXISTS (SELECT 1 FROM session_completion_deliveries d + WHERE d.owner_session_id=t.owner_session_id AND d.event_id=e.id AND d.mapping_version=?) +ORDER BY t.finished_at,t.id LIMIT ?`, + Args: []any{database.CompletionMappingV1, 256}, + }, + } +} diff --git a/internal/database/completion_repository_test.go b/internal/database/completion_repository_test.go index 3c92b1af..3aea3ef8 100644 --- a/internal/database/completion_repository_test.go +++ b/internal/database/completion_repository_test.go @@ -3,6 +3,7 @@ package database_test import ( "context" "encoding/json" + "slices" "strings" "testing" "unicode/utf8" @@ -150,6 +151,57 @@ func TestCompletionRepairFindsTerminalEventBeforeLaterDiagnostics(t *testing.T) assert.Equal(t, 1, repaired) } +func TestCompletionRepairOrdersSameSecondCandidatesByTaskID(t *testing.T) { + t.Parallel() + + fixture := newTaskTestFixture(t) + ctx := t.Context() + owner := fixture.createOwner(ctx) + repositories, err := database.NewRepositories(fixture.connection) + require.NoError(t, err) + + taskIDs := make([]string, 0, 3) + + for range 3 { + task, createErr := repositories.Tasks.Create(ctx, newTask(owner.ID)) + require.NoError(t, createErr) + + finish := newTaskFinish(task.ID, []database.TaskState{database.TaskQueued}, database.TaskSucceeded, + taskSucceededEvent) + finish.Result = task.ID + changed, finishErr := repositories.Tasks.Finish(ctx, &finish) + require.NoError(t, finishErr) + require.True(t, changed) + + taskIDs = append(taskIDs, task.ID) + } + + finishedAt := "2026-08-30T12:00:00Z" + _, err = fixture.connection.ExecContext(ctx, + `UPDATE tasks SET finished_at = ? WHERE owner_session_id = ?`, finishedAt, owner.ID) + require.NoError(t, err) + _, err = fixture.connection.ExecContext(ctx, + `DELETE FROM session_completion_deliveries WHERE owner_session_id = ?`, owner.ID) + require.NoError(t, err) + + slices.Sort(taskIDs) + + for index, wantTaskID := range taskIDs { + repaired, repairErr := repositories.Completions.Repair(ctx, 1) + require.NoError(t, repairErr) + require.Equal(t, 1, repaired) + + pending, pendingErr := repositories.Completions.Pending(ctx, owner.ID, 16) + require.NoError(t, pendingErr) + require.Len(t, pending, index+1) + assert.Equal(t, wantTaskID, pending[index].TaskID, "repair %d", index) + } + + repaired, err := repositories.Completions.Repair(ctx, 1) + require.NoError(t, err) + assert.Zero(t, repaired) +} + func TestCompletionEnvelopeTreatsOutputAsTypedPlainData(t *testing.T) { t.Parallel() diff --git a/internal/database/migrations/00024_stabilize_session_timestamp_order.sql b/internal/database/migrations/00024_stabilize_session_timestamp_order.sql new file mode 100644 index 00000000..e55ece64 --- /dev/null +++ b/internal/database/migrations/00024_stabilize_session_timestamp_order.sql @@ -0,0 +1,17 @@ +-- +goose Up +DROP INDEX IF EXISTS idx_sessions_cwd_parent_updated; +DROP INDEX IF EXISTS idx_sessions_parent_updated; + +CREATE INDEX idx_sessions_cwd_parent_updated + ON sessions(cwd, parent_session_id, updated_at DESC, id DESC); +CREATE INDEX idx_sessions_parent_updated + ON sessions(parent_session_id, updated_at DESC, id DESC); + +-- +goose Down +DROP INDEX IF EXISTS idx_sessions_cwd_parent_updated; +DROP INDEX IF EXISTS idx_sessions_parent_updated; + +CREATE INDEX idx_sessions_cwd_parent_updated + ON sessions(cwd, parent_session_id, updated_at DESC); +CREATE INDEX idx_sessions_parent_updated + ON sessions(parent_session_id, updated_at DESC); diff --git a/internal/database/migrations_test.go b/internal/database/migrations_test.go index 9294769f..bed5d12f 100644 --- a/internal/database/migrations_test.go +++ b/internal/database/migrations_test.go @@ -13,9 +13,11 @@ import ( ) const ( - schemaIndexType = "index" - createdAtColumnName = "created_at" - sessionIDColumnName = "session_id" + schemaIndexType = "index" + createdAtColumnName = "created_at" + sessionIDColumnName = "session_id" + parentSessionIDColumnName = "parent_session_id" + updatedAtColumnName = "updated_at" deployedWorkflowMigrationV8 = `-- +goose Up CREATE TABLE workflow_runs ( @@ -361,6 +363,35 @@ func TestStartupQueryIndexMigration(t *testing.T) { "idx_tasks_state_created") } +func TestSessionTimestampOrderIndexMigrationUsesStableIDSuffix(t *testing.T) { + t.Parallel() + + connection := newMigratedThroughVersion(t, 23) + ctx := t.Context() + migrationRoot, err := database.MigrationFS() + require.NoError(t, err) + provider, err := database.NewMigrationProvider(connection, migrationRoot) + require.NoError(t, err) + + _, err = provider.UpTo(ctx, 24) + require.NoError(t, err) + assertIndexColumns(ctx, t, connection, "idx_sessions_cwd_parent_updated", []string{ + "cwd", parentSessionIDColumnName, updatedAtColumnName, "id", + }) + assertIndexColumns(ctx, t, connection, "idx_sessions_parent_updated", []string{ + parentSessionIDColumnName, updatedAtColumnName, "id", + }) + + _, err = provider.Down(ctx) + require.NoError(t, err) + assertIndexColumns(ctx, t, connection, "idx_sessions_cwd_parent_updated", []string{ + "cwd", parentSessionIDColumnName, updatedAtColumnName, + }) + assertIndexColumns(ctx, t, connection, "idx_sessions_parent_updated", []string{ + parentSessionIDColumnName, updatedAtColumnName, + }) +} + func TestTranscriptTailIndexMigrationUsesCursorColumns(t *testing.T) { t.Parallel() diff --git a/internal/database/session_entry_repository.go b/internal/database/session_entry_repository.go index 4270e2dd..b7b649cb 100644 --- a/internal/database/session_entry_repository.go +++ b/internal/database/session_entry_repository.go @@ -86,7 +86,7 @@ func (repository *SessionRepository) LeafEntry(ctx context.Context, sessionID st SELECT %s FROM session_entries WHERE session_id = ? -ORDER BY created_at DESC +ORDER BY created_at DESC, id DESC LIMIT 1`, entrySelectColumns) return repository.queryEntry(ctx, query, "leaf_entry", "load leaf entry", sessionID) @@ -98,7 +98,7 @@ func (repository *SessionRepository) Entries(ctx context.Context, sessionID stri SELECT %s FROM session_entries WHERE session_id = ? -ORDER BY created_at ASC`, entrySelectColumns) +ORDER BY created_at ASC, id ASC`, entrySelectColumns) return repository.queryEntries(ctx, query, "list_entries", "scan_entry", "entries", sessionID) } @@ -214,7 +214,7 @@ func (repository *SessionRepository) Children( SELECT %s FROM session_entries WHERE session_id = ? AND parent_id IS NULL -ORDER BY created_at ASC`, entrySelectColumns) +ORDER BY created_at ASC, id ASC`, entrySelectColumns) args := []any{sessionID} if parentID != nil { @@ -222,7 +222,7 @@ ORDER BY created_at ASC`, entrySelectColumns) SELECT %s FROM session_entries WHERE session_id = ? AND parent_id = ? -ORDER BY created_at ASC`, entrySelectColumns) +ORDER BY created_at ASC, id ASC`, entrySelectColumns) args = append(args, *parentID) } diff --git a/internal/database/session_repository.go b/internal/database/session_repository.go index 92fe34c4..4bc1a9d7 100644 --- a/internal/database/session_repository.go +++ b/internal/database/session_repository.go @@ -139,7 +139,7 @@ func (repository *SessionRepository) LatestSession(ctx context.Context, cwd stri SELECT id, cwd, name, parent_session_id, created_at, updated_at FROM sessions WHERE cwd = ? AND parent_session_id IS NULL -ORDER BY updated_at DESC +ORDER BY updated_at DESC, id DESC LIMIT 1` return repository.loadSession(ctx, query, "latest_session", "load latest session", cwd) @@ -185,7 +185,7 @@ func (repository *SessionRepository) ListSessions(ctx context.Context, cwd strin SELECT id, cwd, name, parent_session_id, created_at, updated_at FROM sessions WHERE cwd = ? AND parent_session_id IS NULL -ORDER BY updated_at DESC` +ORDER BY updated_at DESC, id DESC` rows := []sessionRow{} if err := repository.sql.Query(ctx, &rows, query, cwd); err != nil { @@ -209,7 +209,7 @@ func (repository *SessionRepository) ListChildSessions( SELECT id, cwd, name, parent_session_id, created_at, updated_at FROM sessions WHERE parent_session_id = ? -ORDER BY updated_at DESC` +ORDER BY updated_at DESC, id DESC` rows := []sessionRow{} if err := repository.sql.Query(ctx, &rows, query, parentSessionID); err != nil { diff --git a/internal/database/session_repository_test.go b/internal/database/session_repository_test.go index 7b27c45d..ef4e99c2 100644 --- a/internal/database/session_repository_test.go +++ b/internal/database/session_repository_test.go @@ -3,6 +3,7 @@ package database_test import ( "context" "database/sql" + "slices" "strings" "testing" "time" @@ -372,6 +373,119 @@ func TestSessionRepository_AppendMessagePreservesInputTimestamp(t *testing.T) { } } +func TestSessionRepository_OrdersEqualTimestampEntriesByID(t *testing.T) { + t.Parallel() + + repository := newTestSessionRepository(t) + ctx := context.Background() + session, err := repository.CreateSession(ctx, "/work", "entry-order", "") + require.NoError(t, err) + + helper := sessionTestHelper{ctx: ctx, t: t, repository: repository} + createdAt := time.Date(2026, time.August, 30, 12, 0, 0, 0, time.UTC) + roots := []*database.EntryEntity{ + helper.appendMessageAt(session.ID, nil, database.RoleUser, "root-1", createdAt), + helper.appendMessageAt(session.ID, nil, database.RoleUser, "root-2", createdAt), + helper.appendMessageAt(session.ID, nil, database.RoleUser, "root-3", createdAt), + } + children := []*database.EntryEntity{ + helper.appendMessageAt(session.ID, &roots[0].ID, database.RoleAssistant, "child-1", createdAt), + helper.appendMessageAt(session.ID, &roots[0].ID, database.RoleAssistant, "child-2", createdAt), + } + + wantAll := sortedEntryIDs(append(append([]*database.EntryEntity{}, roots...), children...)) + entries, err := repository.Entries(ctx, session.ID) + require.NoError(t, err) + assert.Equal(t, wantAll, sessionEntryIDs(entries)) + + leaf, found, err := repository.LeafEntry(ctx, session.ID) + require.NoError(t, err) + require.True(t, found) + assert.Equal(t, wantAll[len(wantAll)-1], leaf.ID) + + rootEntries, err := repository.Children(ctx, session.ID, nil) + require.NoError(t, err) + assert.Equal(t, sortedEntryIDs(roots), sessionEntryIDs(rootEntries)) + + childEntries, err := repository.Children(ctx, session.ID, &roots[0].ID) + require.NoError(t, err) + assert.Equal(t, sortedEntryIDs(children), sessionEntryIDs(childEntries)) + + tree, err := repository.Tree(ctx, session.ID) + require.NoError(t, err) + assert.Equal(t, sortedEntryIDs(roots), treeEntryIDs(tree)) + + for index := range tree { + if tree[index].Entry.ID == roots[0].ID { + assert.Equal(t, sortedEntryIDs(children), treeEntryIDs(tree[index].Children)) + } + } + + leafIDs := sortedEntryIDs([]*database.EntryEntity{roots[1], roots[2], children[0], children[1]}) + wantLeafID := leafIDs[len(leafIDs)-1] + branch, err := repository.Branch(ctx, session.ID, "") + require.NoError(t, err) + require.NotEmpty(t, branch) + assert.Equal(t, wantLeafID, branch[len(branch)-1].ID) + + for run := range 20 { + repeatedEntries, entriesErr := repository.Entries(ctx, session.ID) + require.NoError(t, entriesErr) + assert.Equal(t, wantAll, sessionEntryIDs(repeatedEntries), "run %d", run) + + repeatedRoots, rootsErr := repository.Children(ctx, session.ID, nil) + require.NoError(t, rootsErr) + assert.Equal(t, sortedEntryIDs(roots), sessionEntryIDs(repeatedRoots), "run %d", run) + } +} + +func TestSessionRepository_OrdersEqualTimestampSessionsByID(t *testing.T) { + t.Parallel() + + repository, connection := newTestSessionRepositoryWithConnection(t) + ctx := context.Background() + + parent, err := repository.CreateSession(ctx, "/work", "parent", "") + require.NoError(t, err) + + topLevel := make([]*database.SessionEntity, 1, 3) + topLevel[0] = parent + + for _, name := range []string{"top-2", "top-3"} { + session, createErr := repository.CreateSession(ctx, "/work", name, "") + require.NoError(t, createErr) + + topLevel = append(topLevel, session) + } + + children := make([]*database.SessionEntity, 0, 3) + + for _, name := range []string{"child-1", "child-2", "child-3"} { + session, createErr := repository.CreateSession(ctx, "/work", name, parent.ID) + require.NoError(t, createErr) + + children = append(children, session) + } + + updatedAt := time.Date(2026, time.August, 30, 12, 0, 0, 0, time.UTC).Format(time.RFC3339Nano) + _, err = connection.ExecContext(ctx, `UPDATE sessions SET updated_at = ? WHERE cwd = ?`, updatedAt, "/work") + require.NoError(t, err) + + wantTopLevel := sortedSessionIDsDescending(topLevel) + latest, found, err := repository.LatestSession(ctx, "/work") + require.NoError(t, err) + require.True(t, found) + assert.Equal(t, wantTopLevel[0], latest.ID) + + sessions, err := repository.ListSessions(ctx, "/work") + require.NoError(t, err) + assert.Equal(t, wantTopLevel, sessionIDs(sessions)) + + childSessions, err := repository.ListChildSessions(ctx, parent.ID) + require.NoError(t, err) + assert.Equal(t, sortedSessionIDsDescending(children), sessionIDs(childSessions)) +} + func TestSessionRepository_LoadsAndListsSessions(t *testing.T) { t.Parallel() @@ -504,6 +618,47 @@ func sessionEntryIDs(entries []database.EntryEntity) []string { return ids } +func sortedEntryIDs(entries []*database.EntryEntity) []string { + ids := make([]string, 0, len(entries)) + for index := range entries { + ids = append(ids, entries[index].ID) + } + + slices.Sort(ids) + + return ids +} + +func treeEntryIDs(nodes []database.TreeNodeEntity) []string { + ids := make([]string, 0, len(nodes)) + for index := range nodes { + ids = append(ids, nodes[index].Entry.ID) + } + + return ids +} + +func sortedSessionIDsDescending(sessions []*database.SessionEntity) []string { + ids := make([]string, 0, len(sessions)) + for index := range sessions { + ids = append(ids, sessions[index].ID) + } + + slices.Sort(ids) + slices.Reverse(ids) + + return ids +} + +func sessionIDs(sessions []database.SessionEntity) []string { + ids := make([]string, 0, len(sessions)) + for index := range sessions { + ids = append(ids, sessions[index].ID) + } + + return ids +} + func TestSessionRepository_BranchReturnsErrorForMissingEntryID(t *testing.T) { t.Parallel() @@ -690,6 +845,14 @@ func newMetadataFixture(ctx context.Context, t *testing.T) metadataFixture { func newTestSessionRepository(t *testing.T) *database.SessionRepository { t.Helper() + repository, _ := newTestSessionRepositoryWithConnection(t) + + return repository +} + +func newTestSessionRepositoryWithConnection(t *testing.T) (*database.SessionRepository, *sql.DB) { + t.Helper() + connection, err := sql.Open(sqliteDriver(), ":memory:") require.NoError(t, err) t.Cleanup(func() { @@ -699,7 +862,7 @@ func newTestSessionRepository(t *testing.T) *database.SessionRepository { require.NoError(t, database.Migrate(context.Background(), connection)) - return testutil.SessionRepository(t, connection) + return testutil.SessionRepository(t, connection), connection } func newMigratedThroughVersion(t *testing.T, version int64) *sql.DB { diff --git a/internal/database/session_store.go b/internal/database/session_store.go index d6cdacfb..7bed04b8 100644 --- a/internal/database/session_store.go +++ b/internal/database/session_store.go @@ -5,7 +5,7 @@ import ( "encoding/json" "errors" "fmt" - "sort" + "slices" "strings" "time" @@ -113,10 +113,12 @@ func (repository *SessionRepository) Tree(ctx context.Context, sessionID string) } for parentID := range childrenByParent { - sort.Slice(childrenByParent[parentID], func(leftIndex, rightIndex int) bool { - return childrenByParent[parentID][leftIndex].CreatedAt.Before( - childrenByParent[parentID][rightIndex].CreatedAt, - ) + slices.SortFunc(childrenByParent[parentID], func(left, right EntryEntity) int { + if byTime := left.CreatedAt.Compare(right.CreatedAt); byTime != 0 { + return byTime + } + + return strings.Compare(left.ID, right.ID) }) } diff --git a/internal/database/task_repository_branches_test.go b/internal/database/task_repository_branches_test.go index 904f9226..fc3f3a5f 100644 --- a/internal/database/task_repository_branches_test.go +++ b/internal/database/task_repository_branches_test.go @@ -2,6 +2,7 @@ package database_test import ( "path/filepath" + "slices" "strings" "testing" "time" @@ -66,6 +67,64 @@ func TestTaskRepositoryListFiltersAndLimits(t *testing.T) { assert.Len(t, allQueued, 4) } +func TestTaskRepositoryClaimsSameSecondQueuedTasksInStableIDOrder(t *testing.T) { + t.Parallel() + + fixture := newTaskTestFixture(t) + ctx, tasks := t.Context(), fixture.tasks + owner := fixture.createOwner(ctx) + + created := make([]*database.TaskEntity, 0, 3) + + for range 3 { + task, err := tasks.Create(ctx, newTask(owner.ID)) + require.NoError(t, err) + + created = append(created, task) + } + + createdAt := time.Date(2026, time.August, 30, 12, 0, 0, 0, time.UTC).Format(time.RFC3339Nano) + _, err := fixture.connection.ExecContext(ctx, `UPDATE tasks SET created_at = ? WHERE owner_session_id = ?`, + createdAt, owner.ID) + require.NoError(t, err) + + want := taskIDsFromPointers(created) + slices.Sort(want) + + queued, err := tasks.ListQueuedExcluding(ctx, nil, len(want)) + require.NoError(t, err) + assert.Equal(t, want, taskIDs(queued)) + + for index, wantID := range want { + candidates, listErr := tasks.ListByStates( + ctx, database.TaskKindAgent, []database.TaskState{database.TaskQueued}, 1, + ) + require.NoError(t, listErr) + require.Len(t, candidates, 1) + assert.Equal(t, wantID, candidates[0].ID, "claim %d", index) + + claimed, claimErr := tasks.ClaimQueued(ctx, &database.TaskClaim{ + TaskID: candidates[0].ID, LeaseOwner: testWorker, + LeaseExpiresAt: time.Now().Add(time.Minute), EventKind: taskStartedEvent, + }) + require.NoError(t, claimErr) + require.True(t, claimed) + } + + remaining, err := tasks.ListByStates(ctx, database.TaskKindAgent, []database.TaskState{database.TaskQueued}, 1) + require.NoError(t, err) + assert.Empty(t, remaining) +} + +func taskIDsFromPointers(tasks []*database.TaskEntity) []string { + ids := make([]string, len(tasks)) + for index := range tasks { + ids[index] = tasks[index].ID + } + + return ids +} + func TestTaskRepositoryClaimInterruptedAndRunningEvents(t *testing.T) { t.Parallel() diff --git a/internal/terminal/agent_tasks.go b/internal/terminal/agent_tasks.go index faa360ed..620c0022 100644 --- a/internal/terminal/agent_tasks.go +++ b/internal/terminal/agent_tasks.go @@ -2031,28 +2031,66 @@ func (app *App) appendMissingSessionMessages(messages []database.SessionMessageE } if appended { - slices.SortStableFunc(app.transcript.History, func(left, right chatMessage) int { - return left.CreatedAt.Compare(right.CreatedAt) - }) + slices.SortStableFunc(app.transcript.History, compareTranscriptMessages) app.transcript.LineCache.reset() } } +func compareTranscriptMessages(left, right chatMessage) int { + if byTime := left.CreatedAt.Compare(right.CreatedAt); byTime != 0 { + return byTime + } + + leftEntryID := transcriptEntryID(left) + rightEntryID := transcriptEntryID(right) + + if leftEntryID == "" && rightEntryID == "" { + return 0 + } + + if leftEntryID == "" { + return -1 + } + + if rightEntryID == "" { + return 1 + } + + return strings.Compare(leftEntryID, rightEntryID) +} + +func transcriptEntryID(message chatMessage) string { + if message.Identity == nil { + return "" + } + + return message.Identity.EntryID +} + func (app *App) hasSessionMessage(message *database.SessionMessageEntity) bool { + if message.EntryID != "" { + return app.hasDurableSessionMessage(message.EntryID) + } + role := transcript.FromDatabaseRole(message.Role) for index := range app.transcript.History { history := &app.transcript.History[index] - if message.EntryID != "" && history.EntryID != nil && *history.EntryID == message.EntryID { + missingDurableID := history.Identity == nil || history.Identity.EntryID == "" + matchingContent := history.Role == role && history.Content == message.Content + + if missingDurableID && matchingContent && history.CreatedAt.Equal(message.CreatedAt) { return true } + } - if history.EntryID == nil && history.CreatedAt.Equal(message.CreatedAt) && - history.Role == role && history.Content == message.Content { - if message.EntryID != "" { - history.EntryID = cloneStringPtr(&message.EntryID) - } + return false +} +func (app *App) hasDurableSessionMessage(entryID string) bool { + for index := range app.transcript.History { + identity := app.transcript.History[index].Identity + if identity != nil && identity.EntryID == entryID { return true } } diff --git a/internal/terminal/agent_tasks_behavior_internal_test.go b/internal/terminal/agent_tasks_behavior_internal_test.go index 5d0b7d5c..d8442cad 100644 --- a/internal/terminal/agent_tasks_behavior_internal_test.go +++ b/internal/terminal/agent_tasks_behavior_internal_test.go @@ -695,7 +695,7 @@ func TestMainSelectionReconcilesOptimisticPromptByEntryID(t *testing.T) { require.NoError(t, err) message := newChatMessage(transcript.RoleUser, prompt) - message.EntryID = cloneStringPtr(&entry.ID) + message.Identity = &chatMessageIdentity{EntryID: entry.ID, PromptID: 0} app.appendMessage(message) require.NoError(t, app.inspectAgentTask(t.Context(), behaviorTaskID)) @@ -726,7 +726,7 @@ func TestMainSelectionAfterParentPromptCompletionDoesNotDuplicateDurableMessages userMessage := newChatMessage(transcript.RoleUser, prompt) app.activePrompt.Prompt = prompt - app.activePrompt.UserMessageTimestamp = userMessage.CreatedAt.UnixNano() + userMessage.Identity = &chatMessageIdentity{EntryID: "", PromptID: app.activePrompt.ID} app.appendMessage(userMessage) promptID := app.activePrompt.ID @@ -783,8 +783,8 @@ func TestMainSelectionAfterParentPromptCompletionDoesNotDuplicateDurableMessages message := &app.transcript.History[index] messageCounts[message.Content]++ - if message.EntryID != nil { - entryIDsByContent[message.Content] = append(entryIDsByContent[message.Content], *message.EntryID) + if message.Identity != nil && message.Identity.EntryID != "" { + entryIDsByContent[message.Content] = append(entryIDsByContent[message.Content], message.Identity.EntryID) } } diff --git a/internal/terminal/app.go b/internal/terminal/app.go index d4d0690b..f892cf95 100644 --- a/internal/terminal/app.go +++ b/internal/terminal/app.go @@ -51,23 +51,27 @@ const ( modePanel appMode = "panel" ) +type chatMessageIdentity struct { + EntryID string + PromptID uint64 +} + type chatMessage struct { Attachments *attachmentSummaries + Identity *chatMessageIdentity CreatedAt time.Time - EntryID *string Role transcript.Role Content string } type activePromptState struct { - Cancel context.CancelFunc - SessionID string - UserEntryID string - Prompt string - Images []imageAttachment - UserMessageTimestamp int64 - ID uint64 - Canceled bool + Cancel context.CancelFunc + SessionID string + UserEntryID string + Prompt string + Images []imageAttachment + ID uint64 + Canceled bool } type resizeCoalescedEvent struct { @@ -816,13 +820,13 @@ func (app *App) appendSessionMessages(messages []database.SessionMessageEntity) func chatMessageFromSessionMessage(message *database.SessionMessageEntity) chatMessage { chat := chatMessage{ CreatedAt: message.CreatedAt, - EntryID: nil, Role: transcript.FromDatabaseRole(message.Role), Content: message.Content, Attachments: databaseAttachmentSummaries(message.Parts), + Identity: nil, } if message.EntryID != "" { - chat.EntryID = cloneStringPtr(&message.EntryID) + chat.Identity = &chatMessageIdentity{EntryID: message.EntryID, PromptID: 0} } return chat @@ -839,8 +843,8 @@ func (app *App) addMessage(role transcript.Role, content string) { func newChatMessage(role transcript.Role, content string) chatMessage { return chatMessage{ Attachments: nil, + Identity: nil, CreatedAt: time.Now().UTC(), - EntryID: nil, Role: role, Content: content, } diff --git a/internal/terminal/async_events.go b/internal/terminal/async_events.go index 19de0cfb..562fd2c4 100644 --- a/internal/terminal/async_events.go +++ b/internal/terminal/async_events.go @@ -669,7 +669,7 @@ func (app *App) applyPromptUserEntry(_ context.Context, sessionID, entryID strin previousSessionID := app.activePrompt.SessionID app.activePrompt.SessionID = sessionID app.activePrompt.UserEntryID = entryID - app.bindPromptUserMessageEntryID(entryID) + app.bindPromptUserMessageEntryID(promptID, entryID) if app.sessionID == previousSessionID { if app.sessionID != sessionID { @@ -680,18 +680,16 @@ func (app *App) applyPromptUserEntry(_ context.Context, sessionID, entryID strin } } -func (app *App) bindPromptUserMessageEntryID(entryID string) { - if entryID == "" || app.activePrompt.UserMessageTimestamp == 0 { +func (app *App) bindPromptUserMessageEntryID(promptID uint64, entryID string) { + if promptID == 0 || entryID == "" { return } for index := range app.transcript.History { message := &app.transcript.History[index] - - isPromptUserMessage := message.CreatedAt.UnixNano() == app.activePrompt.UserMessageTimestamp && - message.Role == transcript.RoleUser - if isPromptUserMessage { - message.EntryID = &entryID + if message.Identity != nil && message.Identity.PromptID == promptID && message.Role == transcript.RoleUser { + message.Identity.EntryID = entryID + app.transcript.HasOlder = true return } @@ -743,7 +741,7 @@ func (app *App) applySteeringConsumed(encoded string, promptID uint64) { } message := newChatMessage(transcript.RoleUser, draft.Text) - message.EntryID = &event.EntryID + message.Identity = &chatMessageIdentity{EntryID: event.EntryID, PromptID: 0} message.Attachments = summarizeAttachments(draft.Images) app.appendMessage(message) } diff --git a/internal/terminal/async_events_internal_test.go b/internal/terminal/async_events_internal_test.go index 8c137e13..bb27c31f 100644 --- a/internal/terminal/async_events_internal_test.go +++ b/internal/terminal/async_events_internal_test.go @@ -524,7 +524,7 @@ func promptUserEntryLifecycleCase() promptLifecycleCase { app.activePrompt = newTestActivePrompt(nil) app.activePrompt.ID = 3 message := newChatMessage(transcript.RoleUser, app.activePrompt.Prompt) - app.activePrompt.UserMessageTimestamp = message.CreatedAt.UnixNano() + message.Identity = &chatMessageIdentity{EntryID: "", PromptID: app.activePrompt.ID} app.appendMessage(message) }, assert: func(t *testing.T, app *App) { @@ -533,8 +533,7 @@ func promptUserEntryLifecycleCase() promptLifecycleCase { assert.Equal(t, asyncTestSessionID, app.activePrompt.SessionID) assert.Equal(t, asyncTestEntryID, app.activePrompt.UserEntryID) require.Len(t, app.transcript.History, 1) - require.NotNil(t, app.transcript.History[0].EntryID) - assert.Equal(t, asyncTestEntryID, *app.transcript.History[0].EntryID) + assert.Equal(t, asyncTestEntryID, app.transcript.History[0].Identity.EntryID) }, wantHandled: true, } @@ -551,7 +550,7 @@ func promptLifecycleEventCases() []promptLifecycleCase { app.transcript.Streaming.Blocks = []chatMessage{{ Attachments: nil, CreatedAt: time.Time{}, - EntryID: nil, + Identity: nil, Role: transcript.RoleAssistant, Content: asyncTestPartial, }} diff --git a/internal/terminal/extension_events_internal_test.go b/internal/terminal/extension_events_internal_test.go index d0cd14c2..149f3bce 100644 --- a/internal/terminal/extension_events_internal_test.go +++ b/internal/terminal/extension_events_internal_test.go @@ -53,7 +53,7 @@ func TestVimModePreservesModifiedEnterDelivery(t *testing.T) { app.working = true app.activePrompt = &activePromptState{ Cancel: nil, SessionID: app.sessionID, UserEntryID: "", Prompt: "", Images: nil, - UserMessageTimestamp: 0, ID: 1, Canceled: false, + ID: 1, Canceled: false, } app.composerBuffer.SetText("later") diff --git a/internal/terminal/interrupt_internal_test.go b/internal/terminal/interrupt_internal_test.go index 767fdf0f..9fa89fff 100644 --- a/internal/terminal/interrupt_internal_test.go +++ b/internal/terminal/interrupt_internal_test.go @@ -183,13 +183,12 @@ func newInterruptTestApp(t *testing.T, cancel context.CancelFunc) *App { func newTestActivePrompt(cancel context.CancelFunc) *activePromptState { return &activePromptState{ - Cancel: cancel, - SessionID: "", - UserEntryID: "", - Images: nil, - Prompt: interruptTestPrompt, - ID: 1, - UserMessageTimestamp: 0, - Canceled: false, + Cancel: cancel, + SessionID: "", + UserEntryID: "", + Images: nil, + Prompt: interruptTestPrompt, + ID: 1, + Canceled: false, } } diff --git a/internal/terminal/prompt_response.go b/internal/terminal/prompt_response.go index 46f026c2..ae189b47 100644 --- a/internal/terminal/prompt_response.go +++ b/internal/terminal/prompt_response.go @@ -15,7 +15,7 @@ func (app *App) applyPromptResponse(ctx context.Context, response *assistant.Pro } if response != nil { - app.bindPromptUserMessageEntryID(response.UserEntryID) + app.bindPromptUserMessageEntryID(promptID, response.UserEntryID) } if app.activePrompt.Canceled { @@ -50,7 +50,7 @@ func (app *App) applyPromptResponse(ctx context.Context, response *assistant.Pro message := newChatMessage(transcript.RoleAssistant, response.Text) if response.AssistantEntryID != "" { - message.EntryID = &response.AssistantEntryID + message.Identity = &chatMessageIdentity{EntryID: response.AssistantEntryID, PromptID: 0} } app.appendMessage(message) diff --git a/internal/terminal/prompt_response_internal_test.go b/internal/terminal/prompt_response_internal_test.go index ccf274df..dc064346 100644 --- a/internal/terminal/prompt_response_internal_test.go +++ b/internal/terminal/prompt_response_internal_test.go @@ -46,7 +46,7 @@ func TestApplyPromptResponseAssignsDurableEntryIDs(t *testing.T) { app := newRenderTestApp(t) app.activePrompt = newTestActivePrompt(nil) message := newChatMessage(transcript.RoleUser, "prompt") - app.activePrompt.UserMessageTimestamp = message.CreatedAt.UnixNano() + message.Identity = &chatMessageIdentity{EntryID: "", PromptID: app.activePrompt.ID} app.appendMessage(message) response := newTestPromptResponse("response") @@ -56,10 +56,8 @@ func TestApplyPromptResponseAssignsDurableEntryIDs(t *testing.T) { app.applyPromptResponse(context.Background(), response, app.activePrompt.ID) require.Len(t, app.transcript.History, 2) - require.NotNil(t, app.transcript.History[0].EntryID) - require.NotNil(t, app.transcript.History[1].EntryID) - assert.Equal(t, terminalTestUserID, *app.transcript.History[0].EntryID) - assert.Equal(t, "assistant-entry", *app.transcript.History[1].EntryID) + assert.Equal(t, terminalTestUserID, app.transcript.History[0].Identity.EntryID) + assert.Equal(t, "assistant-entry", app.transcript.History[1].Identity.EntryID) } func TestApplyPromptResponseNilClearsStreamedToolEvents(t *testing.T) { diff --git a/internal/terminal/prompt_send.go b/internal/terminal/prompt_send.go index 33462d56..54d6074b 100644 --- a/internal/terminal/prompt_send.go +++ b/internal/terminal/prompt_send.go @@ -46,22 +46,20 @@ func (app *App) sendDraft(ctx context.Context, draft promptDraft, visible bool) app.streamedToolEvents = 0 userMessage := newChatMessage(transcript.RoleUser, draft.Text) - userMessageTimestamp := int64(0) if visible { userMessage.Attachments = summarizeAttachments(draft.Images) - userMessageTimestamp = userMessage.CreatedAt.UnixNano() + userMessage.Identity = &chatMessageIdentity{EntryID: "", PromptID: promptID} } app.activePrompt = &activePromptState{ - Cancel: cancel, - SessionID: app.sessionID, - UserEntryID: "", - Prompt: draft.Text, - Images: cloneImageAttachments(draft.Images), - ID: promptID, - UserMessageTimestamp: userMessageTimestamp, - Canceled: false, + Cancel: cancel, + SessionID: app.sessionID, + UserEntryID: "", + Prompt: draft.Text, + Images: cloneImageAttachments(draft.Images), + ID: promptID, + Canceled: false, } if visible { app.appendMessage(userMessage) diff --git a/internal/terminal/prompt_send_internal_test.go b/internal/terminal/prompt_send_internal_test.go index 25876e0e..8fd6dddb 100644 --- a/internal/terminal/prompt_send_internal_test.go +++ b/internal/terminal/prompt_send_internal_test.go @@ -322,7 +322,8 @@ func TestActiveEnterSteersRuntime(t *testing.T) { consumed := app.transcript.History[len(app.transcript.History)-1] assert.Equal(t, transcript.RoleUser, consumed.Role) assert.Equal(t, "steer this", consumed.Content) - require.NotNil(t, consumed.EntryID) + require.NotNil(t, consumed.Identity) + assert.NotEmpty(t, consumed.Identity.EntryID) close(client.release) } @@ -392,7 +393,7 @@ func TestActiveInputKeyRouting(t *testing.T) { app.working = true app.activePrompt = &activePromptState{ Cancel: nil, SessionID: app.sessionID, UserEntryID: "", Prompt: "", Images: nil, - UserMessageTimestamp: 0, ID: 1, Canceled: false, + ID: 1, Canceled: false, } app.composerBuffer.SetText("draft") @@ -485,7 +486,7 @@ func TestRestoreReturnedSteeringPrecedesFollowUps(t *testing.T) { app := newRenderTestApp(t) app.activePrompt = &activePromptState{ Cancel: nil, SessionID: "", UserEntryID: "", Prompt: "", Images: nil, - UserMessageTimestamp: 0, ID: 7, Canceled: false, + ID: 7, Canceled: false, } app.queuedMessages = promptDrafts("follow-up") diff --git a/internal/terminal/running_tools_internal_test.go b/internal/terminal/running_tools_internal_test.go index 15f3bb3c..e605537f 100644 --- a/internal/terminal/running_tools_internal_test.go +++ b/internal/terminal/running_tools_internal_test.go @@ -155,7 +155,7 @@ func TestAgentTaskCompletionEventDrawsCollapsedExpandableToolResult(t *testing.T app.working = true app.activePrompt = &activePromptState{ Cancel: func() {}, SessionID: "", UserEntryID: "", - Images: nil, Prompt: "", ID: 1, UserMessageTimestamp: 0, Canceled: false, + Images: nil, Prompt: "", ID: 1, Canceled: false, } app.scrollOffset = 10 app.agentTasks = []database.AgentTaskEntity{testAgentTask(database.TaskRunning)} @@ -224,7 +224,7 @@ func TestAgentCompletionSurvivesPromptStreamingReset(t *testing.T) { app := newRenderTestApp(t) app.activePrompt = &activePromptState{ Cancel: func() {}, SessionID: "", UserEntryID: "", - Images: nil, Prompt: "", ID: 1, UserMessageTimestamp: 0, Canceled: false, + Images: nil, Prompt: "", ID: 1, Canceled: false, } content := formatAgentCompletionForUI("Agent explore finished.\n\nreview complete") app.addAgentCompletionMessage(content) @@ -248,7 +248,7 @@ func TestAgentCompletionStaysLiveAcrossQueuedContinuation(t *testing.T) { app := newRenderTestApp(t) app.activePrompt = &activePromptState{ Cancel: func() {}, SessionID: "", UserEntryID: "", - Images: nil, Prompt: "", ID: 1, UserMessageTimestamp: 0, Canceled: false, + Images: nil, Prompt: "", ID: 1, Canceled: false, } content := formatAgentCompletionForUI("Agent explore finished.\n\nreview complete") app.addAgentCompletionMessage(content) diff --git a/internal/terminal/scroll.go b/internal/terminal/scroll.go index 329cac91..bff1f827 100644 --- a/internal/terminal/scroll.go +++ b/internal/terminal/scroll.go @@ -142,14 +142,14 @@ func (app *App) hydrateOlderTranscript(ctx context.Context) error { } oldest := &app.transcript.History[0] - if oldest.EntryID == nil { + if !hasDurableChatMessageIdentity(oldest) { app.transcript.HasOlder = false return nil } messages, err := app.runtime.SessionRepository().TranscriptMessagesBefore( - ctx, app.sessionID, oldest.CreatedAt, *oldest.EntryID, transcriptHydrationBatch+1, + ctx, app.sessionID, oldest.CreatedAt, oldest.Identity.EntryID, transcriptHydrationBatch+1, ) if err != nil { return terminalError(err, "load older messages") @@ -183,6 +183,10 @@ func (app *App) hydrateOlderTranscript(ctx context.Context) error { return nil } +func hasDurableChatMessageIdentity(message *chatMessage) bool { + return message.Identity != nil && message.Identity.EntryID != "" +} + func (app *App) prependPromptHistory(messages []database.SessionMessageEntity) { texts := make([]string, 0, len(messages)+len(app.promptHistory)) images := make([][]imageAttachment, 0, len(messages)+len(app.promptHistoryImages)) diff --git a/internal/terminal/transcript_reconciliation_internal_test.go b/internal/terminal/transcript_reconciliation_internal_test.go index c46f30e8..4a9e14ed 100644 --- a/internal/terminal/transcript_reconciliation_internal_test.go +++ b/internal/terminal/transcript_reconciliation_internal_test.go @@ -22,8 +22,7 @@ func TestAppendMissingSessionMessagesReconcilesByEntryID(t *testing.T) { app := newRenderTestApp(t) local := newChatMessage(transcript.RoleUser, reconciliationContent) - entryID := reconciliationFirstEntry - local.EntryID = &entryID + local.Identity = &chatMessageIdentity{EntryID: reconciliationFirstEntry, PromptID: 0} app.appendMessage(local) app.appendMissingSessionMessages([]database.SessionMessageEntity{ @@ -31,8 +30,7 @@ func TestAppendMissingSessionMessagesReconcilesByEntryID(t *testing.T) { }) require.Len(t, app.transcript.History, 1) - require.NotNil(t, app.transcript.History[0].EntryID) - assert.Equal(t, reconciliationFirstEntry, *app.transcript.History[0].EntryID) + assert.Equal(t, reconciliationFirstEntry, app.transcript.History[0].Identity.EntryID) } func TestAppendMissingSessionMessagesPreservesRepeatedContentWithDistinctEntryIDs(t *testing.T) { @@ -50,14 +48,81 @@ func TestAppendMissingSessionMessagesPreservesRepeatedContentWithDistinctEntryID }) require.Len(t, app.transcript.History, 2) - require.NotNil(t, app.transcript.History[0].EntryID) - require.NotNil(t, app.transcript.History[1].EntryID) assert.Equal(t, []string{reconciliationFirstEntry, reconciliationSecondEntry}, []string{ - *app.transcript.History[0].EntryID, - *app.transcript.History[1].EntryID, + app.transcript.History[0].Identity.EntryID, + app.transcript.History[1].Identity.EntryID, }) } +func TestAppendMissingSessionMessagesDoesNotReconcileDurableEntryByTimestampAndContent(t *testing.T) { + t.Parallel() + + app := newRenderTestApp(t) + createdAt := time.Now().UTC() + local := newChatMessage(transcript.RoleUser, reconciliationContent) + local.CreatedAt = createdAt + local.Identity = &chatMessageIdentity{EntryID: "", PromptID: 1} + app.appendMessage(local) + + app.appendMissingSessionMessages([]database.SessionMessageEntity{ + testSessionMessage(createdAt, reconciliationFirstEntry), + }) + + require.Len(t, app.transcript.History, 2) + require.NotNil(t, app.transcript.History[0].Identity) + assert.Empty(t, app.transcript.History[0].Identity.EntryID) + assert.Equal(t, reconciliationFirstEntry, app.transcript.History[1].Identity.EntryID) +} + +func TestAppendMissingSessionMessagesOrdersEqualTimestampDurableEntriesByID(t *testing.T) { + t.Parallel() + + app := newRenderTestApp(t) + createdAt := time.Now().UTC() + app.appendMissingSessionMessages([]database.SessionMessageEntity{ + testSessionMessage(createdAt, reconciliationSecondEntry), + testSessionMessage(createdAt, reconciliationFirstEntry), + }) + + require.Len(t, app.transcript.History, 2) + assert.Equal(t, []string{reconciliationFirstEntry, reconciliationSecondEntry}, []string{ + app.transcript.History[0].Identity.EntryID, + app.transcript.History[1].Identity.EntryID, + }) +} + +func TestAppendMissingSessionMessagesPreservesEqualTimestampLocalInsertionOrder(t *testing.T) { + t.Parallel() + + for run := range 100 { + app := newRenderTestApp(t) + createdAt := time.Date(2026, time.August, 30, 12, 0, 0, 0, time.UTC) + first := newChatMessage(transcript.RoleUser, "first-local") + first.CreatedAt = createdAt + first.Identity = &chatMessageIdentity{EntryID: "", PromptID: 2} + second := newChatMessage(transcript.RoleUser, "second-local") + second.CreatedAt = createdAt + second.Identity = &chatMessageIdentity{EntryID: "", PromptID: 1} + + app.appendMessage(first) + app.appendMessage(second) + app.appendMissingSessionMessages([]database.SessionMessageEntity{ + testSessionMessage(createdAt, reconciliationSecondEntry), + testSessionMessage(createdAt, reconciliationFirstEntry), + }) + + require.Len(t, app.transcript.History, 4, "run %d", run) + assert.Equal(t, []string{"first-local", "second-local"}, []string{ + app.transcript.History[0].Content, + app.transcript.History[1].Content, + }, "run %d", run) + assert.Equal(t, []string{reconciliationFirstEntry, reconciliationSecondEntry}, []string{ + app.transcript.History[2].Identity.EntryID, + app.transcript.History[3].Identity.EntryID, + }, "run %d", run) + } +} + func TestAppendMissingSessionMessagesIsIdempotent(t *testing.T) { t.Parallel() @@ -71,20 +136,27 @@ func TestAppendMissingSessionMessagesIsIdempotent(t *testing.T) { assert.Equal(t, []string{reconciliationContent}, app.promptHistory) } -func TestBindPromptUserMessageEntryIDTargetsTrackedMessage(t *testing.T) { +func TestBindPromptUserMessageEntryIDTargetsLocalPromptIdentity(t *testing.T) { t.Parallel() app := newRenderTestApp(t) - app.appendMessage(newChatMessage(transcript.RoleUser, reconciliationContent)) - app.appendMessage(newChatMessage(transcript.RoleUser, reconciliationContent)) - app.activePrompt = newTestActivePrompt(nil) - app.activePrompt.UserMessageTimestamp = app.transcript.History[1].CreatedAt.UnixNano() - - app.bindPromptUserMessageEntryID(reconciliationSecondEntry) - - assert.Nil(t, app.transcript.History[0].EntryID) - require.NotNil(t, app.transcript.History[1].EntryID) - assert.Equal(t, reconciliationSecondEntry, *app.transcript.History[1].EntryID) + createdAt := time.Now().UTC() + first := newChatMessage(transcript.RoleUser, reconciliationContent) + first.CreatedAt = createdAt + first.Identity = &chatMessageIdentity{EntryID: "", PromptID: 1} + second := newChatMessage(transcript.RoleUser, reconciliationContent) + second.CreatedAt = createdAt + second.Identity = &chatMessageIdentity{EntryID: "", PromptID: 2} + + app.appendMessage(first) + app.appendMessage(second) + + app.bindPromptUserMessageEntryID(first.Identity.PromptID, reconciliationFirstEntry) + app.bindPromptUserMessageEntryID(second.Identity.PromptID, reconciliationSecondEntry) + + assert.Equal(t, reconciliationFirstEntry, app.transcript.History[0].Identity.EntryID) + assert.Equal(t, reconciliationSecondEntry, app.transcript.History[1].Identity.EntryID) + assert.True(t, app.transcript.HasOlder) } func testSessionMessage(createdAt time.Time, entryID string) database.SessionMessageEntity { diff --git a/internal/timestamp/timestamp.go b/internal/timestamp/timestamp.go new file mode 100644 index 00000000..9f374538 --- /dev/null +++ b/internal/timestamp/timestamp.go @@ -0,0 +1,290 @@ +// Package timestamp provides compact durable wall-clock timestamps. +// +// UnixSeconds stores a UTC instant as whole seconds since +// 1970-01-01T00:00:00Z. Its range is 0 through 4294967295, ending at +// 2106-02-07T06:28:15Z. Observed instants are floored to their containing +// second, while deadlines and duration additions are rounded upward so they +// are never encoded earlier than the source instant. Zero is the Unix epoch, +// not an absent value; use Optional when absence is meaningful. +package timestamp + +import ( + "bytes" + "database/sql/driver" + "encoding/json" + "fmt" + "strconv" + "time" + + "github.com/samber/oops" +) + +// UnixSeconds is a compact UTC Unix timestamp with one-second resolution. +type UnixSeconds uint32 + +// Max is the latest instant representable by UnixSeconds. +const ( + Max UnixSeconds = 1<<32 - 1 + decimalBase = 10 +) + +// Optional represents a UnixSeconds value that may be absent. +// Its zero value is absent; a present Unix epoch has Valid set to true. +type Optional struct { + UnixSeconds UnixSeconds + Valid bool +} + +// FromUnix validates seconds before converting them to UnixSeconds. +func FromUnix(seconds int64) (UnixSeconds, error) { + if seconds < 0 || seconds > int64(Max) { + return 0, invalidRangeError(seconds) + } + + return UnixSeconds(seconds), nil +} + +// FromTime converts an observed instant by flooring it to its containing second. +func FromTime(value time.Time) (UnixSeconds, error) { + return FromUnix(value.Unix()) +} + +// MustFromTime converts an observed instant or panics. It is intended for static +// fixtures and tests whose timestamps are known to be in range. +func MustFromTime(value time.Time) UnixSeconds { + converted, err := FromTime(value) + if err != nil { + panic(err) + } + + return converted +} + +// Time returns the UTC standard-library representation for presentation and APIs +// that require time.Time. +func (timestamp UnixSeconds) Time() time.Time { + return time.Unix(timestamp.Unix(), 0).UTC() +} + +// DeadlineFromTime converts a deadline by rounding it up to the next second +// when it has a fractional component. It rejects source instants outside the +// representable range, even when rounding could move them into range. +func DeadlineFromTime(value time.Time) (UnixSeconds, error) { + seconds := value.Unix() + if seconds < 0 || seconds > int64(Max) { + return 0, invalidRangeError(seconds) + } + + if value.Nanosecond() == 0 { + return UnixSeconds(seconds), nil + } + + if seconds == int64(Max) { + return 0, oops.In("timestamp").Code("timestamp_overflow"). + Errorf("deadline exceeds maximum Unix second %d", Max) + } + + return FromUnix(seconds + 1) +} + +// Unix returns the timestamp as signed Unix seconds for boundary operations. +func (timestamp UnixSeconds) Unix() int64 { + return int64(timestamp) +} + +// Add returns the timestamp plus duration, rounded upward to whole seconds. +// It rejects exact results outside the UnixSeconds range before narrowing. +func (timestamp UnixSeconds) Add(duration time.Duration) (UnixSeconds, error) { + wholeSeconds := int64(duration / time.Second) + remainder := duration % time.Second + result := timestamp.Unix() + wholeSeconds + + if result < 0 || result > int64(Max) || result == 0 && remainder < 0 { + return 0, oops.In("timestamp").Code("timestamp_arithmetic_out_of_range"). + Errorf("adding %s to Unix second %d is outside the supported range", duration, timestamp) + } + + if remainder > 0 { + if result == int64(Max) { + return 0, oops.In("timestamp").Code("timestamp_overflow"). + Errorf("adding %s to Unix second %d exceeds the supported range", duration, timestamp) + } + + result++ + } + + return FromUnix(result) +} + +// Sub returns the signed duration from other to timestamp. +func (timestamp UnixSeconds) Sub(other UnixSeconds) time.Duration { + seconds := timestamp.Unix() - other.Unix() + + return time.Duration(seconds) * time.Second +} + +// Compare compares timestamp with other and returns -1, 0, or 1. +func (timestamp UnixSeconds) Compare(other UnixSeconds) int { + switch { + case timestamp < other: + return -1 + case timestamp > other: + return 1 + default: + return 0 + } +} + +// Before reports whether timestamp occurs before other. +func (timestamp UnixSeconds) Before(other UnixSeconds) bool { + return timestamp < other +} + +// After reports whether timestamp occurs after other. +func (timestamp UnixSeconds) After(other UnixSeconds) bool { + return timestamp > other +} + +// Equal reports whether timestamp and other represent the same instant. +func (timestamp UnixSeconds) Equal(other UnixSeconds) bool { + return timestamp == other +} + +// Format formats the timestamp in UTC for presentation using a time layout. +func (timestamp UnixSeconds) Format(layout string) string { + return timestamp.Time().Format(layout) +} + +// MarshalJSON encodes timestamp as a JSON number. +func (timestamp UnixSeconds) MarshalJSON() ([]byte, error) { + return strconv.AppendUint(nil, uint64(timestamp), decimalBase), nil +} + +// UnmarshalJSON decodes a JSON integer after validating its range. +func (timestamp *UnixSeconds) UnmarshalJSON(data []byte) error { + trimmed := bytes.TrimSpace(data) + if bytes.Equal(trimmed, []byte("null")) { + return oops.In("timestamp").Code("invalid_timestamp_json"). + Errorf("Unix seconds cannot be null") + } + + var raw uint64 + if err := json.Unmarshal(trimmed, &raw); err != nil { + return oops.In("timestamp").Code("invalid_timestamp_json").Wrapf(err, "decode Unix seconds") + } + + if raw > uint64(Max) { + return oops.In("timestamp").Code("timestamp_out_of_range"). + Errorf("Unix seconds %d are outside the supported range 0..%d", raw, Max) + } + + *timestamp = UnixSeconds(raw) + + return nil +} + +// Value returns a signed integer suitable for database/sql and SQLite. +func (timestamp UnixSeconds) Value() (driver.Value, error) { + return timestamp.Unix(), nil +} + +// Scan validates and stores a signed SQL integer. NULL and all other SQL +// storage classes are invalid for a required timestamp. +func (timestamp *UnixSeconds) Scan(value any) error { + seconds, ok := value.(int64) + if !ok { + return invalidSQLTypeError(value) + } + + decoded, err := FromUnix(seconds) + if err != nil { + return oops.In("timestamp").Code("invalid_timestamp_sql").Wrapf(err, "scan Unix seconds") + } + + *timestamp = decoded + + return nil +} + +// MarshalJSON encodes a present timestamp as a number and an absent one as null. +func (optional Optional) MarshalJSON() ([]byte, error) { + if !optional.Valid { + return []byte("null"), nil + } + + return optional.UnixSeconds.MarshalJSON() +} + +// UnmarshalJSON decodes null as absent and a valid integer as present. +func (optional *Optional) UnmarshalJSON(data []byte) error { + trimmed := bytes.TrimSpace(data) + if bytes.Equal(trimmed, []byte("null")) { + *optional = Optional{UnixSeconds: 0, Valid: false} + + return nil + } + + var decoded UnixSeconds + if err := decoded.UnmarshalJSON(trimmed); err != nil { + return oops.In("timestamp").Code("invalid_optional_timestamp_json"). + Wrapf(err, "decode optional Unix seconds") + } + + *optional = Optional{UnixSeconds: decoded, Valid: true} + + return nil +} + +// Value returns nil for an absent timestamp and a signed integer when present. +func (optional Optional) Value() (value driver.Value, err error) { + if !optional.Valid { + return + } + + return optional.UnixSeconds.Value() +} + +// Scan stores SQL NULL as absent and validates a present signed integer. +func (optional *Optional) Scan(value any) error { + if value == nil { + *optional = Optional{UnixSeconds: 0, Valid: false} + + return nil + } + + var decoded UnixSeconds + if err := decoded.Scan(value); err != nil { + return oops.In("timestamp").Code("invalid_optional_timestamp_sql"). + Wrapf(err, "scan optional Unix seconds") + } + + *optional = Optional{UnixSeconds: decoded, Valid: true} + + return nil +} + +func invalidRangeError(seconds int64) error { + return oops.In("timestamp").Code("timestamp_out_of_range"). + Errorf("Unix seconds %d are outside the supported range 0..%d", seconds, Max) +} + +func invalidSQLTypeError(value any) error { + if value == nil { + return oops.In("timestamp").Code("invalid_timestamp_sql"). + Errorf("required Unix seconds cannot be NULL") + } + + return oops.In("timestamp").Code("invalid_timestamp_sql"). + Errorf("scan Unix seconds: expected int64, got %T (%s)", value, sqlValueDescription(value)) +} + +func sqlValueDescription(value any) string { + switch typed := value.(type) { + case string: + return strconv.Quote(typed) + case []byte: + return strconv.Quote(string(typed)) + default: + return fmt.Sprint(typed) + } +} diff --git a/internal/timestamp/timestamp_test.go b/internal/timestamp/timestamp_test.go new file mode 100644 index 00000000..d8a66d27 --- /dev/null +++ b/internal/timestamp/timestamp_test.go @@ -0,0 +1,468 @@ +package timestamp_test + +import ( + "database/sql" + "database/sql/driver" + "encoding/json" + "math" + "reflect" + "testing" + "time" + "unsafe" + + "github.com/omarluq/librecode/internal/timestamp" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const ( + negativeName = "negative" + epochName = "epoch" + nullName = "null" + nullJSON = `null` + overflowName = "overflow" + malformedName = "malformed" + maximumName = "maximum" +) + +func TestFromUnixValidatesFullRange(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + seconds int64 + want timestamp.UnixSeconds + wantError bool + }{ + {name: negativeName, seconds: -1, want: 0, wantError: true}, + {name: epochName, seconds: 0, want: 0, wantError: false}, + {name: "signed 32-bit maximum", seconds: math.MaxInt32, want: math.MaxInt32, wantError: false}, + {name: "past signed 32-bit", seconds: int64(math.MaxInt32) + 1, want: 1 << 31, wantError: false}, + {name: "unsigned 32-bit maximum", seconds: int64(math.MaxUint32), want: timestamp.Max, wantError: false}, + {name: "past unsigned 32-bit", seconds: int64(math.MaxUint32) + 1, want: 0, wantError: true}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + got, err := timestamp.FromUnix(test.seconds) + if test.wantError { + require.ErrorContains(t, err, "outside the supported range") + } else { + require.NoError(t, err) + } + + assert.Equal(t, test.want, got) + }) + } +} + +func TestTimeConversionsUseDocumentedQuantization(t *testing.T) { + t.Parallel() + + maxTime := time.Unix(int64(math.MaxUint32), 0).UTC() + tests := []struct { + input time.Time + convert func(time.Time) (timestamp.UnixSeconds, error) + name string + want timestamp.UnixSeconds + wantError bool + }{ + { + name: "observation exact second", input: time.Unix(10, 0), convert: timestamp.FromTime, + want: 10, wantError: false, + }, + { + name: "observation fractional second floors", input: time.Unix(10, 999_999_999), + convert: timestamp.FromTime, want: 10, wantError: false, + }, + { + name: "observation before epoch rejected", input: time.Unix(-1, 999_999_999), + convert: timestamp.FromTime, want: 0, wantError: true, + }, + { + name: "deadline exact second", input: time.Unix(10, 0), convert: timestamp.DeadlineFromTime, + want: 10, wantError: false, + }, + { + name: "deadline fractional second ceils", input: time.Unix(10, 1), + convert: timestamp.DeadlineFromTime, want: 11, wantError: false, + }, + { + name: "deadline before epoch rejected", input: time.Unix(-1, 999_999_999), + convert: timestamp.DeadlineFromTime, want: 0, wantError: true, + }, + { + name: "deadline maximum exact second", input: maxTime, convert: timestamp.DeadlineFromTime, + want: timestamp.Max, wantError: false, + }, + { + name: "deadline ceiling overflows", input: maxTime.Add(time.Nanosecond), + convert: timestamp.DeadlineFromTime, want: 0, wantError: true, + }, + { + name: "deadline second past maximum", input: maxTime.Add(time.Second), + convert: timestamp.DeadlineFromTime, want: 0, wantError: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + got, err := test.convert(test.input) + if test.wantError { + require.Error(t, err) + } else { + require.NoError(t, err) + } + + assert.Equal(t, test.want, got) + }) + } +} + +func TestAddRoundsUpAndChecksRange(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + base timestamp.UnixSeconds + duration time.Duration + want timestamp.UnixSeconds + wantError bool + }{ + {name: "zero", base: 10, duration: 0, want: 10, wantError: false}, + {name: "whole positive second", base: 10, duration: time.Second, want: 11, wantError: false}, + {name: "fractional positive second ceils", base: 10, duration: time.Nanosecond, want: 11, wantError: false}, + {name: "mixed positive duration ceils", base: 10, duration: time.Second + 1, want: 12, wantError: false}, + {name: "whole negative second", base: 10, duration: -time.Second, want: 9, wantError: false}, + {name: "fractional negative duration ceils", base: 10, duration: -time.Nanosecond, want: 10, wantError: false}, + {name: "mixed negative duration ceils", base: 10, duration: -time.Second - 1, want: 9, wantError: false}, + {name: "underflow whole second", base: 0, duration: -time.Second, want: 0, wantError: true}, + {name: "underflow fractional second", base: 0, duration: -time.Nanosecond, want: 0, wantError: true}, + {name: "maximum unchanged", base: timestamp.Max, duration: 0, want: timestamp.Max, wantError: false}, + {name: "overflow whole second", base: timestamp.Max, duration: time.Second, want: 0, wantError: true}, + {name: "overflow while ceiling", base: timestamp.Max, duration: time.Nanosecond, want: 0, wantError: true}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + got, err := test.base.Add(test.duration) + if test.wantError { + require.Error(t, err) + } else { + require.NoError(t, err) + } + + assert.Equal(t, test.want, got) + }) + } +} + +func TestComparisonSubtractionAndFormatting(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + left timestamp.UnixSeconds + right timestamp.UnixSeconds + wantCompare int + wantBefore bool + wantAfter bool + wantEqual bool + }{ + { + name: "before", left: 1, right: 2, wantCompare: -1, + wantBefore: true, wantAfter: false, wantEqual: false, + }, + { + name: "equal", left: 2, right: 2, wantCompare: 0, + wantBefore: false, wantAfter: false, wantEqual: true, + }, + { + name: "after", left: timestamp.Max, right: 0, wantCompare: 1, + wantBefore: false, wantAfter: true, wantEqual: false, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + assert.Equal(t, test.wantCompare, test.left.Compare(test.right)) + assert.Equal(t, test.wantBefore, test.left.Before(test.right)) + assert.Equal(t, test.wantAfter, test.left.After(test.right)) + assert.Equal(t, test.wantEqual, test.left.Equal(test.right)) + }) + } + + assert.Equal(t, time.Duration(math.MaxUint32)*time.Second, timestamp.Max.Sub(0)) + assert.Equal(t, -time.Duration(math.MaxUint32)*time.Second, timestamp.UnixSeconds(0).Sub(timestamp.Max)) + assert.Equal(t, "1970-01-01 00:00:00 UTC", timestamp.UnixSeconds(0).Format("2006-01-02 15:04:05 MST")) + assert.Equal(t, "2106-02-07T06:28:15Z", timestamp.Max.Format(time.RFC3339)) +} + +func TestUnixSecondsJSONRoundTripsNumbers(t *testing.T) { + t.Parallel() + + for _, value := range []timestamp.UnixSeconds{0, math.MaxInt32, 1 << 31, timestamp.Max} { + t.Run(value.Format("20060102T150405"), func(t *testing.T) { + t.Parallel() + + encoded, err := json.Marshal(value) + require.NoError(t, err) + assert.Equal(t, value.Unix(), mustJSONInteger(t, encoded)) + + var decoded timestamp.UnixSeconds + require.NoError(t, json.Unmarshal(encoded, &decoded)) + assert.Equal(t, value, decoded) + }) + } +} + +func TestUnixSecondsJSONRejectsInvalidValues(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + input string + }{ + {name: nullName, input: nullJSON}, + {name: negativeName, input: `-1`}, + {name: "fractional", input: `1.5`}, + {name: overflowName, input: `4294967296`}, + {name: "string", input: `"1"`}, + {name: "boolean", input: `true`}, + {name: "object", input: `{}`}, + {name: malformedName, input: `1x`}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + value := timestamp.UnixSeconds(42) + require.Error(t, json.Unmarshal([]byte(test.input), &value)) + assert.Equal(t, timestamp.UnixSeconds(42), value) + }) + } +} + +func TestOptionalJSONDistinguishesAbsentFromEpoch(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + input string + want timestamp.Optional + }{ + {name: "absent", input: nullJSON, want: timestamp.Optional{UnixSeconds: 0, Valid: false}}, + {name: epochName, input: `0`, want: timestamp.Optional{UnixSeconds: 0, Valid: true}}, + {name: maximumName, input: `4294967295`, want: timestamp.Optional{UnixSeconds: timestamp.Max, Valid: true}}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + var decoded timestamp.Optional + require.NoError(t, json.Unmarshal([]byte(test.input), &decoded)) + assert.Equal(t, test.want, decoded) + + encoded, err := json.Marshal(decoded) + require.NoError(t, err) + assert.JSONEq(t, test.input, string(encoded)) + }) + } + + invalid := []string{`-1`, `1.5`, `4294967296`, `"1"`, `true`, `{}`, `1x`} + for _, input := range invalid { + value := timestamp.Optional{UnixSeconds: 42, Valid: true} + require.Error(t, json.Unmarshal([]byte(input), &value)) + assert.Equal(t, timestamp.Optional{UnixSeconds: 42, Valid: true}, value) + } +} + +func TestSQLValueUsesSignedIntegersAndNull(t *testing.T) { + t.Parallel() + + tests := []struct { + value driver.Valuer + want driver.Value + name string + }{ + {name: "required epoch", value: timestamp.UnixSeconds(0), want: int64(0)}, + {name: "required maximum", value: timestamp.Max, want: int64(math.MaxUint32)}, + {name: "optional absent", value: timestamp.Optional{UnixSeconds: 0, Valid: false}, want: nil}, + { + name: "optional epoch", value: timestamp.Optional{UnixSeconds: 0, Valid: true}, + want: int64(0), + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + got, err := test.value.Value() + require.NoError(t, err) + assert.Equal(t, test.want, got) + }) + } +} + +func TestUnixSecondsSQLScanValidatesStorageClassAndRange(t *testing.T) { + t.Parallel() + + tests := []struct { + input any + name string + want timestamp.UnixSeconds + wantError bool + }{ + {name: "null", input: nil, want: 0, wantError: true}, + {name: "negative", input: int64(-1), want: 0, wantError: true}, + {name: "epoch", input: int64(0), want: 0, wantError: false}, + {name: "signed 32-bit maximum", input: int64(math.MaxInt32), want: math.MaxInt32, wantError: false}, + {name: "past signed 32-bit", input: int64(math.MaxInt32) + 1, want: 1 << 31, wantError: false}, + {name: "maximum", input: int64(math.MaxUint32), want: timestamp.Max, wantError: false}, + {name: "overflow", input: int64(math.MaxUint32) + 1, want: 0, wantError: true}, + {name: "text", input: "1", want: 0, wantError: true}, + {name: "bytes", input: []byte("1"), want: 0, wantError: true}, + {name: "real", input: float64(1), want: 0, wantError: true}, + {name: "malformed", input: struct{}{}, want: 0, wantError: true}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + value := timestamp.UnixSeconds(42) + + err := value.Scan(test.input) + if test.wantError { + require.Error(t, err) + assert.Equal(t, timestamp.UnixSeconds(42), value) + + return + } + + require.NoError(t, err) + assert.Equal(t, test.want, value) + }) + } +} + +func TestOptionalSQLScanDistinguishesNullFromEpoch(t *testing.T) { + t.Parallel() + + tests := []struct { + input any + name string + want timestamp.Optional + wantError bool + }{ + {name: nullName, input: nil, want: timestamp.Optional{UnixSeconds: 0, Valid: false}, wantError: false}, + {name: epochName, input: int64(0), want: timestamp.Optional{UnixSeconds: 0, Valid: true}, wantError: false}, + { + name: "maximum", input: int64(math.MaxUint32), + want: timestamp.Optional{UnixSeconds: timestamp.Max, Valid: true}, wantError: false, + }, + { + name: negativeName, input: int64(-1), + want: timestamp.Optional{UnixSeconds: 0, Valid: false}, wantError: true, + }, + { + name: overflowName, input: int64(math.MaxUint32) + 1, + want: timestamp.Optional{UnixSeconds: 0, Valid: false}, wantError: true, + }, + { + name: "text", input: "1", + want: timestamp.Optional{UnixSeconds: 0, Valid: false}, wantError: true, + }, + { + name: "real", input: float64(1), + want: timestamp.Optional{UnixSeconds: 0, Valid: false}, wantError: true, + }, + { + name: malformedName, input: []byte("bad"), + want: timestamp.Optional{UnixSeconds: 0, Valid: false}, wantError: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + value := timestamp.Optional{UnixSeconds: 42, Valid: true} + + err := value.Scan(test.input) + if test.wantError { + require.Error(t, err) + assert.Equal(t, timestamp.Optional{UnixSeconds: 42, Valid: true}, value) + + return + } + + require.NoError(t, err) + assert.Equal(t, test.want, value) + }) + } +} + +func TestCompactLayoutsAreInlineAndPointerFree(t *testing.T) { + t.Parallel() + + assert.Equal(t, uintptr(4), unsafe.Sizeof(timestamp.UnixSeconds(0))) + assert.Equal(t, uintptr(8), unsafe.Sizeof(timestamp.Optional{UnixSeconds: 0, Valid: false})) + assert.False(t, containsPointer(reflect.TypeFor[timestamp.UnixSeconds]())) + assert.False(t, containsPointer(reflect.TypeFor[timestamp.Optional]())) + assert.Equal(t, reflect.TypeFor[timestamp.UnixSeconds](), reflect.TypeFor[timestamp.Optional]().Field(0).Type) +} + +func mustJSONInteger(t *testing.T, data []byte) int64 { + t.Helper() + + var value int64 + require.NoError(t, json.Unmarshal(data, &value)) + + return value +} + +func containsPointer(valueType reflect.Type) bool { + switch valueType.Kind() { + case reflect.Array: + return containsPointer(valueType.Elem()) + case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice, reflect.String, + reflect.UnsafePointer: + return true + case reflect.Struct: + for field := range valueType.Fields() { + if containsPointer(field.Type) { + return true + } + } + case reflect.Invalid, reflect.Bool, + reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64, + reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64, reflect.Uintptr, + reflect.Float32, reflect.Float64, reflect.Complex64, reflect.Complex128: + return false + } + + return false +} + +var ( + _ json.Marshaler = timestamp.UnixSeconds(0) + _ json.Unmarshaler = (*timestamp.UnixSeconds)(nil) + _ driver.Valuer = timestamp.UnixSeconds(0) + _ sql.Scanner = (*timestamp.UnixSeconds)(nil) + _ json.Marshaler = timestamp.Optional{UnixSeconds: 0, Valid: false} + _ json.Unmarshaler = (*timestamp.Optional)(nil) + _ driver.Valuer = timestamp.Optional{UnixSeconds: 0, Valid: false} + _ sql.Scanner = (*timestamp.Optional)(nil) +)