fix(sessionwatch): detect file hash rewrites

Author: wesm

internal/db/sessions.go

@@ -1323,20 +1323,49 @@ func (db *DB) GetSessionMessageCount(
 	return count, true
 }
 
-// GetSessionVersion returns the message count and file mtime
-// for change detection in SSE watchers.
+// SessionVersionMarker returns a compact marker for one or more
+// version fields. Inputs are length-framed so adjacent fields cannot
+// collide by concatenation.
+func SessionVersionMarker(parts ...string) int64 {
+	const (
+		offset64 = uint64(14695981039346656037)
+		prime64  = uint64(1099511628211)
+	)
+	h := offset64
+	write := func(s string) {
+		for _, b := range []byte(s) {
+			h ^= uint64(b)
+			h *= prime64
+		}
+	}
+	for _, part := range parts {
+		write(fmt.Sprintf("%d:", len(part)))
+		write(part)
+	}
+	return int64(h)
+}
+
+// GetSessionVersion returns the message count and a compact version
+// marker for change detection in SSE watchers.
 func (db *DB) GetSessionVersion(
 	id string,
-) (count int, fileMtime int64, ok bool) {
+) (count int, version int64, ok bool) {
+	var fileMtime int64
+	var fileHash, localModifiedAt string
 	err := db.getReader().QueryRow(
-		"SELECT message_count, COALESCE(file_mtime, 0)"+
+		"SELECT message_count, COALESCE(file_mtime, 0),"+
+			" COALESCE(file_hash, ''), COALESCE(local_modified_at, '')"+
 			" FROM sessions WHERE id = ?",
 		id,
-	).Scan(&count, &fileMtime)
+	).Scan(&count, &fileMtime, &fileHash, &localModifiedAt)
 	if err != nil {
 		return 0, 0, false
 	}
-	return count, fileMtime, true
+	return count, SessionVersionMarker(
+		fmt.Sprintf("%d", fileMtime),
+		fileHash,
+		localModifiedAt,
+	), true
 }
 
 // IncrementalInfo holds the data needed for incremental

internal/db/store.go

