diff --git a/internal/channel/session/manager.go b/internal/channel/session/manager.go index 8d3e166..4a0084a 100644 --- a/internal/channel/session/manager.go +++ b/internal/channel/session/manager.go @@ -19,6 +19,11 @@ import ( // channel maps to exactly one internal session_id. created_at/updated_at follow // the project's timestamp convention; actor_role records the identity lane so the // mapping survives restarts. +// +// The created_at/updated_at DEFAULT expressions are a fallback only: every write +// path (insertOrIgnoreSessionSQL) supplies explicit fixed-width UTC timestamps +// via formatSessionTime, so the 3-digit-millisecond strftime defaults are never +// exercised and no row mixes timestamp widths. const createTableSQL = ` CREATE TABLE IF NOT EXISTS channel_sessions ( platform TEXT NOT NULL, @@ -43,10 +48,13 @@ SELECT session_id FROM channel_sessions WHERE platform = ? AND channel_id = ?` // listSessionsSQL selects every persisted channel mapping, newest first. Group // sessions key on a channel_id carrying groupChannelPrefix; direct sessions use -// their channel id. +// their channel id. created_at is fixed-width UTC text (see sessionTimeLayout), +// so its text order is its time order; rowid breaks created_at ties so mappings +// written in the same coarse clock tick still list in a stable order +// (channel_sessions is an ordinary rowid table, not WITHOUT ROWID). const listSessionSQL = ` SELECT platform, channel_id, session_id, actor_role FROM channel_sessions -ORDER BY created_at DESC` +ORDER BY created_at DESC, rowid DESC` // groupChannelPrefix marks a channel id as a group conversation so a session // reconstructed from the backing table (where no IsGroup flag is persisted) can @@ -178,7 +186,7 @@ func (m *SessionManager) lookupCached(key string) (Session, bool) { // winning writer's session_id becomes the binding. It is called under the write // lock so the in-memory cache and the table stay in agreement. func (m *SessionManager) persist(ctx context.Context, platform, cid string, actor adapter.Actor) (string, error) { - now := time.Now().UTC().Format(time.RFC3339Nano) + now := formatSessionTime(time.Now()) if _, err := m.db.ExecContext(ctx, insertOrIgnoreSessionSQL, platform, cid, newSessionID(), string(actor.Role), now, now); err != nil { return "", fmt.Errorf("session: persist %s/%s: %w", platform, cid, err) } diff --git a/internal/channel/session/manager_test.go b/internal/channel/session/manager_test.go index f2cb45f..86078bb 100644 --- a/internal/channel/session/manager_test.go +++ b/internal/channel/session/manager_test.go @@ -318,3 +318,98 @@ func TestNewRejectsNilDB(t *testing.T) { t.Fatalf("New(nil db) error = %v, want ErrNilDB", err) } } + +// insertSessionAt writes a channel_sessions row directly with a caller-supplied +// raw created_at/updated_at, standing in for a mapping written by past code. +func insertSessionAt(t *testing.T, handle *sql.DB, cid, at string) { + t.Helper() + if _, err := handle.ExecContext(context.Background(), + `INSERT INTO channel_sessions (platform, channel_id, session_id, actor_role, created_at, updated_at) + VALUES ('qq', ?, ?, 'owner', ?, ?)`, + cid, "sess-"+cid, at, at); err != nil { + t.Fatalf("insert %s: %v", cid, err) + } +} + +// TestListNewestFirstDeterministic covers the ordering contract: three mappings +// written within the same coarse clock tick share one created_at, so only the +// rowid tiebreak can define their order. List is newest first, so the row +// inserted last (highest rowid) must come first. The cache is deliberately empty +// so List reads the persisted order instead of cached entries. +func TestListNewestFirstDeterministic(t *testing.T) { + t.Parallel() + + m, handle := newTestManager(t) + ctx := context.Background() + + const at = "2026-09-07T03:20:05.830000000Z" + for _, cid := range []string{"direct-a", "direct-b", "direct-c"} { + insertSessionAt(t, handle, cid, at) + } + + got, err := m.List(ctx) + if err != nil { + t.Fatalf("List = %v", err) + } + want := []string{"direct-c", "direct-b", "direct-a"} + if len(got) != len(want) { + t.Fatalf("List = %d sessions, want %d", len(got), len(want)) + } + for i, w := range want { + if got[i].ChannelID != w { + t.Errorf("got[%d].ChannelID = %q, want %q (rowid DESC tiebreak)", i, got[i].ChannelID, w) + } + } +} + +// TestListNewestFirstByCreatedAt checks the created_at DESC half: the mapping +// with the latest fixed-width timestamp is listed first even though it was +// inserted first and therefore has the lowest rowid. +func TestListNewestFirstByCreatedAt(t *testing.T) { + t.Parallel() + + m, handle := newTestManager(t) + ctx := context.Background() + + insertSessionAt(t, handle, "direct-newest", "2026-09-07T03:20:05.830978900Z") + insertSessionAt(t, handle, "direct-oldest", "2026-09-07T03:20:05.830000000Z") + + got, err := m.List(ctx) + if err != nil { + t.Fatalf("List = %v", err) + } + if len(got) != 2 { + t.Fatalf("List = %d sessions, want 2", len(got)) + } + if got[0].ChannelID != "direct-newest" || got[1].ChannelID != "direct-oldest" { + t.Errorf("List order = [%q %q], want newest first", got[0].ChannelID, got[1].ChannelID) + } +} + +// TestPersistWritesFixedWidthTimestamps pins the write path: GetOrCreate must +// store created_at/updated_at with the fixed-width layout so channel_sessions +// text ordering matches time ordering. RFC3339Nano would drop trailing zeros +// and persist a variable-width value instead. +func TestPersistWritesFixedWidthTimestamps(t *testing.T) { + t.Parallel() + + m, handle := newTestManager(t) + ctx := context.Background() + + if _, err := m.GetOrCreate(ctx, directMsg("o-100"), testOwner()); err != nil { + t.Fatalf("GetOrCreate = %v", err) + } + var created, updated string + if err := handle.QueryRowContext(ctx, + `SELECT created_at, updated_at FROM channel_sessions WHERE channel_id = ?`, + "direct-o-100").Scan(&created, &updated); err != nil { + t.Fatalf("read stored timestamps: %v", err) + } + if len(created) != sessionTimeLen || len(updated) != sessionTimeLen { + t.Errorf("stored timestamps = %q/%q (%d/%d bytes), want fixed width %d", + created, updated, len(created), len(updated), sessionTimeLen) + } + if created != updated { + t.Errorf("created_at %q != updated_at %q, want the same value on insert", created, updated) + } +} diff --git a/internal/channel/session/timefmt.go b/internal/channel/session/timefmt.go new file mode 100644 index 0000000..7ba44c0 --- /dev/null +++ b/internal/channel/session/timefmt.go @@ -0,0 +1,21 @@ +package session + +import "time" + +// sessionTimeLayout is the fixed-width timestamp layout used for every stored +// created_at/updated_at value. Unlike time.RFC3339Nano it never drops trailing +// fractional zeros, so every value is exactly sessionTimeLen bytes wide. SQLite +// compares these TEXT columns as plain strings, and equal width is what makes +// that comparison agree with chronological order. Values are written in UTC, so +// the Z07:00 zone renders as a single "Z". +const sessionTimeLayout = "2006-01-02T15:04:05.000000000Z07:00" + +// sessionTimeLen is the byte length of a UTC value formatted with +// sessionTimeLayout: "2006-01-02T15:04:05.000000000Z" — the Z07:00 zone +// collapses to "Z" in UTC. +const sessionTimeLen = 30 + +// formatSessionTime encodes t in UTC with the fixed-width sessionTimeLayout. +func formatSessionTime(t time.Time) string { + return t.UTC().Format(sessionTimeLayout) +} diff --git a/internal/runtime/persist/persist.go b/internal/runtime/persist/persist.go index 0fb4ef7..c31f8d5 100644 --- a/internal/runtime/persist/persist.go +++ b/internal/runtime/persist/persist.go @@ -94,9 +94,13 @@ const ( loadSQL = `SELECT id, agent_id, step, state, created_at, updated_at FROM checkpoints WHERE id = ?` + // updated_at is stored as fixed-width UTC text (see formatTime), so its text + // order is its time order. step breaks ties between checkpoints written in + // the same tick, and rowid makes that tiebreak deterministic (checkpoints is + // an ordinary rowid table, not WITHOUT ROWID). loadLatestSQL = `SELECT id, agent_id, step, state, created_at, updated_at FROM checkpoints WHERE agent_id = ? - ORDER BY updated_at DESC, step DESC LIMIT 1` + ORDER BY updated_at DESC, step DESC, rowid DESC LIMIT 1` listSQL = `SELECT id, agent_id, step, state, created_at, updated_at FROM checkpoints WHERE agent_id = ? @@ -112,6 +116,8 @@ const ( // which already constrains the pool to a single connection for write // serialization. The caller owns db and must Close it when done. log may be nil // and is replaced with a discard logger so the Store can always log safely. +// Legacy variable-width timestamps are normalized in place (see +// normalizeTimestamps) so text ordering matches time ordering. func New(ctx context.Context, db *sql.DB, log *slog.Logger) (*Store, error) { if db == nil { return nil, fmt.Errorf("persist: new: %w", ErrNilDB) @@ -124,6 +130,9 @@ func New(ctx context.Context, db *sql.DB, log *slog.Logger) (*Store, error) { if _, err := db.ExecContext(ctx, createIndexSQL); err != nil { return nil, fmt.Errorf("persist: create index: %w", err) } + if err := normalizeTimestamps(ctx, db); err != nil { + return nil, err + } return &Store{db: db, log: log}, nil } @@ -270,16 +279,6 @@ func scanCheckpoint(sc rowScanner) (*Checkpoint, error) { }, nil } -// formatTime renders t as a UTC RFC3339 timestamp with nanosecond precision. -func formatTime(t time.Time) string { - return t.UTC().Format(time.RFC3339Nano) -} - -// parseTime parses a timestamp written by formatTime. -func parseTime(s string) (time.Time, error) { - return time.Parse(time.RFC3339Nano, s) -} - // discardLogger returns log if non-nil, otherwise a logger that writes nowhere, // so the Store can always log safely. func discardLogger(log *slog.Logger) *slog.Logger { diff --git a/internal/runtime/persist/persist_test.go b/internal/runtime/persist/persist_test.go index d9755c4..1305142 100644 --- a/internal/runtime/persist/persist_test.go +++ b/internal/runtime/persist/persist_test.go @@ -360,3 +360,67 @@ func TestStore_Save_MissingID_异常(t *testing.T) { t.Fatalf("Save(blank id) err = %v, want wrap of ErrMissingID", err) } } + +// TestStore_LoadLatest_变宽时间戳_取较晚 is the regression for the variable-width +// timestamp bug on the durable-checkpoint recovery path. time.RFC3339Nano drops +// trailing fractional zeros, so the older checkpoint was once stored as +// ":05.83Z" and the later one as ":05.8309789Z". SQLite compares updated_at as +// plain text, and "…83Z" ranks above "…8309789Z", so LoadLatest used to recover +// the older checkpoint — a silent wrong resume. The older checkpoint also +// carries the larger step, so the step DESC tiebreak cannot mask the bad order. +func TestStore_LoadLatest_变宽时间戳_取较晚(t *testing.T) { + t.Parallel() + ctx := context.Background() + s := newTestStore(t) + + const ( + olderRaw = "2026-09-07T03:20:05.83Z" // X.8300000 under RFC3339Nano + newerRaw = "2026-09-07T03:20:05.8309789Z" // X.8309789 under RFC3339Nano + ) + if _, err := s.db.ExecContext(ctx, insertLegacyCheckpointSQL, + "cp-older", "agent-a", 9, `{"step":9}`, olderRaw, olderRaw); err != nil { + t.Fatalf("insert older checkpoint: %v", err) + } + if _, err := s.db.ExecContext(ctx, insertLegacyCheckpointSQL, + "cp-newer", "agent-a", 2, `{"step":2}`, newerRaw, newerRaw); err != nil { + t.Fatalf("insert newer checkpoint: %v", err) + } + + // Re-open so the fixed-width normalization repairs the legacy rows. + if _, err := New(ctx, s.db, testLogger()); err != nil { + t.Fatalf("New (normalize): %v", err) + } + + got, err := s.LoadLatest(ctx, "agent-a") + if err != nil { + t.Fatalf("LoadLatest error: %v", err) + } + if got.ID != "cp-newer" { + t.Fatalf("LoadLatest = %q (step %d), want cp-newer (the later updated_at)", got.ID, got.Step) + } + if want := time.Date(2026, 9, 7, 3, 20, 5, 830978900, time.UTC); !got.UpdatedAt.Equal(want) { + t.Fatalf("LoadLatest updated_at = %v, want %v", got.UpdatedAt, want) + } +} + +// TestStore_Save_定宽时间戳 pins that every write uses the fixed-width layout, so +// text ordering equals time ordering for rows created by the current code. +func TestStore_Save_定宽时间戳(t *testing.T) { + t.Parallel() + ctx := context.Background() + s := newTestStore(t) + + if err := s.Save(ctx, &Checkpoint{ID: "cp-1", AgentID: "agent-a", Step: 1, State: []byte(`{}`)}); err != nil { + t.Fatalf("Save error: %v", err) + } + var created, updated string + if err := s.db.QueryRowContext(ctx, + `SELECT created_at, updated_at FROM checkpoints WHERE id = ?`, "cp-1"). + Scan(&created, &updated); err != nil { + t.Fatalf("read stored timestamps: %v", err) + } + if len(created) != storedTimeLen || len(updated) != storedTimeLen { + t.Errorf("stored timestamps = %q/%q (%d/%d bytes), want fixed width %d", + created, updated, len(created), len(updated), storedTimeLen) + } +} diff --git a/internal/runtime/persist/timefmt.go b/internal/runtime/persist/timefmt.go new file mode 100644 index 0000000..3851731 --- /dev/null +++ b/internal/runtime/persist/timefmt.go @@ -0,0 +1,126 @@ +package persist + +import ( + "context" + "database/sql" + "fmt" + "time" +) + +// timeLayout is the fixed-width timestamp layout used for every stored +// created_at/updated_at value. Unlike time.RFC3339Nano it never drops trailing +// fractional zeros, so every value is exactly storedTimeLen bytes wide. SQLite +// compares these TEXT columns as plain strings, and equal width is what makes +// that comparison agree with chronological order. UTC values render the zone as +// a single "Z". +const timeLayout = "2006-01-02T15:04:05.000000000Z07:00" + +// storedTimeLen is the byte length of a UTC value formatted with timeLayout: +// "2006-01-02T15:04:05.000000000Z" — the Z07:00 zone collapses to "Z" in UTC. +const storedTimeLen = 30 + +// checkpointsTable names the table whose timestamps are normalized when the +// Store opens. It must match the DDL in persist.go. +const checkpointsTable = "checkpoints" + +// formatTime encodes t in UTC with the fixed-width timeLayout. Always writing +// UTC keeps ordering and round-trips stable across processes, and the fixed +// width is what lets ORDER BY updated_at order these TEXT columns by time. +func formatTime(t time.Time) string { + return t.UTC().Format(timeLayout) +} + +// parseTime decodes a stored timestamp. time.RFC3339Nano accepts both the +// fixed-width values this package writes and the legacy variable-width ones, so +// the same parser serves reads and the open-time normalization. +func parseTime(s string) (time.Time, error) { + return time.Parse(time.RFC3339Nano, s) +} + +// normalizeTimestamps rewrites every checkpoints timestamp that is not +// storedTimeLen bytes wide into the fixed-width timeLayout. Rows written before +// the layout existed carry variable-width RFC3339Nano text whose string order is +// not its time order; this brings them in line so LoadLatest picks the truly +// latest checkpoint. +// +// It is idempotent: only rows whose column length differs from storedTimeLen are +// selected, so running it again on a normalized table is a no-op. A row whose +// value does not parse is left exactly as it is — one bad legacy value must not +// keep the Store from opening. +func normalizeTimestamps(ctx context.Context, db *sql.DB) error { + if err := normalizeTimeColumn(ctx, db, "created_at"); err != nil { + return err + } + return normalizeTimeColumn(ctx, db, "updated_at") +} + +// normalizeTimeColumn rewrites every checkpoints. value that is not +// storedTimeLen bytes wide. col is a compile-time constant, never user input, so +// splicing it into the statement is safe; the length comparison and every value +// written are parameterized. All rewrites are collected before any UPDATE runs, +// so no write happens while the SELECT cursor is still open. +func normalizeTimeColumn(ctx context.Context, db *sql.DB, col string) error { + fixes, err := collectLegacyTimestamps(ctx, db, col) + if err != nil { + return err + } + for _, f := range fixes { + if _, err := db.ExecContext(ctx, + "UPDATE "+checkpointsTable+" SET "+f.col+" = ? WHERE id = ?", f.value, f.id); err != nil { + return fmt.Errorf("persist: normalize %s.%s: %w", checkpointsTable, f.col, err) + } + } + return nil +} + +// timeFix is one pending timestamp rewrite: which column, which row (by primary +// key id), and the fixed-width value to write. +type timeFix struct { + col string + id string + value string +} + +// collectLegacyTimestamps reads the rows of checkpoints.col whose value is not +// storedTimeLen bytes wide and returns the parseable ones as pending rewrites. +// Unparseable values are skipped, so normalization can never fail a row: the only +// errors are genuine read failures. +func collectLegacyTimestamps(ctx context.Context, db *sql.DB, col string) ([]timeFix, error) { + rows, err := db.QueryContext(ctx, + "SELECT id, "+col+" FROM "+checkpointsTable+" WHERE length("+col+") != ?", storedTimeLen) + if err != nil { + return nil, fmt.Errorf("persist: query legacy %s.%s: %w", checkpointsTable, col, err) + } + defer func() { _ = rows.Close() }() + + fixes := make([]timeFix, 0) + for rows.Next() { + var id, raw string + if err := rows.Scan(&id, &raw); err != nil { + return nil, fmt.Errorf("persist: scan legacy %s.%s: %w", checkpointsTable, col, err) + } + fixed, ok := reformatTime(raw) + if !ok { + continue + } + fixes = append(fixes, timeFix{col: col, id: id, value: fixed}) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("persist: iterate legacy %s.%s: %w", checkpointsTable, col, err) + } + return fixes, nil +} + +// reformatTime re-encodes raw with timeLayout. The bool is false when raw is +// already storedTimeLen wide or when it does not parse, in which case the caller +// leaves the stored value untouched. +func reformatTime(raw string) (string, bool) { + if len(raw) == storedTimeLen { + return "", false + } + t, err := parseTime(raw) + if err != nil || t.IsZero() { + return "", false + } + return formatTime(t), true +} diff --git a/internal/runtime/persist/timefmt_test.go b/internal/runtime/persist/timefmt_test.go new file mode 100644 index 0000000..004962d --- /dev/null +++ b/internal/runtime/persist/timefmt_test.go @@ -0,0 +1,118 @@ +package persist + +import ( + "context" + "testing" + "time" +) + +// insertLegacyCheckpointSQL writes a checkpoints row with caller-supplied raw +// timestamps, standing in for a row persisted before the fixed-width layout. +const insertLegacyCheckpointSQL = `INSERT INTO checkpoints + (id, agent_id, step, state, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?)` + +func TestFormatTime_定宽_正常(t *testing.T) { + t.Parallel() + tests := []struct { + name string + at time.Time + want string + }{ + {"零小数秒", time.Date(2026, 9, 7, 3, 20, 5, 0, time.UTC), "2026-09-07T03:20:05.000000000Z"}, + {"毫秒", time.Date(2026, 9, 7, 3, 20, 5, 830000000, time.UTC), "2026-09-07T03:20:05.830000000Z"}, + {"纳秒", time.Date(2026, 9, 7, 3, 20, 5, 830978900, time.UTC), "2026-09-07T03:20:05.830978900Z"}, + } + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + got := formatTime(tt.at) + if got != tt.want { + t.Errorf("formatTime = %q, want %q", got, tt.want) + } + if len(got) != storedTimeLen { + t.Errorf("formatTime length = %d, want %d", len(got), storedTimeLen) + } + }) + } +} + +func TestFormatTime_非UTC_正常(t *testing.T) { + t.Parallel() + zone := time.FixedZone("UTC+8", 8*60*60) + // 11:20:05.830+08:00 is 03:20:05.830Z, so the stored text is UTC. + at := time.Date(2026, 9, 7, 11, 20, 5, 830000000, zone) + if got, want := formatTime(at), "2026-09-07T03:20:05.830000000Z"; got != want { + t.Errorf("formatTime = %q, want %q", got, want) + } +} + +// TestNew_归一化历史时间戳_正常 writes a legacy variable-width row directly, +// re-opens the Store to trigger normalization, and checks both the repaired text +// and its idempotence on a further open. +func TestNew_归一化历史时间戳_正常(t *testing.T) { + t.Parallel() + ctx := context.Background() + s := newTestStore(t) + + // RFC3339Nano drops the trailing fractional zeros, so this legacy value is + // only 23 bytes wide and loses the text-ordering guarantee. + const legacy = "2026-09-07T03:20:05.83Z" + if _, err := s.db.ExecContext(ctx, insertLegacyCheckpointSQL, + "legacy", "agent-a", 1, `{}`, legacy, legacy); err != nil { + t.Fatalf("insert legacy row: %v", err) + } + + if _, err := New(ctx, s.db, testLogger()); err != nil { + t.Fatalf("New (normalize): %v", err) + } + var created, updated string + if err := s.db.QueryRowContext(ctx, + `SELECT created_at, updated_at FROM checkpoints WHERE id = ?`, "legacy"). + Scan(&created, &updated); err != nil { + t.Fatalf("read normalized timestamps: %v", err) + } + const want = "2026-09-07T03:20:05.830000000Z" + if created != want || updated != want { + t.Errorf("normalized timestamps = %q/%q, want %q", created, updated, want) + } + + // A further open must be a no-op: the length condition selects nothing once + // the row is fixed width. + if _, err := New(ctx, s.db, testLogger()); err != nil { + t.Fatalf("New (second pass): %v", err) + } + var again string + if err := s.db.QueryRowContext(ctx, + `SELECT created_at FROM checkpoints WHERE id = ?`, "legacy").Scan(&again); err != nil { + t.Fatalf("read created_at after second open: %v", err) + } + if again != created { + t.Errorf("second open changed created_at: %q -> %q", created, again) + } +} + +// TestNew_无法解析的时间戳_跳过 pins the "leave it alone" rule: one bad legacy +// value must not keep the Store from opening, and must not be rewritten. +func TestNew_无法解析的时间戳_跳过(t *testing.T) { + t.Parallel() + ctx := context.Background() + s := newTestStore(t) + + if _, err := s.db.ExecContext(ctx, insertLegacyCheckpointSQL, + "broken", "agent-a", 1, `{}`, "not-a-time", "not-a-time"); err != nil { + t.Fatalf("insert unparseable row: %v", err) + } + + if _, err := New(ctx, s.db, testLogger()); err != nil { + t.Fatalf("New with unparseable timestamp: %v", err) + } + var raw string + if err := s.db.QueryRowContext(ctx, + `SELECT created_at FROM checkpoints WHERE id = ?`, "broken").Scan(&raw); err != nil { + t.Fatalf("read created_at: %v", err) + } + if raw != "not-a-time" { + t.Errorf("created_at = %q, want the unparseable value left as is", raw) + } +}