fix(postgres): normalize message timestamps for push

Author: wesm

internal/db/messages.go

@@ -1220,6 +1220,20 @@ func (db *DB) MessageContentHashFingerprint(sessionID string) (string, error) {
 // (selectMessageCols coalesces the same way); without it a single
 // imported NULL row would error here and abort the whole parse-diff run.
 func (db *DB) MessageRoleTimeFingerprint(sessionID string) (string, error) {
+	return db.MessageRoleTimeFingerprintWithTimestampNormalizer(
+		sessionID, nil,
+	)
+}
+
+// MessageRoleTimeFingerprintWithTimestampNormalizer returns the same
+// fingerprint as MessageRoleTimeFingerprint after applying normalizeTimestamp
+// to each timestamp value. It lets callers compare against stores that preserve
+// a different timestamp representation while keeping the query and field
+// ordering identical to the raw parse-diff fingerprint.
+func (db *DB) MessageRoleTimeFingerprintWithTimestampNormalizer(
+	sessionID string,
+	normalizeTimestamp func(string) string,
+) (string, error) {
 	rows, err := db.getReader().Query(
 		`SELECT ordinal, role, COALESCE(timestamp, '')
 		 FROM messages
@@ -1240,6 +1254,9 @@ func (db *DB) MessageRoleTimeFingerprint(sessionID string) (string, error) {
 			return "", err
 		}
 		role = SanitizeUTF8(role)
+		if normalizeTimestamp != nil {
+			timestamp = normalizeTimestamp(timestamp)
+		}
 		fmt.Fprintf(&b, "%d|%d:%s|%d:%s;",
 			ordinal, len(role), role, len(timestamp), timestamp)
 	}

internal/postgres/push.go

@@ -1135,8 +1135,8 @@ func (s *Sync) pushMessages(
 				err,
 			)
 		}
-		localRoleTimeFP, err := s.local.MessageRoleTimeFingerprint(
-			sessionID,
+		localRoleTimeFP, err := localMessageRoleTimePGFingerprint(
+			s.local, sessionID,
 		)
 		if err != nil {
 			return 0, fmt.Errorf(
@@ -1601,6 +1601,23 @@ func pgMessageContentHashFingerprint(
 	return b.String(), rows.Err()
 }
 
+func localMessageRoleTimePGFingerprint(
+	local *db.DB, sessionID string,
+) (string, error) {
+	return local.MessageRoleTimeFingerprintWithTimestampNormalizer(
+		sessionID,
+		pgPushTimestampFingerprintText,
+	)
+}
+
+func pgPushTimestampFingerprintText(value string) string {
+	t, ok := ParseSQLiteTimestamp(value)
+	if !ok {
+		return ""
+	}
+	return FormatISO8601(t.Truncate(time.Microsecond))
+}
+
 func pgMessageRoleTimeFingerprint(
 	ctx context.Context, tx *sql.Tx, sessionID string,
 ) (string, error) {

internal/postgres/push_pgtest_test.go

@@ -262,6 +262,68 @@ func TestPushMessageFlagsRewriteRegression(t *testing.T) {
 		"private chain of thought")
 }
 
+func TestPushMessageNanosecondTimestampNoRewriteRegression(t *testing.T) {
+	pgURL := testPGURL(t)
+
+	const schema = "agentsview_push_msgtime_nanos_test"
+	pg, err := Open(pgURL, schema, true)
+	require.NoError(t, err, "Open")
+	defer pg.Close()
+
+	ctx := context.Background()
+	_, err = pg.Exec(`DROP SCHEMA IF EXISTS ` + schema + ` CASCADE`)
+	require.NoError(t, err, "drop schema")
+	require.NoError(t, EnsureSchema(ctx, pg, schema), "EnsureSchema")
+
+	localDB, err := db.Open(filepath.Join(t.TempDir(), "local.db"))
+	require.NoError(t, err, "db.Open")
+	defer localDB.Close()
+
+	sync := &Sync{
+		pg:         pg,
+		local:      localDB,
+		machine:    "test-machine",
+		schema:     schema,
+		schemaDone: true,
+	}
+
+	const sessID = "message-nanotime-rewrite-001"
+	sess := db.Session{
+		ID:               sessID,
+		Project:          "test-proj",
+		Machine:          "test-machine",
+		Agent:            "shelley",
+		MessageCount:     1,
+		UserMessageCount: 1,
+		CreatedAt:        "2026-01-01T00:00:00Z",
+	}
+	require.NoError(t, localDB.UpsertSession(sess), "UpsertSession")
+	require.NoError(t, localDB.InsertMessages([]db.Message{{
+		SessionID:     sessID,
+		Ordinal:       1,
+		Role:          "user",
+		Content:       "question",
+		ContentLength: len("question"),
+		Timestamp:     "2026-01-01T00:00:00.123456789Z",
+	}}), "InsertMessages")
+
+	_, err = sync.Push(ctx, false, nil)
+	require.NoError(t, err, "Push first timestamp")
+	assertPGMessageTimestamp(t, pg, sessID, 1,
+		"2026-01-01T00:00:00.123456Z")
+
+	ctidBefore := pgMessageCTID(t, pg, sessID, 1)
+	require.NoError(t, localDB.SetSyncState("last_push_at", ""),
+		"clearing last_push_at")
+	require.NoError(t, localDB.SetSyncState(lastPushBoundaryStateKey, ""),
+		"clearing boundary state")
+
+	_, err = sync.Push(ctx, false, nil)
+	require.NoError(t, err, "Push same timestamp")
+	assert.Equal(t, ctidBefore, pgMessageCTID(t, pg, sessID, 1),
+		"microsecond-equivalent timestamps should hit the fast path")
+}
+
 // TestPushSessionTerminationStatus verifies that pushSession round-trips
 // the termination_status column to PG: a non-nil value writes the string,
 // and a subsequent push with nil clears the column back to NULL via the
@@ -547,6 +609,40 @@ func assertPGMessageThinking(
 	assert.Equal(t, wantThinkingText, gotThinkingText)
 }
 
+func assertPGMessageTimestamp(
+	t *testing.T,
+	pg *sql.DB,
+	sessionID string,
+	ordinal int,
+	want string,
+) {
+	t.Helper()
+	var gotTime sql.NullTime
+	require.NoError(t, pg.QueryRow(
+		`SELECT timestamp FROM messages
+		  WHERE session_id = $1 AND ordinal = $2`,
+		sessionID, ordinal,
+	).Scan(&gotTime), "read pg message timestamp")
+	require.True(t, gotTime.Valid, "timestamp should be non-NULL")
+	assert.Equal(t, want, FormatISO8601(gotTime.Time))
+}
+
+func pgMessageCTID(
+	t *testing.T,
+	pg *sql.DB,
+	sessionID string,
+	ordinal int,
+) string {
+	t.Helper()
+	var ctid string
+	require.NoError(t, pg.QueryRow(
+		`SELECT ctid::text FROM messages
+		  WHERE session_id = $1 AND ordinal = $2`,
+		sessionID, ordinal,
+	).Scan(&ctid), "read pg message ctid")
+	return ctid
+}
+
 // TestPushMessagesSanitizesNULBytes verifies that a message whose
 // model and source fields carry NUL bytes (observed in production:
 // the Antigravity gen_metadata heuristic persisted a raw protobuf

internal/postgres/push_test.go

@@ -2,6 +2,7 @@ package postgres
 
 import (
 	"encoding/json"
+	"path/filepath"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -307,6 +308,42 @@ func TestSessionPushFingerprintNoFieldCollisions(
 		"length-prefixed fingerprints should not collide")
 }
 
+func TestLocalMessageRoleTimePGFingerprintNormalizesNanoseconds(
+	t *testing.T,
+) {
+	localDB, err := db.Open(filepath.Join(t.TempDir(), "local.db"))
+	require.NoError(t, err, "db.Open")
+	defer localDB.Close()
+
+	const sessID = "pg-role-time-nanos"
+	require.NoError(t, localDB.UpsertSession(db.Session{
+		ID:        sessID,
+		Project:   "proj",
+		Machine:   "host",
+		Agent:     "shelley",
+		CreatedAt: "2026-03-11T12:34:56Z",
+	}), "UpsertSession")
+	require.NoError(t, localDB.InsertMessages([]db.Message{{
+		SessionID:     sessID,
+		Ordinal:       1,
+		Role:          "assistant",
+		Content:       "answer",
+		ContentLength: len("answer"),
+		Timestamp:     "2026-03-11T12:34:56.123456789Z",
+	}}), "InsertMessages")
+
+	got, err := localMessageRoleTimePGFingerprint(localDB, sessID)
+	require.NoError(t, err)
+	assert.Equal(t,
+		"1|9:assistant|27:2026-03-11T12:34:56.123456Z;",
+		got)
+
+	raw, err := localDB.MessageRoleTimeFingerprint(sessID)
+	require.NoError(t, err)
+	assert.NotEqual(t, raw, got,
+		"PG push fingerprint must not use raw nanosecond text")
+}
+
 func TestFinalizePushStatePersistsEmptyBoundary(
 	t *testing.T,
 ) {