@@ -45,7 +45,7 @@ type Store interface {
 	SecretFindingSource(ctx context.Context, f SecretFinding) (string, bool, error)
 
 	// SSE change detection.
-	GetSessionVersion(id string) (count int, fileMtime int64, ok bool)
+	GetSessionVersion(id string) (count int, version int64, ok bool)
 
 	// Metadata.
 	GetStats(ctx context.Context, excludeOneShot, excludeAutomated bool) (Stats, error)

internal/duckdb/store.go

@@ -399,21 +399,31 @@ func (s *Store) GetChildSessions(ctx context.Context, parentID string) ([]db.Ses
 
 func (s *Store) GetSessionVersion(id string) (int, int64, bool) {
 	var count int
+	var fileMtime sql.NullInt64
+	var fileHash sql.NullString
 	var updated any
 	err := s.duck.QueryRow(
-		`SELECT message_count, COALESCE(local_modified_at, ended_at, started_at, created_at)
+		`SELECT message_count, file_mtime, file_hash,
+		        COALESCE(local_modified_at, ended_at, started_at, created_at)
 		 FROM sessions WHERE id = ?`,
 		id,
-	).Scan(&count, &updated)
+	).Scan(&count, &fileMtime, &fileHash, &updated)
 	if err != nil {
 		return 0, 0, false
 	}
-	formatted := formatDBTime(updated)
-	var h int64
-	for _, c := range formatted {
-		h = h*31 + int64(c)
+	fileMtimePart := ""
+	if fileMtime.Valid {
+		fileMtimePart = fmt.Sprintf("%d", fileMtime.Int64)
 	}
-	return count, h, true
+	fileHashPart := ""
+	if fileHash.Valid {
+		fileHashPart = fileHash.String
+	}
+	return count, db.SessionVersionMarker(
+		fileMtimePart,
+		fileHashPart,
+		formatDBTime(updated),
+	), true
 }
 
 func (s *Store) GetStats(ctx context.Context, excludeOneShot, excludeAutomated bool) (db.Stats, error) {

internal/postgres/store.go

@@ -54,8 +54,8 @@ func (s *Store) SetCursorSecret(secret []byte) {
 // and pins) is writable through dedicated methods.
 func (s *Store) ReadOnly() bool { return true }
 
-// GetSessionVersion returns the message count and a hash of
-// updated_at for SSE change detection.
+// GetSessionVersion returns the message count and a compact version
+// marker for SSE change detection.
 func (s *Store) GetSessionVersion(
 	id string,
 ) (int, int64, bool) {
@@ -69,12 +69,7 @@ func (s *Store) GetSessionVersion(
 	if err != nil {
 		return 0, 0, false
 	}
-	formatted := FormatISO8601(updatedAt)
-	var h int64
-	for _, c := range formatted {
-		h = h*31 + int64(c)
-	}
-	return count, h, true
+	return count, db.SessionVersionMarker(FormatISO8601(updatedAt)), true
 }
 
 // ------------------------------------------------------------

internal/sessionwatch/watcher.go

@@ -43,7 +43,7 @@ func New(d db.Store, engine *sync.Engine) *Watcher {
 }
 
 // Events polls the database for session changes and signals the
-// returned channel when the message count changes. This is
+// returned channel when the session version changes. This is
 // decoupled from file I/O — the file watcher handles syncing
 // files to the database, and this monitor detects the resulting
 // DB changes.
@@ -60,15 +60,15 @@ func (w *Watcher) Events(
 		defer close(ch)
 
 		// Seed initial state from the database.
-		lastCount, lastDBMtime, _ := w.db.GetSessionVersion(
+		lastCount, lastDBVersion, _ := w.db.GetSessionVersion(
 			sessionID,
 		)
 
 		if w.engine == nil {
 			// PG read mode: poll GetSessionVersion only,
 			// no file watching or fallback sync.
 			w.pollDBOnly(ctx, ch, sessionID,
-				lastCount, lastDBMtime)
+				lastCount, lastDBVersion)
 			return
 		}
 
@@ -91,7 +91,7 @@ func (w *Watcher) Events(
 				changed := w.checkDBForChanges(
 					sessionID,
 					&lastCount,
-					&lastDBMtime,
+					&lastDBVersion,
 					&sourcePath,
 					&lastFileMtime,
 					&fileMtimeChangedAt,
@@ -114,7 +114,7 @@ func (w *Watcher) Events(
 // no sync engine or file watcher.
 func (w *Watcher) pollDBOnly(
 	ctx context.Context, ch chan<- struct{},
-	sessionID string, lastCount int, lastDBMtime int64,
+	sessionID string, lastCount int, lastDBVersion int64,
 ) {
 	ticker := time.NewTicker(PollInterval)
 	defer ticker.Stop()
@@ -124,10 +124,10 @@ func (w *Watcher) pollDBOnly(
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			count, dbMtime, ok := w.db.GetSessionVersion(sessionID)
-			if ok && (count != lastCount || dbMtime != lastDBMtime) {
+			count, dbVersion, ok := w.db.GetSessionVersion(sessionID)
+			if ok && (count != lastCount || dbVersion != lastDBVersion) {
 				lastCount = count
-				lastDBMtime = dbMtime
+				lastDBVersion = dbVersion
 				select {
 				case ch <- struct{}{}:
 				case <-ctx.Done():
@@ -138,28 +138,25 @@ func (w *Watcher) pollDBOnly(
 	}
 }
 
-// checkDBForChanges polls the database for a session's
-// message_count and file_mtime. If either changed, it
-// returns true. As a fallback, it monitors source file
-// mtime and triggers a direct sync when the watcher
-// hasn't updated the DB.
+// checkDBForChanges polls the database for a session version change.
+// As a fallback, it monitors source file mtime and triggers a direct
+// sync when the watcher hasn't updated the DB.
 func (w *Watcher) checkDBForChanges(
 	sessionID string,
 	lastCount *int,
-	lastDBMtime *int64,
+	lastDBVersion *int64,
 	sourcePath *string,
 	lastFileMtime *int64,
 	fileMtimeChangedAt *time.Time,
 ) bool {
-	// Primary: check if the DB has new data (message count
-	// or file_mtime changed, covering both message appends
-	// and metadata-only updates like progress events).
-	if count, dbMtime, ok := w.db.GetSessionVersion(
+	// Primary: check if the DB has new data. The version marker covers
+	// message appends and metadata/content-only updates.
+	if count, dbVersion, ok := w.db.GetSessionVersion(
 		sessionID,
 	); ok && (count != *lastCount ||
-		dbMtime != *lastDBMtime) {
+		dbVersion != *lastDBVersion) {
 		*lastCount = count
-		*lastDBMtime = dbMtime
+		*lastDBVersion = dbVersion
 		// DB was updated; clear any pending fallback.
 		*fileMtimeChangedAt = time.Time{}
 		return true
@@ -208,12 +205,12 @@ func (w *Watcher) checkDBForChanges(
 			return false
 		}
 		// Re-check the DB after syncing.
-		if count, dbMtime, ok := w.db.GetSessionVersion(
+		if count, dbVersion, ok := w.db.GetSessionVersion(
 			sessionID,
 		); ok && (count != *lastCount ||
-			dbMtime != *lastDBMtime) {
+			dbVersion != *lastDBVersion) {
 			*lastCount = count
-			*lastDBMtime = dbMtime
+			*lastDBVersion = dbVersion
 			return true
 		}
 	}

internal/sessionwatch/watcher_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.kenn.io/agentsview/internal/db"
+	"go.kenn.io/agentsview/internal/dbtest"
 	"go.kenn.io/agentsview/internal/parser"
 	"go.kenn.io/agentsview/internal/sync"
 )
@@ -71,3 +72,44 @@ func TestCheckDBForChanges_FileDisappears(t *testing.T) {
 	assert.Empty(t, path)
 	assert.Equal(t, int64(0), lastMtime)
 }
+
+func TestCheckDBForChanges_FileHashChange(t *testing.T) {
+	t.Parallel()
+	w := testWatcher(t)
+	database, ok := w.db.(*db.DB)
+	require.True(t, ok, "test watcher should use SQLite DB")
+
+	const sessionID = "hash-change"
+	var mtime int64 = 12345
+	hash1 := "shelley-fingerprint-1"
+	dbtest.SeedSession(t, database, sessionID, "proj", func(s *db.Session) {
+		s.MessageCount = 2
+		s.FileMtime = &mtime
+		s.FileHash = &hash1
+	})
+
+	lastCount, lastDBMtime, ok := w.db.GetSessionVersion(sessionID)
+	require.True(t, ok, "initial session version")
+
+	hash2 := "shelley-fingerprint-2"
+	dbtest.SeedSession(t, database, sessionID, "proj", func(s *db.Session) {
+		s.MessageCount = 2
+		s.FileMtime = &mtime
+		s.FileHash = &hash2
+	})
+
+	sourcePath := ""
+	var lastFileMtime int64
+	var mchanged time.Time
+	changed := w.checkDBForChanges(
+		sessionID,
+		&lastCount,
+		&lastDBMtime,
+		&sourcePath,
+		&lastFileMtime,
+		&mchanged,
+	)
+
+	assert.True(t, changed,
+		"file_hash-only rewrites must refresh session watchers")
+}

internal/sync/shelley_integration_test.go

@@ -114,6 +114,14 @@ func newShelleyEngine(t *testing.T, dir string) (*sync.Engine, *db.DB) {
 	return engine, database
 }
 
+func sessionIDs(sessions []db.Session) []string {
+	ids := make([]string, 0, len(sessions))
+	for _, s := range sessions {
+		ids = append(ids, s.ID)
+	}
+	return ids
+}
+
 func mainConvoMsgs() []shelleyMsg {
 	return []shelleyMsg{
 		{1, "user",
@@ -357,10 +365,20 @@ func TestSyncShelleySameSecondInPlaceRewrite(t *testing.T) {
 	engine, database := newShelleyEngine(t, dir)
 	require.False(t, engine.SyncAll(context.Background(), nil).Aborted)
 	assertMessageContent(t, database, "shelley:cMAIN1", "q", "partial")
+	before, err := database.GetSessionFull(
+		context.Background(), "shelley:cMAIN1",
+	)
+	require.NoError(t, err, "GetSessionFull before rewrite")
+	require.NotNil(t, before, "session before rewrite")
+	require.NotNil(t, before.FileMtime, "file_mtime before rewrite")
+	require.NotNil(t, before.FileHash, "file_hash before rewrite")
+	require.NotNil(t, before.LocalModifiedAt,
+		"local_modified_at before rewrite")
 
 	// Rewrite the agent message in place. Crucially, updated_at is left
 	// untouched (same second) and sequence_id is unchanged, so only the
 	// content fingerprint differs.
+	time.Sleep(20 * time.Millisecond)
 	conn, err := sql.Open("sqlite3", dbPath)
 	require.NoError(t, err)
 	_, err = conn.Exec(
@@ -376,6 +394,32 @@ func TestSyncShelleySameSecondInPlaceRewrite(t *testing.T) {
 	assertMessageContent(
 		t, database, "shelley:cMAIN1", "q", "the full streamed answer",
 	)
+	after, err := database.GetSessionFull(
+		context.Background(), "shelley:cMAIN1",
+	)
+	require.NoError(t, err, "GetSessionFull after rewrite")
+	require.NotNil(t, after, "session after rewrite")
+	require.NotNil(t, after.FileMtime, "file_mtime after rewrite")
+	require.NotNil(t, after.FileHash, "file_hash after rewrite")
+	require.NotNil(t, after.LocalModifiedAt,
+		"local_modified_at after rewrite")
+	assert.Equal(t, *before.FileMtime, *after.FileMtime,
+		"same-second rewrite keeps the real Shelley updated_at timestamp")
+	assert.NotEqual(t, *before.FileHash, *after.FileHash,
+		"same-count same-second rewrite changes the content fingerprint")
+	assert.Greater(t, *after.LocalModifiedAt, *before.LocalModifiedAt,
+		"successful rewrite must bump local_modified_at for push windows")
+
+	candidates, err := database.ListSessionsModifiedBetween(
+		context.Background(),
+		*before.LocalModifiedAt,
+		time.Now().UTC().Add(time.Second).Format(time.RFC3339Nano),
+		nil,
+		nil,
+	)
+	require.NoError(t, err, "ListSessionsModifiedBetween after rewrite")
+	assert.Contains(t, sessionIDs(candidates), "shelley:cMAIN1",
+		"local_modified_at must select the rewritten session for pushes")
 }
 
 // TestSyncShelleyLengthPreservingRewrite is the case a byte-length signal