diff --git a/docs/internal/session-format-sources.md b/docs/internal/session-format-sources.md index 21eb12774..a6e437f03 100644 --- a/docs/internal/session-format-sources.md +++ b/docs/internal/session-format-sources.md @@ -180,6 +180,13 @@ Grok section and remove the explicit registry exception in the coverage test. `promptSource` (per user turn, e.g. `"typed"`, `"queued"`, `"system"`, `"sdk"`). Neither key is documented upstream or covered by the codeburn notes; the evidence remains local observation under `no-public-source`. + Reverified 2026-07-31 against the transcript shape reported in + [#1265](https://github.com/kenn-io/agentsview/pull/1265): the extension can + also prepend one `ide_opened_file`/`ide_selection` wrapper directly onto a + real operator prompt inside a single `user` record (envelope first, prompt + text after the closing tag, one shared `uuid` for the whole record); the + parser splits these into a hidden system-metadata message plus the visible + prompt. Reverified 2026-08-09 against controlled `--resume --fork-session` reproductions and inspection of the Claude Code 2.1.226 bundle ([#1370](https://github.com/kenn-io/agentsview/issues/1370)): the background diff --git a/internal/db/db.go b/internal/db/db.go index 3df3e49ce..fda700ad8 100644 --- a/internal/db/db.go +++ b/internal/db/db.go @@ -414,7 +414,12 @@ CREATE INDEX IF NOT EXISTS idx_provider_freshness_updated_at // (87: Codex fork replay boundary correction. Turn identifiers are opaque; // existing Codex-format rows need re-parsing so copied parent turns with any // identifier shape remain excluded until the first child-owned turn.) -const dataVersion = 87 +// (88: Claude Code IDE context wrappers prepended onto a real prompt in +// the same entry are now split into a hidden system-metadata message plus +// the real prompt, instead of leaving the raw wrapper in first_message and +// the visible transcript. Existing rows need re-parsing so first_message +// and message content drop the leading markup.) +const dataVersion = 88 const tokenCoverageRepairStatsKey = "token_coverage_repair_v1" diff --git a/internal/db/db_test.go b/internal/db/db_test.go index 22adfd95c..15f1ed1f2 100644 --- a/internal/db/db_test.go +++ b/internal/db/db_test.go @@ -1008,9 +1008,10 @@ func TestMigration_ToolResultEventsTable(t *testing.T) { "expected tool_result_events table after reopen") } -func TestCurrentDataVersionUsageReparses(t *testing.T) { - assert.Equal(t, 87, CurrentDataVersion(), - "Codex replay accounting, VS Code Copilot response items, and opaque Codex fork boundaries require sequential reparses") +func TestCurrentDataVersionClaudeIDEEnvelopeSplit(t *testing.T) { + assert.Equal(t, 88, CurrentDataVersion(), + "version 88 splits Claude IDE envelopes off mixed prompts after "+ + "the Codex fork replay boundary reparse") } func TestInsertMessages_PreservesToolResultEvents(t *testing.T) { @@ -1721,11 +1722,15 @@ func TestReplaceSessionMessagesPreservesPins(t *testing.T) { ctx := context.Background() insertSession(t, d, "s1", "p") - insertMessages(t, d, + oldMessages := []Message{ userMsg("s1", 0, "msg0"), asstMsg("s1", 1, "msg1"), userMsg("s1", 2, "msg2"), - ) + } + for i := range oldMessages { + oldMessages[i].SourceUUID = fmt.Sprintf("uuid-%d", i) + } + insertMessages(t, d, oldMessages...) msgs, err := d.GetAllMessages(ctx, "s1") require.NoError(t, err, "GetAllMessages") @@ -1747,11 +1752,16 @@ func TestReplaceSessionMessagesPreservesPins(t *testing.T) { // Full replace (simulates a resync of an OpenCode or // explicitly re-synced session). - require.NoError(t, d.ReplaceSessionMessages("s1", []Message{ + newMessages := []Message{ userMsg("s1", 0, "msg0-updated"), asstMsg("s1", 1, "msg1-updated"), userMsg("s1", 2, "msg2-updated"), - }), "ReplaceSessionMessages") + } + for i := range newMessages { + newMessages[i].SourceUUID = fmt.Sprintf("uuid-%d", i) + } + require.NoError(t, d.ReplaceSessionMessages("s1", newMessages), + "ReplaceSessionMessages") newMsgs, err := d.GetAllMessages(ctx, "s1") require.NoError(t, err, "GetAllMessages after replace") @@ -1791,10 +1801,13 @@ func TestReplaceSessionMessagesDropsPinsForRemovedOrdinals(t *testing.T) { ctx := context.Background() insertSession(t, d, "s1", "p") - insertMessages(t, d, + oldMessages := []Message{ userMsg("s1", 0, "msg0"), asstMsg("s1", 1, "msg1"), - ) + } + oldMessages[0].SourceUUID = "uuid-0" + oldMessages[1].SourceUUID = "uuid-1" + insertMessages(t, d, oldMessages...) msgs, err := d.GetAllMessages(ctx, "s1") require.NoError(t, err, "GetAllMessages") @@ -1805,9 +1818,10 @@ func TestReplaceSessionMessagesDropsPinsForRemovedOrdinals(t *testing.T) { } // Replace with only ordinal-0 (ordinal-1 is gone). - require.NoError(t, d.ReplaceSessionMessages("s1", []Message{ - userMsg("s1", 0, "msg0-updated"), - }), "ReplaceSessionMessages") + replacement := userMsg("s1", 0, "msg0-updated") + replacement.SourceUUID = "uuid-0" + require.NoError(t, d.ReplaceSessionMessages("s1", []Message{replacement}), + "ReplaceSessionMessages") pins, err := d.ListPinnedMessages(ctx, "s1", "") require.NoError(t, err, "ListPinnedMessages") @@ -1889,6 +1903,7 @@ func TestReplaceSessionMessagesPinFallsBackToOrdinal(t *testing.T) { insertMessages(t, d, userMsg("s1", 0, "msg0"), asstMsg("s1", 1, "msg1"), + userMsg("s1", 2, "removed"), ) msgs, err := d.GetAllMessages(ctx, "s1") @@ -1896,10 +1911,11 @@ func TestReplaceSessionMessagesPinFallsBackToOrdinal(t *testing.T) { _, err = d.PinMessage("s1", msgs[1].ID, nil) require.NoError(t, err, "PinMessage") - // Replace with the same ordinals (and still no source_uuid). + // Truncation forces a full replacement. The pinned legacy row remains + // unchanged at its old ordinal, so the guarded fallback can restore it. require.NoError(t, d.ReplaceSessionMessages("s1", []Message{ - userMsg("s1", 0, "msg0-v2"), - asstMsg("s1", 1, "msg1-v2"), + userMsg("s1", 0, "msg0"), + asstMsg("s1", 1, "msg1"), }), "ReplaceSessionMessages") pins, err := d.ListPinnedMessages(ctx, "s1", "") @@ -1908,6 +1924,407 @@ func TestReplaceSessionMessagesPinFallsBackToOrdinal(t *testing.T) { assert.Equal(t, 1, pins[0].Ordinal, "pin ordinal") } +func TestReplaceSessionMessagesPinFallbackAllowsSourceUUIDEnrichment( + t *testing.T, +) { + d := testDB(t) + ctx := context.Background() + + insertSession(t, d, "s1", "p") + insertMessages(t, d, + userMsg("s1", 0, "msg0"), + asstMsg("s1", 1, "msg1"), + ) + msgs, err := d.GetAllMessages(ctx, "s1") + require.NoError(t, err, "GetAllMessages") + _, err = d.PinMessage("s1", msgs[1].ID, nil) + require.NoError(t, err, "PinMessage") + + enriched := asstMsg("s1", 1, "msg1") + enriched.SourceUUID = "new-provider-uuid" + require.NoError(t, d.ReplaceSessionMessages("s1", []Message{ + userMsg("s1", 0, "msg0"), + enriched, + }), "ReplaceSessionMessages") + + pins, err := d.ListPinnedMessages(ctx, "s1", "") + require.NoError(t, err, "ListPinnedMessages") + require.Len(t, pins, 1, "UUID enrichment must preserve the pin") + assert.Equal(t, 1, pins[0].Ordinal, "pin ordinal") +} + +func TestReplaceSessionContentDuplicateSourceUUIDRestoresOnePin(t *testing.T) { + d := testDB(t) + ctx := context.Background() + + insertSession(t, d, "s1", "p") + insertMessages(t, d, + Message{ + SessionID: "s1", Ordinal: 0, Role: "user", + Content: "pinned", SourceUUID: "duplicate", + }, + Message{ + SessionID: "s1", Ordinal: 1, Role: "assistant", + Content: "not pinned", SourceUUID: "duplicate", + }, + Message{ + SessionID: "s1", Ordinal: 2, Role: "user", + Content: "removed", SourceUUID: "tail", + }, + ) + msgs, err := d.GetAllMessages(ctx, "s1") + require.NoError(t, err, "GetAllMessages") + _, err = d.PinMessage("s1", msgs[0].ID, nil) + require.NoError(t, err, "PinMessage") + + require.NoError(t, d.ReplaceSessionContent("s1", []Message{ + { + SessionID: "s1", Ordinal: 0, Role: "user", + Content: "pinned", SourceUUID: "duplicate", + }, + { + SessionID: "s1", Ordinal: 1, Role: "assistant", + Content: "not pinned", SourceUUID: "duplicate", + }, + }, SessionSignalUpdate{}, nil), "ReplaceSessionContent") + + pins, err := d.ListPinnedMessages(ctx, "s1", "") + require.NoError(t, err, "ListPinnedMessages") + require.Len(t, pins, 1, "duplicate UUID must not duplicate the pin") + assert.Equal(t, 0, pins[0].Ordinal, "pin stays on its original message") +} + +func TestReplaceSessionContentSourceUUIDBecomesDuplicateRestoresOnePin( + t *testing.T, +) { + d := testDB(t) + ctx := context.Background() + + insertSession(t, d, "s1", "p") + insertMessages(t, d, + Message{ + SessionID: "s1", Ordinal: 0, Role: "user", + Content: "pinned", SourceUUID: "becomes-duplicate", + }, + Message{ + SessionID: "s1", Ordinal: 1, Role: "assistant", + Content: "old tail", SourceUUID: "old-tail", + }, + ) + msgs, err := d.GetAllMessages(ctx, "s1") + require.NoError(t, err, "GetAllMessages") + _, err = d.PinMessage("s1", msgs[0].ID, nil) + require.NoError(t, err, "PinMessage") + + require.NoError(t, d.ReplaceSessionContent("s1", []Message{ + { + SessionID: "s1", Ordinal: 0, Role: "user", + Content: "pinned", SourceUUID: "becomes-duplicate", + }, + { + SessionID: "s1", Ordinal: 1, Role: "assistant", + Content: "new duplicate", SourceUUID: "becomes-duplicate", + }, + }, SessionSignalUpdate{}, nil), "ReplaceSessionContent") + + pins, err := d.ListPinnedMessages(ctx, "s1", "") + require.NoError(t, err, "ListPinnedMessages") + require.Len(t, pins, 1, + "a newly duplicated UUID must use the guarded identity fallback") + assert.Equal(t, 0, pins[0].Ordinal, "pin stays on its original message") +} + +func TestReplaceSessionContentIdenticalDuplicatesKeepPin(t *testing.T) { + d := testDB(t) + ctx := context.Background() + + insertSession(t, d, "s1", "p") + insertMessages(t, d, + Message{ + SessionID: "s1", Ordinal: 0, Role: "user", + Content: "same", SourceUUID: "duplicate", + }, + Message{ + SessionID: "s1", Ordinal: 1, Role: "user", + Content: "same", SourceUUID: "duplicate", + }, + Message{ + SessionID: "s1", Ordinal: 2, Role: "assistant", + Content: "truncated tail", SourceUUID: "tail", + }, + ) + msgs, err := d.GetAllMessages(ctx, "s1") + require.NoError(t, err, "GetAllMessages") + require.Len(t, msgs, 3, "seeded messages") + _, err = d.PinMessage("s1", msgs[1].ID, nil) + require.NoError(t, err, "PinMessage") + + // Only the tail changes; the identical duplicates survive intact, + // so the pin must keep its saved ordinal instead of being dropped. + require.NoError(t, d.ReplaceSessionContent("s1", []Message{ + { + SessionID: "s1", Ordinal: 0, Role: "user", + Content: "same", SourceUUID: "duplicate", + }, + { + SessionID: "s1", Ordinal: 1, Role: "user", + Content: "same", SourceUUID: "duplicate", + }, + }, SessionSignalUpdate{}, nil), "ReplaceSessionContent") + + pins, err := d.ListPinnedMessages(ctx, "s1", "") + require.NoError(t, err, "ListPinnedMessages") + require.Len(t, pins, 1, + "unchanged identical duplicates must keep the pin") + assert.Equal(t, 1, pins[0].Ordinal, "pin stays at its saved ordinal") +} + +func TestReplaceSessionContentIdenticalDuplicateMultiplicityChangeDropsPin( + t *testing.T, +) { + d := testDB(t) + ctx := context.Background() + + insertSession(t, d, "s1", "p") + insertMessages(t, d, + Message{ + SessionID: "s1", Ordinal: 0, Role: "user", + Content: "same", SourceUUID: "duplicate", + }, + Message{ + SessionID: "s1", Ordinal: 1, Role: "user", + Content: "same", SourceUUID: "duplicate", + }, + Message{ + SessionID: "s1", Ordinal: 2, Role: "assistant", + Content: "replaced tail", SourceUUID: "tail", + }, + ) + msgs, err := d.GetAllMessages(ctx, "s1") + require.NoError(t, err, "GetAllMessages") + require.Len(t, msgs, 3, "seeded messages") + _, err = d.PinMessage("s1", msgs[1].ID, nil) + require.NoError(t, err, "PinMessage") + + // The tail's identity changes (forcing a full replacement) and a + // third identical duplicate takes its place: the saved ordinal can + // no longer prove which duplicate was pinned, so the pin is + // dropped. + require.NoError(t, d.ReplaceSessionContent("s1", []Message{ + { + SessionID: "s1", Ordinal: 0, Role: "user", + Content: "same", SourceUUID: "duplicate", + }, + { + SessionID: "s1", Ordinal: 1, Role: "user", + Content: "same", SourceUUID: "duplicate", + }, + { + SessionID: "s1", Ordinal: 2, Role: "user", + Content: "same", SourceUUID: "duplicate", + }, + }, SessionSignalUpdate{}, nil), "ReplaceSessionContent") + + pins, err := d.ListPinnedMessages(ctx, "s1", "") + require.NoError(t, err, "ListPinnedMessages") + assert.Empty(t, pins, + "changed duplicate multiplicity must drop the ambiguous pin") +} + +func TestReplaceSessionMessagesIdenticalDuplicatesFollowLeadingInsert( + t *testing.T, +) { + d := testDB(t) + ctx := context.Background() + + insertSession(t, d, "s1", "p") + insertMessages(t, d, + Message{ + SessionID: "s1", Ordinal: 0, Role: "user", + Content: "same", SourceUUID: "dup", + }, + Message{ + SessionID: "s1", Ordinal: 1, Role: "user", + Content: "same", SourceUUID: "dup", + }, + ) + msgs, err := d.GetAllMessages(ctx, "s1") + require.NoError(t, err, "GetAllMessages") + require.Len(t, msgs, 2, "seeded messages") + _, err = d.PinMessage("s1", msgs[1].ID, nil) + require.NoError(t, err, "PinMessage") + + // A hidden row inserted before the duplicates shifts both while + // their multiplicity stays equal: the pin must follow its + // occurrence rank, not stay on the saved ordinal where the first + // duplicate now sits. + require.NoError(t, d.ReplaceSessionMessages("s1", []Message{ + { + SessionID: "s1", Ordinal: 0, Role: "user", + Content: "context", SourceUUID: "env", IsSystem: true, + SourceType: "system", SourceSubtype: "ide_opened_file", + }, + { + SessionID: "s1", Ordinal: 1, Role: "user", + Content: "same", SourceUUID: "dup", + }, + { + SessionID: "s1", Ordinal: 2, Role: "user", + Content: "same", SourceUUID: "dup", + }, + }), "ReplaceSessionMessages") + + pins, err := d.ListPinnedMessages(ctx, "s1", "") + require.NoError(t, err, "ListPinnedMessages") + require.Len(t, pins, 1, "shifted duplicates must keep the pin") + assert.Equal(t, 2, pins[0].Ordinal, + "pin follows the second occurrence, not the saved ordinal") +} + +func TestReplaceSessionMessagesLegacyPinFollowsEqualMessageShift( + t *testing.T, +) { + d := testDB(t) + ctx := context.Background() + + insertSession(t, d, "s1", "p") + insertMessages(t, d, + Message{SessionID: "s1", Ordinal: 0, Role: "user", Content: "intro"}, + Message{SessionID: "s1", Ordinal: 1, Role: "user", Content: "x"}, + Message{SessionID: "s1", Ordinal: 2, Role: "user", Content: "x"}, + ) + msgs, err := d.GetAllMessages(ctx, "s1") + require.NoError(t, err, "GetAllMessages") + require.Len(t, msgs, 3, "seeded messages") + _, err = d.PinMessage("s1", msgs[2].ID, nil) + require.NoError(t, err, "PinMessage") + + // A hidden row inserted at the front shifts two equal visible + // messages. The pin on the second "x" must follow its occurrence + // rank to the shifted ordinal instead of re-attaching to the first + // "x" that now occupies the saved ordinal. + require.NoError(t, d.ReplaceSessionMessages("s1", []Message{ + { + SessionID: "s1", Ordinal: 0, Role: "user", + Content: "context", IsSystem: true, + SourceType: "system", SourceSubtype: "ide_opened_file", + }, + {SessionID: "s1", Ordinal: 1, Role: "user", Content: "intro"}, + {SessionID: "s1", Ordinal: 2, Role: "user", Content: "x"}, + {SessionID: "s1", Ordinal: 3, Role: "user", Content: "x"}, + }), "ReplaceSessionMessages") + + pins, err := d.ListPinnedMessages(ctx, "s1", "") + require.NoError(t, err, "ListPinnedMessages") + require.Len(t, pins, 1, "shifted equal messages must keep the pin") + assert.Equal(t, 3, pins[0].Ordinal, + "pin follows the second occurrence, not the saved ordinal") +} + +// TestWriteSessionBatchPreservesLegacyPinWhenMetadataBecomesHidden +// models re-uploading an unchanged transcript across the server change +// that started preserving IsSystem: the first upload stored every row +// with is_system = 0, the re-upload reclassifies the metadata row as +// hidden without moving anything, and the pin on the unchanged visible +// row must survive. +func TestWriteSessionBatchPreservesLegacyPinWhenMetadataBecomesHidden( + t *testing.T, +) { + d := testDB(t) + ctx := context.Background() + + base := Session{ + ID: "upload-1", + Project: "proj", + Machine: defaultMachine, + Agent: "claude", + FirstMessage: new("real question"), + StartedAt: new("2024-01-15T10:00:00Z"), + MessageCount: 2, + UserMessageCount: 1, + } + envelope := "f" + oldUpload := []Message{ + { + SessionID: "upload-1", Ordinal: 0, Role: "user", + Content: envelope, + }, + { + SessionID: "upload-1", Ordinal: 1, Role: "user", + Content: "real question", + }, + } + _, err := d.WriteSessionBatch([]SessionBatchWrite{{ + Session: base, + Messages: oldUpload, + DataVersion: CurrentDataVersion(), + ReplaceMessages: true, + }}) + require.NoError(t, err, "initial upload") + + msgs, err := d.GetAllMessages(ctx, "upload-1") + require.NoError(t, err, "GetAllMessages") + require.Len(t, msgs, 2, "uploaded messages") + _, err = d.PinMessage("upload-1", msgs[1].ID, nil) + require.NoError(t, err, "PinMessage") + + reupload := []Message{ + { + SessionID: "upload-1", Ordinal: 0, Role: "user", + Content: envelope, IsSystem: true, + SourceType: "system", SourceSubtype: "ide_opened_file", + }, + { + SessionID: "upload-1", Ordinal: 1, Role: "user", + Content: "real question", + }, + } + _, err = d.WriteSessionBatch([]SessionBatchWrite{{ + Session: base, + Messages: reupload, + DataVersion: CurrentDataVersion(), + ReplaceMessages: true, + }}) + require.NoError(t, err, "re-upload") + + pins, err := d.ListPinnedMessages(ctx, "upload-1", "") + require.NoError(t, err, "ListPinnedMessages") + require.Len(t, pins, 1, + "reclassified metadata must not drop the unchanged pin") + assert.Equal(t, 1, pins[0].Ordinal, "pin stays at its saved ordinal") +} + +func TestReplaceSessionContentMissingSourceUUIDDropsPin(t *testing.T) { + d := testDB(t) + ctx := context.Background() + + insertSession(t, d, "s1", "p") + insertMessages(t, d, + Message{ + SessionID: "s1", Ordinal: 0, Role: "user", + Content: "gone", SourceUUID: "gone-uuid", + }, + Message{ + SessionID: "s1", Ordinal: 1, Role: "assistant", + Content: "removed", SourceUUID: "tail", + }, + ) + msgs, err := d.GetAllMessages(ctx, "s1") + require.NoError(t, err, "GetAllMessages") + _, err = d.PinMessage("s1", msgs[0].ID, nil) + require.NoError(t, err, "PinMessage") + + require.NoError(t, d.ReplaceSessionContent("s1", []Message{{ + SessionID: "s1", Ordinal: 0, Role: "assistant", + Content: "unrelated", SourceUUID: "other-uuid", + }}, SessionSignalUpdate{}, nil), "ReplaceSessionContent") + + pins, err := d.ListPinnedMessages(ctx, "s1", "") + require.NoError(t, err, "ListPinnedMessages") + assert.Empty(t, pins, + "a vanished UUID must not fall back to an unrelated ordinal") +} + func TestGetSessionFilePath(t *testing.T) { d := testDB(t) @@ -4672,6 +5089,49 @@ func TestCopyOrphanedDataFrom(t *testing.T) { "expected 0 tool_calls for s2, got %d", tcCount) } +func TestCopyOrphanedDataFrom_DuplicateSourceUUIDKeepsOnePin(t *testing.T) { + dir := t.TempDir() + ctx := context.Background() + + srcPath := filepath.Join(dir, "old.db") + srcDB := testDBAtPath(t, srcPath, "src") + insertSession(t, srcDB, "orphan", "proj") + insertMessages(t, srcDB, + Message{ + SessionID: "orphan", Ordinal: 0, Role: "user", + Content: "pinned", ContentLength: 6, + SourceUUID: "duplicate", + }, + Message{ + SessionID: "orphan", Ordinal: 1, Role: "assistant", + Content: "not pinned", ContentLength: 10, + SourceUUID: "duplicate", + }, + ) + var pinnedMessageID int64 + require.NoError(t, srcDB.getReader().QueryRow(` + SELECT id FROM messages + WHERE session_id = 'orphan' AND ordinal = 0`, + ).Scan(&pinnedMessageID), "resolve pinned source message") + _, err := srcDB.PinMessage("orphan", pinnedMessageID, nil) + require.NoError(t, err, "pin source message") + require.NoError(t, srcDB.Close(), "close source database") + + dstPath := filepath.Join(dir, "new.db") + dstDB := testDBAtPath(t, dstPath, "dst") + defer dstDB.Close() + + copied, err := dstDB.CopyOrphanedDataFrom(srcPath) + require.NoError(t, err, "CopyOrphanedDataFrom") + require.Equal(t, 1, copied, "copied orphaned sessions") + + pins, err := dstDB.ListPinnedMessages(ctx, "orphan", "") + require.NoError(t, err, "ListPinnedMessages") + require.Len(t, pins, 1, + "a duplicated source UUID must not duplicate the source pin") + assert.Equal(t, 0, pins[0].Ordinal, "pin stays on its copied ordinal") +} + // TestCopyOrphanedDataFrom_SkipsStaleCodexForkRows covers the // dataVersion 40 upgrade path (#643): a pre-fix DB stored a forked // Codex rollout under the replayed parent's id with double-counted @@ -5783,6 +6243,438 @@ func TestCopySessionMetadataFrom(t *testing.T) { assert.Equal(t, 1, starCount, "stars after") } +func TestCopySessionMetadataFrom_IdenticalDuplicatePins(t *testing.T) { + dir := t.TempDir() + ctx := context.Background() + + identical := func(sessionID string, ordinals ...int) []Message { + msgs := make([]Message, 0, len(ordinals)) + for _, ordinal := range ordinals { + msgs = append(msgs, Message{ + SessionID: sessionID, Ordinal: ordinal, Role: "user", + Content: "same", ContentLength: 4, + SourceUUID: "duplicate", + }) + } + return msgs + } + + srcPath := filepath.Join(dir, "src.db") + srcDB := testDBAtPath(t, srcPath, "src") + // Unchanged duplicate set: the fresh DB has the same two + // identical rows, so the pin keeps its ordinal. + insertSession(t, srcDB, "dup-keep", "proj") + insertMessages(t, srcDB, identical("dup-keep", 0, 1)...) + // Changed duplicate set: the fresh DB gained a third identical + // row, so the old ordinal no longer proves which duplicate was + // pinned and the pin is dropped. + insertSession(t, srcDB, "dup-changed", "proj") + insertMessages(t, srcDB, identical("dup-changed", 0, 1)...) + // Shifted duplicate set: the fresh DB inserted a context row + // before the duplicates, so the pin must follow its occurrence + // rank to the shifted ordinal. + insertSession(t, srcDB, "dup-shifted", "proj") + insertMessages(t, srcDB, identical("dup-shifted", 0, 1)...) + for _, sessionID := range []string{ + "dup-keep", "dup-changed", "dup-shifted", + } { + var msgID int64 + require.NoError(t, srcDB.getReader().QueryRow( + "SELECT id FROM messages WHERE session_id = ? AND ordinal = 1", + sessionID, + ).Scan(&msgID), "resolve %s ordinal 1", sessionID) + pinID, err := srcDB.PinMessage(sessionID, msgID, nil) + require.NoError(t, err, "pin %s", sessionID) + require.NotZero(t, pinID, "pin %s not created", sessionID) + } + require.NoError(t, srcDB.Close(), "close source database") + + dstPath := filepath.Join(dir, "dst.db") + dstDB := testDBAtPath(t, dstPath, "dst") + defer dstDB.Close() + insertSession(t, dstDB, "dup-keep", "proj") + insertMessages(t, dstDB, identical("dup-keep", 0, 1)...) + insertSession(t, dstDB, "dup-changed", "proj") + insertMessages(t, dstDB, identical("dup-changed", 0, 1, 2)...) + insertSession(t, dstDB, "dup-shifted", "proj") + insertMessages(t, dstDB, append([]Message{{ + SessionID: "dup-shifted", Ordinal: 0, Role: "user", + Content: "context", ContentLength: 7, SourceUUID: "env", + }}, identical("dup-shifted", 1, 2)...)...) + + require.NoError(t, dstDB.CopySessionMetadataFrom(srcPath), + "CopySessionMetadataFrom") + + pins, err := dstDB.ListPinnedMessages(ctx, "dup-keep", "") + require.NoError(t, err, "ListPinnedMessages dup-keep") + require.Len(t, pins, 1, + "unchanged identical duplicates must keep the pin") + assert.Equal(t, 1, pins[0].Ordinal, "pin stays at its saved ordinal") + + pins, err = dstDB.ListPinnedMessages(ctx, "dup-changed", "") + require.NoError(t, err, "ListPinnedMessages dup-changed") + assert.Empty(t, pins, + "changed duplicate multiplicity must drop the ambiguous pin") + + pins, err = dstDB.ListPinnedMessages(ctx, "dup-shifted", "") + require.NoError(t, err, "ListPinnedMessages dup-shifted") + require.Len(t, pins, 1, + "shifted duplicates must keep the pin") + assert.Equal(t, 2, pins[0].Ordinal, + "pin follows the second occurrence, not the saved ordinal") +} + +// TestCopySessionMetadataFrom_LegacyPinFollowsShiftedReply models a +// full resync of a pre-uuid session across the IDE-envelope split: the +// re-parse inserts a hidden envelope row, shifting the unchanged +// pinned reply by one ordinal. The pin must follow its visible +// (role, content) occurrence rank to the shifted row instead of being +// dropped at the stale ordinal. +func TestCopySessionMetadataFrom_LegacyPinFollowsShiftedReply( + t *testing.T, +) { + dir := t.TempDir() + ctx := context.Background() + + srcPath := filepath.Join(dir, "src.db") + srcDB := testDBAtPath(t, srcPath, "src") + insertSession(t, srcDB, "s1", "proj") + insertMessages(t, srcDB, + Message{ + SessionID: "s1", Ordinal: 1, Role: "user", + Content: "f explain", + ContentLength: 44, + }, + Message{ + SessionID: "s1", Ordinal: 2, Role: "assistant", + Content: "Legacy reply", ContentLength: 12, + }, + ) + var msgID int64 + require.NoError(t, srcDB.getReader().QueryRow( + "SELECT id FROM messages WHERE session_id = 's1' AND ordinal = 2", + ).Scan(&msgID), "resolve pinned reply") + pinID, err := srcDB.PinMessage("s1", msgID, nil) + require.NoError(t, err, "pin legacy reply") + require.NotZero(t, pinID, "pin not created") + require.NoError(t, srcDB.Close(), "close source database") + + // Fresh DB: the re-parse split the combined prompt and stamped + // provider uuids, shifting the unchanged reply to ordinal 3. + dstPath := filepath.Join(dir, "dst.db") + dstDB := testDBAtPath(t, dstPath, "dst") + defer dstDB.Close() + insertSession(t, dstDB, "s1", "proj") + insertMessages(t, dstDB, + Message{ + SessionID: "s1", Ordinal: 1, Role: "user", + Content: "f", + ContentLength: 36, IsSystem: true, + SourceType: "system", SourceSubtype: "ide_opened_file", + SourceUUID: "u1:ide-context", + }, + Message{ + SessionID: "s1", Ordinal: 2, Role: "user", + Content: "explain", ContentLength: 7, SourceUUID: "u1", + }, + Message{ + SessionID: "s1", Ordinal: 3, Role: "assistant", + Content: "Legacy reply", ContentLength: 12, SourceUUID: "u2", + }, + ) + + require.NoError(t, dstDB.CopySessionMetadataFrom(srcPath), + "CopySessionMetadataFrom") + + pins, err := dstDB.ListPinnedMessages(ctx, "s1", "") + require.NoError(t, err, "ListPinnedMessages") + require.Len(t, pins, 1, + "shifted legacy reply must keep the pin") + assert.Equal(t, 3, pins[0].Ordinal, + "pin follows the reply to its shifted ordinal") +} + +func TestCopySessionMetadataFrom_PinsFollowSourceUUID(t *testing.T) { + dir := t.TempDir() + ctx := context.Background() + + // Source DB: pre-reparse shape where one entry held the IDE + // envelope and the prompt combined at ordinal 1. + srcPath := filepath.Join(dir, "src.db") + srcDB := testDBAtPath(t, srcPath, "src") + insertSession(t, srcDB, "s1", "proj") + insertMessages(t, srcDB, + Message{ + SessionID: "s1", Ordinal: 1, Role: "user", + Content: "f explain", + ContentLength: 44, SourceUUID: "u1", + }, + Message{ + SessionID: "s1", Ordinal: 2, Role: "assistant", + Content: "sure", ContentLength: 4, SourceUUID: "u2", + }, + ) + for _, ordinal := range []int{1, 2} { + var msgID int64 + require.NoError(t, srcDB.getReader().QueryRow( + "SELECT id FROM messages WHERE session_id = 's1' AND ordinal = ?", + ordinal, + ).Scan(&msgID), "resolve s1 message id") + pinID, err := srcDB.PinMessage("s1", msgID, nil) + require.NoError(t, err, "pin s1 ordinal %d", ordinal) + require.NotZero(t, pinID, "pin s1 ordinal %d not created", ordinal) + } + + // Legacy session without source uuids still restores by ordinal. + insertSession(t, srcDB, "s2", "proj") + insertMessages(t, srcDB, Message{ + SessionID: "s2", Ordinal: 1, Role: "user", + Content: "legacy", ContentLength: 6, + }) + pinByOrdinal := func(d *DB, sessionID string, ordinal int) { + t.Helper() + var msgID int64 + require.NoError(t, d.getReader().QueryRow( + "SELECT id FROM messages WHERE session_id = ? AND ordinal = ?", + sessionID, ordinal, + ).Scan(&msgID), "resolve %s ordinal %d", sessionID, ordinal) + pinID, err := d.PinMessage(sessionID, msgID, nil) + require.NoError(t, err, "pin %s ordinal %d", sessionID, ordinal) + require.NotZero(t, pinID, "pin %s ordinal %d not created", + sessionID, ordinal) + } + pinByOrdinal(srcDB, "s2", 1) + + // Session whose pinned message vanished in the re-parse while an + // unrelated message took over its ordinal. + insertSession(t, srcDB, "s3", "proj") + insertMessages(t, srcDB, Message{ + SessionID: "s3", Ordinal: 1, Role: "user", + Content: "gone soon", ContentLength: 9, SourceUUID: "u-gone", + }) + pinByOrdinal(srcDB, "s3", 1) + + // Session whose pinned message's uuid is duplicated in the fresh + // DB; the old ordinal still identifies which duplicate was meant. + insertSession(t, srcDB, "s4", "proj") + insertMessages(t, srcDB, + Message{ + SessionID: "s4", Ordinal: 1, Role: "user", + Content: "dup a", ContentLength: 5, SourceUUID: "u-dup", + }, + Message{ + SessionID: "s4", Ordinal: 2, Role: "user", + Content: "dup b", ContentLength: 5, SourceUUID: "u-dup", + }, + ) + pinByOrdinal(srcDB, "s4", 2) + + // Session where the OLD DB itself holds duplicate uuids and only + // one row survives the re-parse. The uuid cannot identify which + // duplicate the pin was on, so a pin on the removed duplicate + // must not transfer to the survivor. + insertSession(t, srcDB, "s5", "proj") + insertMessages(t, srcDB, + Message{ + SessionID: "s5", Ordinal: 1, Role: "user", + Content: "removed dup", ContentLength: 11, + SourceUUID: "u-old-dup", + }, + Message{ + SessionID: "s5", Ordinal: 2, Role: "user", + Content: "surviving dup", ContentLength: 13, + SourceUUID: "u-old-dup", + }, + ) + pinByOrdinal(srcDB, "s5", 1) + + // Same old-side duplication, but the pin sits on the duplicate + // that survives at its ordinal: the guarded ordinal fallback + // still restores it. + insertSession(t, srcDB, "s6", "proj") + insertMessages(t, srcDB, + Message{ + SessionID: "s6", Ordinal: 1, Role: "user", + Content: "removed dup", ContentLength: 11, + SourceUUID: "u-old-dup6", + }, + Message{ + SessionID: "s6", Ordinal: 2, Role: "user", + Content: "surviving dup", ContentLength: 13, + SourceUUID: "u-old-dup6", + }, + ) + pinByOrdinal(srcDB, "s6", 2) + + // Matching role/content cannot disambiguate identical old-side + // duplicates when only one survives at the pinned ordinal. + insertSession(t, srcDB, "s7", "proj") + insertMessages(t, srcDB, + Message{ + SessionID: "s7", Ordinal: 1, Role: "user", + Content: "same dup", ContentLength: 8, + SourceUUID: "u-identical-dup", + }, + Message{ + SessionID: "s7", Ordinal: 2, Role: "user", + Content: "same dup", ContentLength: 8, + SourceUUID: "u-identical-dup", + }, + ) + pinByOrdinal(srcDB, "s7", 1) + + // Legacy session (no source uuids) whose pinned combined prompt + // is split by the re-parse: the hidden envelope row takes over + // the pinned ordinal. + insertSession(t, srcDB, "s8", "proj") + insertMessages(t, srcDB, Message{ + SessionID: "s8", Ordinal: 1, Role: "user", + Content: "f explain", + ContentLength: 44, + }) + pinByOrdinal(srcDB, "s8", 1) + srcDB.Close() + + // Destination DB: the re-parse split the envelope into its own + // hidden row, shifting the prompt and reply down by one ordinal. + dstPath := filepath.Join(dir, "dst.db") + dstDB := testDBAtPath(t, dstPath, "dst") + defer dstDB.Close() + insertSession(t, dstDB, "s1", "proj") + insertMessages(t, dstDB, + Message{ + SessionID: "s1", Ordinal: 1, Role: "user", + Content: "f", + ContentLength: 36, IsSystem: true, + SourceUUID: "u1:ide-context", + }, + Message{ + SessionID: "s1", Ordinal: 2, Role: "user", + Content: "explain", ContentLength: 7, SourceUUID: "u1", + }, + Message{ + SessionID: "s1", Ordinal: 3, Role: "assistant", + Content: "sure", ContentLength: 4, SourceUUID: "u2", + }, + ) + insertSession(t, dstDB, "s7", "proj") + insertMessages(t, dstDB, Message{ + SessionID: "s7", Ordinal: 1, Role: "user", + Content: "same dup", ContentLength: 8, + SourceUUID: "u-identical-dup", + }) + insertSession(t, dstDB, "s8", "proj") + insertMessages(t, dstDB, + Message{ + SessionID: "s8", Ordinal: 1, Role: "user", + Content: "f", + ContentLength: 36, IsSystem: true, + SourceUUID: "u8:ide-context", + }, + Message{ + SessionID: "s8", Ordinal: 2, Role: "user", + Content: "explain", ContentLength: 7, SourceUUID: "u8", + }, + ) + insertSession(t, dstDB, "s2", "proj") + insertMessages(t, dstDB, Message{ + SessionID: "s2", Ordinal: 1, Role: "user", + Content: "legacy", ContentLength: 6, + }) + insertSession(t, dstDB, "s3", "proj") + insertMessages(t, dstDB, Message{ + SessionID: "s3", Ordinal: 1, Role: "user", + Content: "unrelated", ContentLength: 9, SourceUUID: "u-other", + }) + insertSession(t, dstDB, "s4", "proj") + insertMessages(t, dstDB, + Message{ + SessionID: "s4", Ordinal: 1, Role: "user", + Content: "dup a", ContentLength: 5, SourceUUID: "u-dup", + }, + Message{ + SessionID: "s4", Ordinal: 2, Role: "user", + Content: "dup b", ContentLength: 5, SourceUUID: "u-dup", + }, + ) + + insertSession(t, dstDB, "s5", "proj") + insertMessages(t, dstDB, Message{ + SessionID: "s5", Ordinal: 1, Role: "user", + Content: "surviving dup", ContentLength: 13, + SourceUUID: "u-old-dup", + }) + insertSession(t, dstDB, "s6", "proj") + insertMessages(t, dstDB, + Message{ + SessionID: "s6", Ordinal: 1, Role: "user", + Content: "unrelated", ContentLength: 9, + SourceUUID: "u-fresh6", + }, + Message{ + SessionID: "s6", Ordinal: 2, Role: "user", + Content: "surviving dup", ContentLength: 13, + SourceUUID: "u-old-dup6", + }, + ) + + require.NoError(t, dstDB.CopySessionMetadataFrom(srcPath), + "CopySessionMetadataFrom") + + // Pins follow source_uuid across the ordinal shift instead of + // landing on the hidden envelope row at their old ordinals, and + // no duplicate pin is created by the ordinal fallback. + pins, err := dstDB.ListPinnedMessages(ctx, "s1", "") + require.NoError(t, err, "ListPins s1") + require.Len(t, pins, 2, "pins s1") + gotOrdinals := []int{pins[0].Ordinal, pins[1].Ordinal} + slices.Sort(gotOrdinals) + assert.Equal(t, []int{2, 3}, gotOrdinals, + "pins should follow source_uuid to the shifted ordinals") + + pins, err = dstDB.ListPinnedMessages(ctx, "s2", "") + require.NoError(t, err, "ListPins s2") + require.Len(t, pins, 1, "pins s2") + assert.Equal(t, 1, pins[0].Ordinal, + "legacy pin without source_uuid falls back to ordinal") + + pins, err = dstDB.ListPinnedMessages(ctx, "s3", "") + require.NoError(t, err, "ListPins s3") + assert.Empty(t, pins, + "pin whose uuid vanished must be dropped, not attached to the "+ + "unrelated message now at its ordinal") + + pins, err = dstDB.ListPinnedMessages(ctx, "s4", "") + require.NoError(t, err, "ListPins s4") + require.Len(t, pins, 1, "pins s4") + assert.Equal(t, 2, pins[0].Ordinal, + "duplicated uuid resolves by old ordinal to the same-uuid row") + + pins, err = dstDB.ListPinnedMessages(ctx, "s5", "") + require.NoError(t, err, "ListPins s5") + assert.Empty(t, pins, + "pin on a removed old-side duplicate must not transfer to the "+ + "same-uuid survivor that shifted into its ordinal") + + pins, err = dstDB.ListPinnedMessages(ctx, "s6", "") + require.NoError(t, err, "ListPins s6") + require.Len(t, pins, 1, "pins s6") + assert.Equal(t, 2, pins[0].Ordinal, + "pin on the surviving old-side duplicate restores at its ordinal") + + pins, err = dstDB.ListPinnedMessages(ctx, "s7", "") + require.NoError(t, err, "ListPins s7") + assert.Empty(t, pins, + "pin on indistinguishable old duplicates must be dropped when "+ + "their multiplicity changes") + + pins, err = dstDB.ListPinnedMessages(ctx, "s8", "") + require.NoError(t, err, "ListPins s8") + assert.Empty(t, pins, + "uuid-less pin on a split combined prompt must not attach to "+ + "the hidden envelope row at its old ordinal") +} + func TestCopySessionMetadataCopiesFromSource(t *testing.T) { dir := t.TempDir() ctx := context.Background() diff --git a/internal/db/messages.go b/internal/db/messages.go index 7e070b4ae..0bf264854 100644 --- a/internal/db/messages.go +++ b/internal/db/messages.go @@ -1106,23 +1106,33 @@ func (db *DB) LastClaudeMessageID(sessionID string) string { return s.String } -// savedPin captures the minimal pin state needed to re-attach a pin -// after a full message replacement. source_uuid is the preferred -// identifier because it survives rewrites where the ordinal stream -// shifts (e.g. when newly-emitted compact-boundary messages are -// inserted between previously-seen rows). The ordinal is kept as a -// fallback for legacy pins on rows that lack a source_uuid. +// savedPin captures the message identity needed to re-attach a pin +// after a full message replacement. source_uuid is preferred because +// it survives ordinal shifts. Role and content, together with the +// pin's occurrence rank inside its identity group, guard the +// fallback used for legacy rows and ambiguous source UUIDs: rank +// follows a message across ordinal shifts that leave the group +// intact, where a saved ordinal would name a different occurrence. type savedPin struct { - sourceUUID string - ordinal int - note *string - createdAt string + sourceUUID string + role string + content string + ordinal int + sourceUUIDCount int + sourceIdentityCount int + sourceIdentityRank int + legacyIdentityCount int + legacyIdentityRank int + messageFound int + note *string + createdAt string } // ReplaceSessionMessages deletes existing and inserts new messages // in a single transaction. Any existing pins are preserved by // re-attaching them to the new message rows that share the same -// ordinal (pins for ordinals that no longer exist are dropped). +// unambiguous message identity (pins whose message no longer exists +// are dropped). func (db *DB) ReplaceSessionMessages( sessionID string, msgs []Message, ) error { @@ -1158,6 +1168,15 @@ func (db *DB) ReplaceSessionMessages( return fmt.Errorf("beginning tx: %w", err) } defer func() { _ = tx.Rollback() }() + if useDiff { + needsPinRemap, err := messageDiffNeedsPinRemapTx(tx, plan) + if err != nil { + return err + } + if needsPinRemap { + useDiff = false + } + } queueGenerationBefore, queueExistedBefore, err := artifactExportGenerationTx( tx, sessionID, ) @@ -1376,6 +1395,15 @@ func (db *DB) ReplaceSessionContent( return fmt.Errorf("beginning tx: %w", err) } defer func() { _ = tx.Rollback() }() + if useDiff { + needsPinRemap, err := messageDiffNeedsPinRemapTx(tx, plan) + if err != nil { + return err + } + if needsPinRemap { + useDiff = false + } + } queueGenerationBefore, queueExistedBefore, err := artifactExportGenerationTx( tx, sessionID, ) @@ -1529,9 +1557,56 @@ func savePinsTx(tx *sql.Tx, sessionID string) ([]savedPin, error) { // pinned_messages.message_id would otherwise wipe them when // messages are deleted below. source_uuid comes from the joined // message row; LEFT JOIN keeps pins on legacy rows whose - // message_id no longer resolves cleanly. + // message_id no longer resolves cleanly. The counts capture whether + // source_uuid, or source_uuid plus role and content, uniquely identify + // the old message before it is deleted. pinRows, err := tx.Query(` SELECT p.ordinal, COALESCE(m.source_uuid, ''), + COALESCE(m.role, ''), COALESCE(m.content, ''), + CASE WHEN m.id IS NULL THEN 0 ELSE 1 END, + ( + SELECT COUNT(*) + FROM messages same_uuid + WHERE same_uuid.session_id = m.session_id + AND same_uuid.source_uuid = m.source_uuid + AND m.source_uuid != '' + ), + ( + SELECT COUNT(*) + FROM messages same_identity + WHERE same_identity.session_id = m.session_id + AND same_identity.source_uuid = m.source_uuid + AND same_identity.role = m.role + AND same_identity.content = m.content + AND m.source_uuid != '' + ), + ( + SELECT COUNT(*) + FROM messages identity_rank + WHERE identity_rank.session_id = m.session_id + AND identity_rank.source_uuid = m.source_uuid + AND identity_rank.role = m.role + AND identity_rank.content = m.content + AND identity_rank.ordinal <= m.ordinal + AND m.source_uuid != '' + ), + ( + SELECT COUNT(*) + FROM messages legacy_identity + WHERE legacy_identity.session_id = m.session_id + AND legacy_identity.role = m.role + AND legacy_identity.content = m.content + AND legacy_identity.is_system = 0 + ), + ( + SELECT COUNT(*) + FROM messages legacy_rank + WHERE legacy_rank.session_id = m.session_id + AND legacy_rank.role = m.role + AND legacy_rank.content = m.content + AND legacy_rank.is_system = 0 + AND legacy_rank.ordinal <= m.ordinal + ), p.note, p.created_at FROM pinned_messages p LEFT JOIN messages m ON m.id = p.message_id @@ -1546,7 +1621,11 @@ func savePinsTx(tx *sql.Tx, sessionID string) ([]savedPin, error) { for pinRows.Next() { var sp savedPin if err := pinRows.Scan( - &sp.ordinal, &sp.sourceUUID, &sp.note, &sp.createdAt, + &sp.ordinal, &sp.sourceUUID, &sp.role, &sp.content, + &sp.messageFound, &sp.sourceUUIDCount, + &sp.sourceIdentityCount, &sp.sourceIdentityRank, + &sp.legacyIdentityCount, &sp.legacyIdentityRank, + &sp.note, &sp.createdAt, ); err != nil { return nil, fmt.Errorf("scanning pin: %w", err) } @@ -1561,40 +1640,156 @@ func savePinsTx(tx *sql.Tx, sessionID string) ([]savedPin, error) { func restorePinsTx( tx *sql.Tx, sessionID string, pins []savedPin, ) error { - // Re-attach saved pins. Prefer source_uuid (stable across - // ordinal-shifting rewrites) and fall back to ordinal for - // legacy pins whose source row predates the source_uuid column. - // Pins whose row no longer exists by either key are silently - // dropped. + // Re-attach saved pins only when the old and new message identities + // are both unambiguous. A unique source_uuid may move to another + // ordinal. Duplicate UUIDs and legacy UUID-less rows must retain + // their role, content, and occurrence rank inside an equally sized + // identity group. A legacy UUID-less row may gain a provider UUID + // while retaining that fallback identity. Otherwise the pin is + // dropped rather than duplicated or attached to an unrelated + // message. for _, sp := range pins { + if sp.messageFound == 0 { + continue + } + var err error if sp.sourceUUID != "" { - res, err := tx.Exec(` - INSERT OR IGNORE INTO pinned_messages - (session_id, message_id, ordinal, note, created_at) - SELECT ?, m.id, m.ordinal, ?, ? - FROM messages m - WHERE m.session_id = ? AND m.source_uuid = ?`, - sessionID, sp.note, sp.createdAt, sessionID, sp.sourceUUID, - ) - if err != nil { - return fmt.Errorf( - "restoring pin uuid=%s: %w", sp.sourceUUID, err, - ) - } - if n, _ := res.RowsAffected(); n > 0 { - continue - } + err = restorePinBySourceUUIDTx(tx, sessionID, sp) + } else { + err = restoreLegacyPinByRankTx(tx, sessionID, sp) + } + if err != nil { + return err } - if _, err := tx.Exec(` + } + return nil +} + +func restorePinBySourceUUIDTx( + tx *sql.Tx, sessionID string, sp savedPin, +) error { + if sp.sourceUUIDCount == 1 { + res, err := tx.Exec(` INSERT OR IGNORE INTO pinned_messages (session_id, message_id, ordinal, note, created_at) SELECT ?, m.id, m.ordinal, ?, ? FROM messages m - WHERE m.session_id = ? AND m.ordinal = ?`, - sessionID, sp.note, sp.createdAt, sessionID, sp.ordinal, - ); err != nil { - return fmt.Errorf("restoring pin ord=%d: %w", sp.ordinal, err) + WHERE m.session_id = ? AND m.source_uuid = ? + AND ( + SELECT COUNT(*) + FROM messages same_uuid + WHERE same_uuid.session_id = m.session_id + AND same_uuid.source_uuid = m.source_uuid + ) = 1`, + sessionID, sp.note, sp.createdAt, + sessionID, sp.sourceUUID, + ) + if err != nil { + return fmt.Errorf( + "restoring unique pin uuid=%s: %w", + sp.sourceUUID, err, + ) } + n, err := res.RowsAffected() + if err != nil { + return fmt.Errorf( + "checking restored pin uuid=%s: %w", + sp.sourceUUID, err, + ) + } + if n > 0 { + return nil + } + } + // Identical (uuid, role, content) rows are distinguishable only by + // position, so require the identity multiplicity to be unchanged + // and re-attach at the pin's occurrence rank inside the group. + // Rank, unlike the saved ordinal, follows the pinned occurrence + // across shifts caused by rows inserted before the group. A + // different count means duplicates were inserted or removed and + // the rank no longer identifies an occurrence, so the pin is + // dropped instead. + if _, err := tx.Exec(` + INSERT OR IGNORE INTO pinned_messages + (session_id, message_id, ordinal, note, created_at) + SELECT ?, m.id, m.ordinal, ?, ? + FROM messages m + WHERE m.session_id = ? + AND m.source_uuid = ? + AND m.role = ? AND m.content = ? + AND ( + SELECT COUNT(*) + FROM messages same_identity + WHERE same_identity.session_id = m.session_id + AND same_identity.source_uuid = m.source_uuid + AND same_identity.role = m.role + AND same_identity.content = m.content + ) = ? + AND ( + SELECT COUNT(*) + FROM messages identity_rank + WHERE identity_rank.session_id = m.session_id + AND identity_rank.source_uuid = m.source_uuid + AND identity_rank.role = m.role + AND identity_rank.content = m.content + AND identity_rank.ordinal <= m.ordinal + ) = ?`, + sessionID, sp.note, sp.createdAt, sessionID, + sp.sourceUUID, sp.role, sp.content, + sp.sourceIdentityCount, sp.sourceIdentityRank, + ); err != nil { + return fmt.Errorf( + "restoring ambiguous pin uuid=%s ord=%d: %w", + sp.sourceUUID, sp.ordinal, err, + ) + } + return nil +} + +// restoreLegacyPinByRankTx re-attaches a UUID-less pin to the visible +// row holding the pin's role, content, and occurrence rank within the +// visible (role, content) group, provided the group kept its size. +// Rank follows the pinned occurrence across ordinal shifts; matching +// the saved ordinal instead could attach the pin to an earlier equal +// message that shifted into its place. A changed group size — or an +// edit to the pinned message itself — means the pinned message can no +// longer be identified and the pin is dropped. +func restoreLegacyPinByRankTx( + tx *sql.Tx, sessionID string, sp savedPin, +) error { + _, err := tx.Exec(` + INSERT OR IGNORE INTO pinned_messages + (session_id, message_id, ordinal, note, created_at) + SELECT ?, m.id, m.ordinal, ?, ? + FROM messages m + WHERE m.session_id = ? + AND m.is_system = 0 + AND m.role = ? AND m.content = ? + AND ( + SELECT COUNT(*) + FROM messages legacy_identity + WHERE legacy_identity.session_id = m.session_id + AND legacy_identity.role = m.role + AND legacy_identity.content = m.content + AND legacy_identity.is_system = 0 + ) = ? + AND ( + SELECT COUNT(*) + FROM messages legacy_rank + WHERE legacy_rank.session_id = m.session_id + AND legacy_rank.role = m.role + AND legacy_rank.content = m.content + AND legacy_rank.is_system = 0 + AND legacy_rank.ordinal <= m.ordinal + ) = ?`, + sessionID, sp.note, sp.createdAt, sessionID, + sp.role, sp.content, + sp.legacyIdentityCount, sp.legacyIdentityRank, + ) + if err != nil { + return fmt.Errorf( + "restoring legacy pin ord=%d: %w", sp.ordinal, err, + ) } return nil } diff --git a/internal/db/messages_diff.go b/internal/db/messages_diff.go index 1608b327c..42cbe677c 100644 --- a/internal/db/messages_diff.go +++ b/internal/db/messages_diff.go @@ -100,8 +100,9 @@ type messageDiffUpdate struct { } type messageDiffPlan struct { - updates []messageDiffUpdate - inserts []Message + updates []messageDiffUpdate + inserts []Message + unsafePinUpdateIDs []int64 } // planStoredMessageDiff loads the session's stored messages and @@ -201,6 +202,8 @@ func planSessionMessageDiff( } byOrdinal[m.Ordinal] = m } + storedUUIDCounts := messageSourceUUIDCounts(stored) + incomingUUIDCounts := messageSourceUUIDCounts(incoming) var plan messageDiffPlan seen := make(map[int]bool, len(incoming)) @@ -221,6 +224,13 @@ func planSessionMessageDiff( id: old.ID, msg: m, }) + if !messagePinIdentityStable( + old, m, storedUUIDCounts, incomingUUIDCounts, + ) { + plan.unsafePinUpdateIDs = append( + plan.unsafePinUpdateIDs, old.ID, + ) + } } } for ord := range byOrdinal { @@ -234,6 +244,71 @@ func planSessionMessageDiff( return plan, true } +func messageSourceUUIDCounts(msgs []Message) map[string]int { + counts := make(map[string]int) + for _, msg := range msgs { + if msg.SourceUUID != "" { + counts[msg.SourceUUID]++ + } + } + return counts +} + +func messagePinIdentityStable( + old, incoming Message, + oldUUIDCounts, incomingUUIDCounts map[string]int, +) bool { + if old.SourceUUID != "" && + old.SourceUUID == incoming.SourceUUID && + oldUUIDCounts[old.SourceUUID] == 1 && + incomingUUIDCounts[incoming.SourceUUID] == 1 { + return true + } + if old.Ordinal != incoming.Ordinal || old.Role != incoming.Role { + return false + } + if old.Content == incoming.Content { + return true + } + // A content extension is the same message completed by a later + // parse (e.g. a streamed partial response): the row keeps its + // ordinal, role, and source uuid (the caller refuses uuid + // changes), so the in-place update may retain the pin. The old + // content must be a non-empty prefix so an empty placeholder + // cannot claim an arbitrary replacement as its completion. + return old.Content != "" && + strings.HasPrefix(incoming.Content, old.Content) +} + +// messageDiffNeedsPinRemapTx reports whether an in-place update would +// retain a pin on a row whose identity changed ambiguously. The caller +// can then use the full replacement path, which drops or remaps the pin +// through the guarded identity rules. Unpinned streaming updates retain +// the in-place path. +func messageDiffNeedsPinRemapTx( + tx *sql.Tx, plan messageDiffPlan, +) (bool, error) { + for start := 0; start < len(plan.unsafePinUpdateIDs); start += diffDeleteChunkSize { + end := min(start+diffDeleteChunkSize, len(plan.unsafePinUpdateIDs)) + args := make([]any, 0, end-start) + for _, id := range plan.unsafePinUpdateIDs[start:end] { + args = append(args, id) + } + var exists int + if err := tx.QueryRow( + "SELECT EXISTS (SELECT 1 FROM pinned_messages "+ + "WHERE message_id IN ("+placeholderList(len(args))+"))", + args..., + ).Scan(&exists); err != nil { + return false, fmt.Errorf("checking diff pins: %w", err) + } + if exists != 0 { + return true, nil + } + } + return false, nil +} + // applySessionMessageDiffTx persists a planned diff: changed rows // are updated in place (keeping rowids, so pins survive and the FTS // triggers reindex only those rows), their tool rows are rebuilt, diff --git a/internal/db/messages_diff_test.go b/internal/db/messages_diff_test.go index 6090a9909..3115cd8d4 100644 --- a/internal/db/messages_diff_test.go +++ b/internal/db/messages_diff_test.go @@ -256,7 +256,8 @@ func TestReplaceSessionMessagesKeepsPinOnMergedRow(t *testing.T) { d := testDB(t) v1 := []Message{ diffTestMsg("pin-s", 0, "user", "hello"), - diffTestMsg("pin-s", 1, "assistant", "partial"), + diffTestMsg("pin-s", 1, "assistant", "partial", + func(m *Message) { m.SourceUUID = "pin-tail" }), } seedDiffSession(t, d, "pin-s", v1) ids := messageIDsByOrdinal(t, d, "pin-s") @@ -267,7 +268,8 @@ func TestReplaceSessionMessagesKeepsPinOnMergedRow(t *testing.T) { v2 := []Message{ v1[0], - diffTestMsg("pin-s", 1, "assistant", "partial now complete"), + diffTestMsg("pin-s", 1, "assistant", "partial now complete", + func(m *Message) { m.SourceUUID = "pin-tail" }), diffTestMsg("pin-s", 2, "user", "more"), } require.NoError(t, d.ReplaceSessionMessages("pin-s", v2)) @@ -280,3 +282,115 @@ func TestReplaceSessionMessagesKeepsPinOnMergedRow(t *testing.T) { ).Scan(&n)) assert.Equal(t, 1, n, "pin on the merged row must survive") } + +// TestReplaceSessionMessagesKeepsPinOnCompletedRow covers a pinned +// partial response without a usable UUID whose content is completed by +// a later parse: the extension keeps ordinal, role, and uuid, so the +// row identity is preserved and the pin must survive the in-place +// merge instead of being remapped and dropped. +func TestReplaceSessionMessagesKeepsPinOnCompletedRow(t *testing.T) { + for _, tc := range []struct { + name string + sourceUUIDs []string + }{ + { + name: "empty UUID", + sourceUUIDs: []string{"", "", ""}, + }, + { + name: "duplicate UUID", + sourceUUIDs: []string{"tail", "duplicate", "duplicate"}, + }, + } { + t.Run(tc.name, func(t *testing.T) { + d := testDB(t) + v1 := []Message{ + diffTestMsg("pin-complete", 0, "user", "first"), + diffTestMsg("pin-complete", 1, "assistant", "partial"), + diffTestMsg("pin-complete", 2, "user", "last"), + } + for i := range v1 { + v1[i].SourceUUID = tc.sourceUUIDs[i] + } + seedDiffSession(t, d, "pin-complete", v1) + ids := messageIDsByOrdinal(t, d, "pin-complete") + _, err := d.PinMessage("pin-complete", ids[1], nil) + require.NoError(t, err, "PinMessage") + + v2 := append([]Message(nil), v1...) + v2[1].Content = "partial now complete" + v2[1].ContentLength = len(v2[1].Content) + require.NoError(t, + d.ReplaceSessionMessages("pin-complete", v2)) + + pins, err := d.ListPinnedMessages( + context.Background(), "pin-complete", "", + ) + require.NoError(t, err, "ListPinnedMessages") + require.Len(t, pins, 1, + "the completed row must keep its pin") + assert.Equal(t, 1, pins[0].Ordinal, + "pin stays on the completed message") + }) + } +} + +func TestReplaceSessionMessagesDropsPinOnAmbiguousChangedRow(t *testing.T) { + for _, tc := range []struct { + name string + sourceUUIDs []string + }{ + { + name: "empty UUID", + sourceUUIDs: []string{"", "", ""}, + }, + { + name: "duplicate UUID", + sourceUUIDs: []string{"duplicate", "duplicate", "tail"}, + }, + } { + t.Run(tc.name, func(t *testing.T) { + d := testDB(t) + v1 := []Message{ + diffTestMsg("pin-ambiguous", 0, "user", "first"), + diffTestMsg("pin-ambiguous", 1, "assistant", "pinned"), + diffTestMsg("pin-ambiguous", 2, "user", "last"), + } + for i := range v1 { + v1[i].SourceUUID = tc.sourceUUIDs[i] + } + seedDiffSession(t, d, "pin-ambiguous", v1) + ids := messageIDsByOrdinal(t, d, "pin-ambiguous") + _, err := d.PinMessage("pin-ambiguous", ids[1], nil) + require.NoError(t, err, "PinMessage") + + v2 := append([]Message(nil), v1...) + v2[1].Content = "unrelated replacement" + v2[1].ContentLength = len(v2[1].Content) + require.NoError(t, + d.ReplaceSessionMessages("pin-ambiguous", v2)) + + pins, err := d.ListPinnedMessages( + context.Background(), "pin-ambiguous", "", + ) + require.NoError(t, err, "ListPinnedMessages") + assert.Empty(t, pins, + "an ambiguous identity change must not inherit the pin") + }) + } +} + +func TestMessagePinIdentityStableRequiresSameUniqueUUID(t *testing.T) { + old := diffTestMsg("pin-identity", 1, "assistant", "old content", + func(m *Message) { m.SourceUUID = "uuid-old" }) + incoming := diffTestMsg( + "pin-identity", 1, "assistant", "replacement content", + func(m *Message) { m.SourceUUID = "uuid-new" }, + ) + + assert.False(t, messagePinIdentityStable( + old, incoming, + map[string]int{"uuid-old": 1}, + map[string]int{"uuid-new": 1}, + ), "different unique UUIDs are different pin identities") +} diff --git a/internal/db/orphaned.go b/internal/db/orphaned.go index 0f2f73d99..3c4cbe724 100644 --- a/internal/db/orphaned.go +++ b/internal/db/orphaned.go @@ -1188,24 +1188,191 @@ func (d *DB) CopySessionMetadataFrom( } // Copy pinned messages (table may not exist in older DBs). - // Map old message_id to new message_id via the - // (session_id, ordinal) natural key, since auto-increment - // IDs differ between DBs. + // Auto-increment message IDs differ between DBs, so old + // message_id must be re-resolved against the fresh rows. + // Prefer the source_uuid natural key: a re-parse can insert or + // drop rows (e.g. the v88 IDE-envelope split), shifting ordinals + // so that the old (session_id, ordinal) key lands on an unrelated + // row. The uuid must be unique on BOTH sides: a duplicate in the + // old DB means the uuid does not identify which message the pin + // was on, so transferring it to a lone same-uuid survivor could + // misattach a pin whose real target was removed by the re-parse. + // Duplicated uuids fall back to the pin's occurrence rank inside + // its (uuid, role, content) group, requiring the group to keep its + // size on both sides: rank follows the pinned occurrence across + // ordinal shifts, while a changed group size means the rank no + // longer identifies an occurrence. Legacy pins whose source row + // has no source_uuid fall back the same way over the visible + // (role, content) group. A nonempty uuid with no safe + // match means the pinned message is gone: the pin is dropped rather + // than silently attached to whatever now occupies its ordinal. if oldDBHasTable(ctx, tx, "pinned_messages") { + hasSourceUUID := oldDBHasColumn( + ctx, tx, "messages", "source_uuid", + ) + if hasSourceUUID { + if _, err := tx.ExecContext(ctx, ` + INSERT OR IGNORE INTO main.pinned_messages + (session_id, message_id, ordinal, note, created_at) + SELECT + op.session_id, new_m.id, new_m.ordinal, + op.note, op.created_at + FROM old_db.pinned_messages op + JOIN old_db.messages old_m + ON old_m.id = op.message_id + JOIN main.messages new_m + ON new_m.session_id = old_m.session_id + AND new_m.source_uuid = old_m.source_uuid + WHERE op.session_id IN ( + SELECT id FROM main.sessions + ) + AND old_m.source_uuid != '' + AND ( + SELECT COUNT(*) FROM main.messages x + WHERE x.session_id = old_m.session_id + AND x.source_uuid = old_m.source_uuid + ) = 1 + AND ( + SELECT COUNT(*) FROM old_db.messages y + WHERE y.session_id = old_m.session_id + AND y.source_uuid = old_m.source_uuid + ) = 1`); err != nil { + return fmt.Errorf( + "copying pinned messages by source uuid: %w", err, + ) + } + } + // Rank fallback for duplicated uuids: identical (uuid, role, + // content) rows are distinguishable only by position, so a pin + // transfers to the row holding the same occurrence rank inside + // its identity group, provided the group kept its size on both + // sides. Rank, unlike the old ordinal, follows the pinned + // occurrence across shifts caused by rows inserted before the + // group; a changed group size means the rank no longer + // identifies an occurrence and the pin is dropped. When the + // uuid was unique the source_uuid pass already restored the + // same row and INSERT OR IGNORE dedupes. + if hasSourceUUID { + if _, err := tx.ExecContext(ctx, ` + INSERT OR IGNORE INTO main.pinned_messages + (session_id, message_id, ordinal, note, created_at) + SELECT + op.session_id, new_m.id, new_m.ordinal, + op.note, op.created_at + FROM old_db.pinned_messages op + JOIN old_db.messages old_m + ON old_m.id = op.message_id + JOIN main.messages new_m + ON new_m.session_id = old_m.session_id + AND new_m.source_uuid = old_m.source_uuid + AND new_m.role = old_m.role + AND new_m.content = old_m.content + WHERE op.session_id IN ( + SELECT id FROM main.sessions + ) + AND old_m.source_uuid != '' + AND ( + SELECT COUNT(*) FROM old_db.messages y + WHERE y.session_id = old_m.session_id + AND y.source_uuid = old_m.source_uuid + AND y.role = old_m.role + AND y.content = old_m.content + ) = ( + SELECT COUNT(*) FROM main.messages x + WHERE x.session_id = old_m.session_id + AND x.source_uuid = old_m.source_uuid + AND x.role = old_m.role + AND x.content = old_m.content + ) + AND ( + SELECT COUNT(*) FROM old_db.messages y2 + WHERE y2.session_id = old_m.session_id + AND y2.source_uuid = old_m.source_uuid + AND y2.role = old_m.role + AND y2.content = old_m.content + AND y2.ordinal <= old_m.ordinal + ) = ( + SELECT COUNT(*) FROM main.messages x2 + WHERE x2.session_id = old_m.session_id + AND x2.source_uuid = old_m.source_uuid + AND x2.role = old_m.role + AND x2.content = old_m.content + AND x2.ordinal <= new_m.ordinal + )`); err != nil { + return fmt.Errorf( + "copying duplicated-uuid pinned messages: %w", err, + ) + } + } + // Rank fallback for legacy pins without a uuid, mirroring + // restoreLegacyPinByRankTx: the pin transfers to the visible + // row holding its role, content, and occurrence rank within + // the visible (role, content) group, provided the group kept + // its size on both sides. Rank follows the pinned occurrence + // across shifts from rows the re-parse inserted (e.g. hidden + // IDE-envelope rows); a changed group size means the rank no + // longer identifies an occurrence and the pin is dropped. A + // legacy row may gain a provider uuid in the fresh DB while + // retaining this fallback identity. Old archives may predate + // the is_system column; without it every old row counts as + // visible. + legacyOnly := "" + if hasSourceUUID { + legacyOnly = ` + AND (old_m.source_uuid IS NULL + OR old_m.source_uuid = '')` + } + oldMVisible, oldYVisible, oldY2Visible := "", "", "" + if oldDBHasColumn(ctx, tx, "messages", "is_system") { + oldMVisible = ` + AND old_m.is_system = 0` + oldYVisible = ` + AND y.is_system = 0` + oldY2Visible = ` + AND y2.is_system = 0` + } if _, err := tx.ExecContext(ctx, ` INSERT OR IGNORE INTO main.pinned_messages (session_id, message_id, ordinal, note, created_at) SELECT - op.session_id, new_m.id, op.ordinal, + op.session_id, new_m.id, new_m.ordinal, op.note, op.created_at FROM old_db.pinned_messages op JOIN old_db.messages old_m ON old_m.id = op.message_id JOIN main.messages new_m ON new_m.session_id = old_m.session_id - AND new_m.ordinal = old_m.ordinal + AND new_m.role = old_m.role + AND new_m.content = old_m.content + AND new_m.is_system = 0 WHERE op.session_id IN ( SELECT id FROM main.sessions + )`+legacyOnly+oldMVisible+` + AND ( + SELECT COUNT(*) FROM old_db.messages y + WHERE y.session_id = old_m.session_id + AND y.role = old_m.role + AND y.content = old_m.content`+oldYVisible+` + ) = ( + SELECT COUNT(*) FROM main.messages x + WHERE x.session_id = old_m.session_id + AND x.role = old_m.role + AND x.content = old_m.content + AND x.is_system = 0 + ) + AND ( + SELECT COUNT(*) FROM old_db.messages y2 + WHERE y2.session_id = old_m.session_id + AND y2.role = old_m.role + AND y2.content = old_m.content + AND y2.ordinal <= old_m.ordinal`+oldY2Visible+` + ) = ( + SELECT COUNT(*) FROM main.messages x2 + WHERE x2.session_id = old_m.session_id + AND x2.role = old_m.role + AND x2.content = old_m.content + AND x2.is_system = 0 + AND x2.ordinal <= new_m.ordinal )`); err != nil { return fmt.Errorf("copying pinned messages: %w", err) } @@ -2206,35 +2373,11 @@ func copyPinnedMessagesForIDs( return nil } - // Re-map old message IDs to the newly inserted message rows. - // Prefer source_uuid when available because it survives ordinal - // shifts, then fall back to the same (session_id, ordinal) - // natural key used by tool call copying. - if oldDBHasColumn(ctx, tx, "messages", "source_uuid") { - if _, err := tx.ExecContext(ctx, ` - INSERT OR IGNORE INTO main.pinned_messages - (session_id, message_id, ordinal, note, created_at) - SELECT - op.session_id, new_m.id, new_m.ordinal, - op.note, op.created_at - FROM old_db.pinned_messages op - JOIN old_db.messages old_m - ON old_m.id = op.message_id - JOIN main.messages new_m - ON new_m.session_id = old_m.session_id - AND new_m.source_uuid = old_m.source_uuid - WHERE op.session_id IN ( - SELECT id FROM `+tempIDsTable+` - ) - AND old_m.source_uuid IS NOT NULL - AND old_m.source_uuid <> ''`, - ); err != nil { - return fmt.Errorf( - "copying pinned messages by source_uuid: %w", err, - ) - } - } - + // Re-map old message IDs to the newly inserted message rows. These + // orphaned messages were copied verbatim above, so their ordinals + // cannot shift. source_uuid is not a safe key here because providers + // can duplicate it across messages; joining on it would turn one pin + // into one pin for every matching row. if _, err := tx.ExecContext(ctx, ` INSERT OR IGNORE INTO main.pinned_messages (session_id, message_id, ordinal, note, created_at) @@ -2251,7 +2394,7 @@ func copyPinnedMessagesForIDs( SELECT id FROM `+tempIDsTable+` )`, ); err != nil { - return fmt.Errorf("copying pinned messages by ordinal: %w", err) + return fmt.Errorf("copying pinned messages: %w", err) } return nil } diff --git a/internal/db/recall_test.go b/internal/db/recall_test.go index 9441bf863..85563ccff 100644 --- a/internal/db/recall_test.go +++ b/internal/db/recall_test.go @@ -2548,6 +2548,53 @@ func TestCopyRecallEntriesFromReconcilesShiftedEvidence(t *testing.T) { assert.Equal(t, original.ContentDigest, got.Evidence[0].ContentDigest) } +func TestCopyRecallEntriesFromRevokesIDEEnvelopeSplitEvidence(t *testing.T) { + dir := t.TempDir() + srcPath := filepath.Join(dir, "old-ide-evidence.db") + srcDB, err := Open(srcPath) + require.NoError(t, err) + seedRecallEvidenceWindow(t, srcDB, "s1", 10, "stable", "") + oldMessages, err := srcDB.GetAllMessages(context.Background(), "s1") + require.NoError(t, err) + const envelope = "The user opened a.go." + oldMessages[0].Content = envelope + " \nRun the formatter." + oldMessages[0].ContentLength = len(oldMessages[0].Content) + require.NoError(t, srcDB.ReplaceSessionMessages("s1", oldMessages)) + insertVerifiedRecallSelection( + t, srcDB, "m1", "s1", 10, 11, []string{"tool-a"}, + ) + oldMessages, err = srcDB.GetAllMessages(context.Background(), "s1") + require.NoError(t, err) + _, err = srcDB.getWriter().Exec("PRAGMA user_version = 79") + require.NoError(t, err) + require.NoError(t, srcDB.Close()) + + dstPath := filepath.Join(dir, "new-ide-evidence.db") + dstDB, err := Open(dstPath) + require.NoError(t, err) + defer dstDB.Close() + insertSession(t, dstDB, "s1", "agentsview") + for i := range oldMessages { + oldMessages[i].ID = 0 + oldMessages[i].Ordinal++ + } + oldMessages[0].Content = "Run the formatter." + oldMessages[0].ContentLength = len(oldMessages[0].Content) + hidden := recallEvidenceMessage( + "s1", 10, "user", envelope, "stable-10:ide-context", + ) + hidden.IsSystem = true + insertMessages(t, dstDB, append([]Message{hidden}, oldMessages...)...) + + require.NoError(t, dstDB.CopyRecallEntriesFrom(srcPath)) + + got := requireRecallEntry(t, dstDB, "m1") + assert.False(t, got.ProvenanceOK, + "evidence spanning rewritten content must revoke fail-closed") + require.Len(t, got.Evidence, 1, + "revocation retains the historical evidence row") +} + func TestCopyRecallEntriesFromRevokesChangedEvidence(t *testing.T) { dir := t.TempDir() srcPath := filepath.Join(dir, "old-changed-evidence.db") diff --git a/internal/db/session_batch.go b/internal/db/session_batch.go index f8b6906bb..628f3b56e 100644 --- a/internal/db/session_batch.go +++ b/internal/db/session_batch.go @@ -437,7 +437,9 @@ func writeOneSessionBatchTx( } } if replaceMessages { - if err := restorePinsTx(tx, write.Session.ID, pins); err != nil { + if err := restorePinsTx( + tx, write.Session.ID, pins, + ); err != nil { return 0, err } // A full message replacement re-normalizes every row, so this row is diff --git a/internal/parser/claude.go b/internal/parser/claude.go index 8701c1245..d6f494500 100644 --- a/internal/parser/claude.go +++ b/internal/parser/claude.go @@ -1172,6 +1172,28 @@ func extractMessagesFrom( ordinal++ continue } + // The VS Code extension sometimes prepends an IDE-context + // wrapper directly onto a real prompt in the same entry. + // Split it into a hidden system-metadata message plus the + // real prompt, so first_message and the visible transcript + // show only the prompt. + if subtype, envelope, remainder, ok := + splitClaudeIDEEnvelopePrompt(text); ok { + hidden := claudeIDEEnvelopeMessage(e, ordinal, subtype, envelope) + if remainder == "" || isClaudeSystemMessage(remainder) { + // The remainder is discarded, so no visible + // message remains to carry the entry's tool + // results; keep them on the hidden envelope row + // like the standalone classify branch does. + hidden.ToolResults = trs + } + messages = append(messages, hidden) + ordinal++ + if remainder == "" { + continue + } + text = remainder + } // Skip unclassified noise (e.g. non-caveat // envelopes). if isClaudeSystemMessage(text) { @@ -2114,6 +2136,12 @@ func isCountedClaudeUserTurn(entry dagEntry) bool { if skip || strings.TrimSpace(text) == "" { return false } + if _, _, remainder, ok := splitClaudeIDEEnvelopePrompt(text); ok { + if remainder == "" { + return false + } + text = remainder + } return !isClaudeSystemMessage(text) } @@ -2215,6 +2243,28 @@ func extractMessages(entries []dagEntry) ( ordinal++ continue } + // The VS Code extension sometimes prepends an IDE-context + // wrapper directly onto a real prompt in the same entry. + // Split it into a hidden system-metadata message plus the + // real prompt, so first_message and the visible transcript + // show only the prompt. + if subtype, envelope, remainder, ok := + splitClaudeIDEEnvelopePrompt(text); ok { + hidden := claudeIDEEnvelopeMessage(e, ordinal, subtype, envelope) + if remainder == "" || isClaudeSystemMessage(remainder) { + // The remainder is discarded, so no visible + // message remains to carry the entry's tool + // results; keep them on the hidden envelope row + // like the standalone classify branch does. + hidden.ToolResults = trs + } + messages = append(messages, hidden) + ordinal++ + if remainder == "" { + continue + } + text = remainder + } if isClaudeSystemMessage(text) { continue } @@ -2705,6 +2755,104 @@ func isStandaloneClaudeTaggedMessage(content, tag string) bool { len(afterOpen)-len(closeTag) } +// claudeIDEEnvelopeTags are the VS Code extension's IDE-context +// wrapper tags: standalone messages using these are already +// promoted to hidden system metadata by classifyClaudeSystemMessage. +// splitLeadingClaudeIDEEnvelope handles the remaining case where the +// extension prepends one of these wrappers directly onto a real +// prompt in the same user entry. +var claudeIDEEnvelopeTags = [...]string{"ide_opened_file", "ide_selection"} + +// claudeIDEEnvelopeSourceUUID derives a distinct source UUID for the +// synthetic hidden message a split envelope becomes. The real prompt +// keeps the entry's own uuid: pins and Recall evidence resolve source +// UUIDs and require them to be unique per session, so the two rows +// produced from one entry must not share an identity. An empty entry +// uuid stays empty rather than becoming a shared non-empty value. +func claudeIDEEnvelopeSourceUUID(entryUUID string) string { + if entryUUID == "" { + return "" + } + return entryUUID + ":ide-context" +} + +// splitLeadingClaudeIDEEnvelope detects a well-formed IDE-context +// envelope at the very start of content that is followed by +// additional real prompt text, and separates the two. The standalone +// case (envelope with nothing else) is left alone here; that is +// handled by classifyClaudeSystemMessage so the whole message +// promotes to system metadata. +// +// Splitting keeps the envelope recorded as hidden system metadata +// (same subtype as the standalone case) while letting first_message +// and the visible transcript show only the real prompt that follows, +// instead of raw IDE-context markup. +// splitClaudeIDEEnvelopePrompt splits a leading IDE-context envelope +// off a user entry's text and re-runs the revealed remainder through +// the same command/system-reminder preprocessing a bare prompt gets, +// so command XML normalizes (e.g. "/clear") instead of surfacing raw +// markup in first_message. A remainder that preprocessing skips or +// empties comes back as "": the entry carries no visible prompt. +func splitClaudeIDEEnvelopePrompt( + text string, +) (subtype, envelope, remainder string, ok bool) { + subtype, envelope, remainder, ok = splitLeadingClaudeIDEEnvelope(text) + if !ok { + return "", "", "", false + } + remainder, skip := preprocessClaudeUserText(remainder) + if skip || strings.TrimSpace(remainder) == "" { + return subtype, envelope, "", true + } + return subtype, envelope, remainder, true +} + +// claudeIDEEnvelopeMessage builds the hidden system-metadata message a +// split envelope becomes. Role stays "user" so role-keyed analytics +// treat it as input, matching the standalone classify branch. +func claudeIDEEnvelopeMessage( + e dagEntry, ordinal int, subtype, envelope string, +) ParsedMessage { + return ParsedMessage{ + Ordinal: ordinal, + Role: RoleUser, + Content: envelope, + Timestamp: e.timestamp, + IsSystem: true, + ContentLength: len(envelope), + SourceType: "system", + SourceSubtype: subtype, + SourceUUID: claudeIDEEnvelopeSourceUUID(e.uuid), + SourceParentUUID: e.parentUuid, + IsSidechain: gjson.Get(e.line, "isSidechain").Bool(), + } +} + +func splitLeadingClaudeIDEEnvelope( + content string, +) (subtype, envelope, remainder string, ok bool) { + trimmed := trimClaudeSystemMessagePrefix(content) + for _, tag := range claudeIDEEnvelopeTags { + openTag := "<" + tag + ">" + closeTag := "" + if !strings.HasPrefix(trimmed, openTag) { + continue + } + closeIdx := strings.Index(trimmed, closeTag) + if closeIdx < 0 { + continue + } + envelopeEnd := closeIdx + len(closeTag) + rest := strings.TrimSpace(trimmed[envelopeEnd:]) + if rest == "" { + // Standalone: classifyClaudeSystemMessage handles this. + continue + } + return tag, trimmed[:envelopeEnd], rest, true + } + return "", "", "", false +} + func stripLeadingClaudeSystemReminderContent(content string) string { trimmed := trimClaudeSystemMessagePrefix(content) remainder, stripped := stripLeadingClaudeSystemReminderBlocks(trimmed) diff --git a/internal/parser/claude_parser_test.go b/internal/parser/claude_parser_test.go index 531ebb168..571659861 100644 --- a/internal/parser/claude_parser_test.go +++ b/internal/parser/claude_parser_test.go @@ -324,6 +324,148 @@ func TestParseClaudeSession_SkippedMessages(t *testing.T) { assert.Equal(t, "real user message", msgs[7].Content) }) + t.Run("splits IDE envelope prepended onto a real prompt", func(t *testing.T) { + content := testjsonl.JoinJSONL( + testjsonl.ClaudeUserJSON( + "The user opened /workspace/app/README.md. Explain this file.", + tsZero, + ), + testjsonl.ClaudeUserJSON( + "The user selected package main.\n\nWhat does this do?", + tsZeroS1, + ), + ) + sess, msgs := runClaudeParserTest(t, "test.jsonl", content) + // Each entry splits into a hidden system-metadata message + // plus the real prompt that followed it. + require.Len(t, msgs, 4) + assert.Equal(t, 4, sess.MessageCount) + assert.Equal(t, 2, sess.UserMessageCount) + assert.Equal(t, "Explain this file.", sess.FirstMessage, + "first_message should show the real prompt, not the IDE envelope") + + assert.True(t, msgs[0].IsSystem) + assert.Equal(t, RoleUser, msgs[0].Role) + assert.Equal(t, "system", msgs[0].SourceType) + assert.Equal(t, "ide_opened_file", msgs[0].SourceSubtype) + assert.Equal(t, + "The user opened /workspace/app/README.md.", + msgs[0].Content) + + assert.False(t, msgs[1].IsSystem) + assert.Equal(t, RoleUser, msgs[1].Role) + assert.Equal(t, "Explain this file.", msgs[1].Content) + + assert.True(t, msgs[2].IsSystem) + assert.Equal(t, "ide_selection", msgs[2].SourceSubtype) + assert.Equal(t, + "The user selected package main.", + msgs[2].Content) + + assert.False(t, msgs[3].IsSystem) + assert.Equal(t, "What does this do?", msgs[3].Content) + }) + + t.Run("split IDE envelope remainder gets command preprocessing", func(t *testing.T) { + content := testjsonl.JoinJSONL( + testjsonl.ClaudeUserJSON( + "The user opened /workspace/app/README.md.\n"+ + "clear\n"+ + "/clear", + tsZero, + ), + testjsonl.ClaudeUserJSON("real question", tsZeroS1), + ) + sess, msgs := runClaudeParserTest(t, "test.jsonl", content) + // The command XML revealed by the split normalizes exactly + // like a bare command message instead of being stored as + // raw markup. + require.Len(t, msgs, 3) + assert.True(t, msgs[0].IsSystem) + assert.Equal(t, "ide_opened_file", msgs[0].SourceSubtype) + assert.False(t, msgs[1].IsSystem) + assert.Equal(t, "/clear", msgs[1].Content) + assert.Equal(t, "real question", msgs[2].Content) + assert.Equal(t, "real question", sess.FirstMessage, + "slash command must not become first_message") + }) + + t.Run("split IDE envelope remainder honors preprocessing skip", func(t *testing.T) { + content := testjsonl.JoinJSONL( + testjsonl.ClaudeUserJSON( + "The user selected package main.\n"+ + "", + tsZero, + ), + testjsonl.ClaudeUserJSON("real question", tsZeroS1), + ) + sess, msgs := runClaudeParserTest(t, "test.jsonl", content) + // The remainder is a command envelope that cannot be + // normalized: it is skipped like a standalone one, keeping + // only the hidden envelope message. + require.Len(t, msgs, 2) + assert.True(t, msgs[0].IsSystem) + assert.Equal(t, "ide_selection", msgs[0].SourceSubtype) + assert.Equal(t, "real question", msgs[1].Content) + assert.Equal(t, "real question", sess.FirstMessage) + }) + + t.Run("split IDE envelope with discarded remainder keeps tool results", func(t *testing.T) { + mixedEntry := `{"type":"user","timestamp":"` + tsZero + + `","message":{"content":[` + + `{"type":"tool_result","tool_use_id":"tu-1","content":"tool output"},` + + `{"type":"text","text":"The user selected package main.\n` + + `"}]}}` + content := testjsonl.JoinJSONL( + mixedEntry, + testjsonl.ClaudeUserJSON("real question", tsZeroS1), + ) + + sess, msgs := runClaudeParserTest(t, "test.jsonl", content) + // The discarded command remainder leaves no visible message + // for the entry, so the tool result must ride the hidden + // envelope row instead of vanishing. + require.Len(t, msgs, 2) + assert.True(t, msgs[0].IsSystem) + assert.Equal(t, "ide_selection", msgs[0].SourceSubtype) + require.Len(t, msgs[0].ToolResults, 1) + assert.Equal(t, "tu-1", msgs[0].ToolResults[0].ToolUseID) + assert.Equal(t, "real question", sess.FirstMessage) + + // The incremental path must preserve the same tool result. + path := createTestFile(t, "incremental.jsonl", content) + newMsgs, _, _, err := callParseClaudeSessionFrom(path, 0, 0, "") + require.NoError(t, err) + require.Len(t, newMsgs, 2) + assert.True(t, newMsgs[0].IsSystem) + require.Len(t, newMsgs[0].ToolResults, 1) + assert.Equal(t, "tu-1", newMsgs[0].ToolResults[0].ToolUseID) + }) + + t.Run("split IDE envelope gets a distinct source uuid", func(t *testing.T) { + content := testjsonl.JoinJSONL( + testjsonl.ClaudeEntryJSON( + "user", + "The user opened /workspace/app/README.md. Explain this file.", + tsZero, "uuid-entry-1", "uuid-parent-0", + ), + ) + _, msgs := runClaudeParserTest(t, "test.jsonl", content) + require.Len(t, msgs, 2) + + // Pins and Recall evidence resolve messages by source_uuid and + // require it to be unique per session, so the entry's own uuid + // must stay on the real prompt only; the synthetic hidden + // envelope row gets a derived identity. + assert.True(t, msgs[0].IsSystem) + assert.Equal(t, "uuid-entry-1:ide-context", msgs[0].SourceUUID) + assert.Equal(t, "uuid-parent-0", msgs[0].SourceParentUUID) + + assert.False(t, msgs[1].IsSystem) + assert.Equal(t, "uuid-entry-1", msgs[1].SourceUUID) + assert.Equal(t, "uuid-parent-0", msgs[1].SourceParentUUID) + }) + t.Run("skill invocation shown as user message", func(t *testing.T) { content := testjsonl.JoinJSONL( testjsonl.ClaudeUserJSON( @@ -886,6 +1028,85 @@ func TestParseClaudeSessionFrom_IDEContext(t *testing.T) { } } +func TestParseClaudeSessionFrom_IDEContextPrependedToPrompt(t *testing.T) { + t.Parallel() + + initial := testjsonl.JoinJSONL( + testjsonl.ClaudeUserJSON("hello", tsEarly), + testjsonl.ClaudeAssistantJSON("hi", tsEarlyS1), + ) + path := createTestFile(t, "inc-ide-context-prompt.jsonl", initial) + info, err := os.Stat(path) + require.NoError(t, err) + + appended := testjsonl.JoinJSONL( + testjsonl.ClaudeUserJSON( + "The user opened /workspace/app/README.md. Explain this file.", + tsLate, + ), + ) + f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0o644) + require.NoError(t, err) + _, err = f.WriteString(appended) + require.NoError(t, err) + require.NoError(t, f.Close()) + + newMsgs, _, _, err := callParseClaudeSessionFrom(path, info.Size(), 2, "") + require.NoError(t, err) + require.Len(t, newMsgs, 2, + "the entry splits into a hidden IDE-context message plus the real prompt") + + assert.True(t, newMsgs[0].IsSystem) + assert.Equal(t, "system", newMsgs[0].SourceType) + assert.Equal(t, "ide_opened_file", newMsgs[0].SourceSubtype) + assert.Equal(t, + "The user opened /workspace/app/README.md.", + newMsgs[0].Content) + + assert.False(t, newMsgs[1].IsSystem) + assert.Equal(t, RoleUser, newMsgs[1].Role) + assert.Equal(t, "Explain this file.", newMsgs[1].Content) +} + +func TestParseClaudeSessionFrom_IDEContextPrependedToCommand(t *testing.T) { + t.Parallel() + + initial := testjsonl.JoinJSONL( + testjsonl.ClaudeUserJSON("hello", tsEarly), + testjsonl.ClaudeAssistantJSON("hi", tsEarlyS1), + ) + path := createTestFile(t, "inc-ide-context-command.jsonl", initial) + info, err := os.Stat(path) + require.NoError(t, err) + + appended := testjsonl.JoinJSONL( + testjsonl.ClaudeUserJSON( + "The user opened /workspace/app/README.md.\n"+ + "clear\n"+ + "/clear", + tsLate, + ), + ) + f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0o644) + require.NoError(t, err) + _, err = f.WriteString(appended) + require.NoError(t, err) + require.NoError(t, f.Close()) + + newMsgs, _, _, err := callParseClaudeSessionFrom(path, info.Size(), 2, "") + require.NoError(t, err) + require.Len(t, newMsgs, 2, + "the entry splits into a hidden IDE-context message plus the normalized command") + + assert.True(t, newMsgs[0].IsSystem) + assert.Equal(t, "ide_opened_file", newMsgs[0].SourceSubtype) + + assert.False(t, newMsgs[1].IsSystem) + assert.Equal(t, RoleUser, newMsgs[1].Role) + assert.Equal(t, "/clear", newMsgs[1].Content, + "command XML revealed by the split must normalize, not stay raw") +} + func TestParseClaudeSessionFrom_ReminderPrefixedCommand(t *testing.T) { t.Parallel() diff --git a/internal/parser/fork_test.go b/internal/parser/fork_test.go index 9a45904fb..b79fecfef 100644 --- a/internal/parser/fork_test.go +++ b/internal/parser/fork_test.go @@ -158,6 +158,37 @@ func TestForkDetection_ReminderPrefixedIDEContextDoesNotPromoteObsoleteBranch( assert.Equal(t, "retry answer", results[0].Messages[3].Content) } +func TestForkDetection_IDEEnvelopeWithDiscardedRemainderDoesNotPromoteObsoleteBranch( + t *testing.T, +) { + const discardedRemainder = "context\n" + + "" + content := testjsonl.NewSessionBuilder(). + AddClaudeUserWithUUID("2024-01-01T10:00:00Z", "start", "a", ""). + AddClaudeAssistantWithUUID("2024-01-01T10:00:01Z", "ok", "b", "a"). + AddClaudeUserWithUUID("2024-01-01T10:00:02Z", "q1", "c", "b"). + AddClaudeAssistantWithUUID("2024-01-01T10:00:03Z", "a1", "d", "c"). + AddClaudeUserWithUUID("2024-01-01T10:00:04Z", "q2", "e", "d"). + AddClaudeAssistantWithUUID("2024-01-01T10:00:05Z", "a2", "f", "e"). + AddClaudeUserWithUUID("2024-01-01T10:00:06Z", "q3", "g", "f"). + AddClaudeAssistantWithUUID("2024-01-01T10:00:07Z", "a3", "h", "g"). + AddClaudeUserWithUUID( + "2024-01-01T10:00:08Z", discardedRemainder, "i", "h", + ). + AddClaudeUserWithUUID( + "2024-01-01T10:01:00Z", "real retry", "z", "b", + ). + AddClaudeAssistantWithUUID( + "2024-01-01T10:01:01Z", "retry answer", "zz", "z", + ). + String() + + results := parseTestContent(t, "ide-command-fork.jsonl", content, 1) + require.Len(t, results[0].Messages, 4) + assert.Equal(t, "real retry", results[0].Messages[2].Content) + assert.Equal(t, "retry answer", results[0].Messages[3].Content) +} + func TestForkDetection_NoUUIDs(t *testing.T) { // Entries without uuid fields — should work as before, 1 result. content := testjsonl.NewSessionBuilder(). diff --git a/internal/postgres/curation.go b/internal/postgres/curation.go index 9d1b96ebf..516503817 100644 --- a/internal/postgres/curation.go +++ b/internal/postgres/curation.go @@ -9,6 +9,28 @@ import ( "go.kenn.io/agentsview/internal/db" ) +func lockPinnedMessagesSession( + ctx context.Context, tx *sql.Tx, sessionID string, +) error { + var lockedSessionID string + err := tx.QueryRowContext(ctx, ` + SELECT id + FROM sessions + WHERE id = $1 + FOR UPDATE`, + sessionID, + ).Scan(&lockedSessionID) + if err == sql.ErrNoRows { + return nil + } + if err != nil { + return fmt.Errorf( + "locking pg pins for session %s: %w", sessionID, err, + ) + } + return nil +} + // StarSession marks a session as starred in the shared PG dashboard // metadata. Returns false when the session does not exist. func (s *Store) StarSession(sessionID string) (bool, error) { @@ -119,8 +141,18 @@ func (s *Store) BulkStarSessions(sessionIDs []string) error { func (s *Store) PinMessage( sessionID string, messageID int64, note *string, ) (int64, error) { + ctx := context.Background() + tx, err := s.pg.BeginTx(ctx, nil) + if err != nil { + return 0, fmt.Errorf("beginning pin transaction: %w", err) + } + defer func() { _ = tx.Rollback() }() + if err := lockPinnedMessagesSession(ctx, tx, sessionID); err != nil { + return 0, err + } + var id int64 - err := s.pg.QueryRow(` + err = tx.QueryRowContext(ctx, ` WITH upsert AS ( INSERT INTO pinned_messages ( session_id, message_id, ordinal, source_uuid, note @@ -140,23 +172,41 @@ func (s *Store) PinMessage( sessionID, messageID, note, ).Scan(&id) if err == sql.ErrNoRows { + if err := tx.Commit(); err != nil { + return 0, fmt.Errorf("committing empty pin transaction: %w", err) + } return 0, nil } if err != nil { return 0, fmt.Errorf("pinning message: %w", err) } + if err := tx.Commit(); err != nil { + return 0, fmt.Errorf("committing pin transaction: %w", err) + } return id, nil } // UnpinMessage removes a shared PG pin. func (s *Store) UnpinMessage(sessionID string, messageID int64) error { - if _, err := s.pg.Exec( + ctx := context.Background() + tx, err := s.pg.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("beginning unpin transaction: %w", err) + } + defer func() { _ = tx.Rollback() }() + if err := lockPinnedMessagesSession(ctx, tx, sessionID); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `DELETE FROM pinned_messages WHERE session_id = $1 AND message_id = $2`, sessionID, messageID, ); err != nil { return fmt.Errorf("unpinning message: %w", err) } + if err := tx.Commit(); err != nil { + return fmt.Errorf("committing unpin transaction: %w", err) + } return nil } diff --git a/internal/postgres/curation_pgtest_test.go b/internal/postgres/curation_pgtest_test.go index 784924c3a..df1e6a391 100644 --- a/internal/postgres/curation_pgtest_test.go +++ b/internal/postgres/curation_pgtest_test.go @@ -4,6 +4,7 @@ package postgres import ( "context" + "database/sql" "testing" "github.com/stretchr/testify/assert" @@ -12,6 +13,16 @@ import ( "go.kenn.io/agentsview/internal/db" ) +func reconcilePinnedMessages( + ctx context.Context, tx *sql.Tx, sessionID string, +) error { + pins, err := snapshotPinnedMessages(ctx, tx, sessionID) + if err != nil { + return err + } + return restorePinnedMessages(ctx, tx, sessionID, pins) +} + func TestStoreStarsAndPins(t *testing.T) { pgURL := testPGURL(t) @@ -247,6 +258,406 @@ func TestPushPreservesMultiplePGPinsBySourceUUID(t *testing.T) { assert.Equal(t, 3, pin.Ordinal) } +// TestPushDropsEditedLegacyPinInBothStores documents the intended +// limit for UUID-less pins: a replacement that edits the pinned +// message destroys the only identity the pin can follow, so both the +// local SQLite archive and the next PostgreSQL push drop the pin — +// the stores stay consistent instead of diverging on heuristics. +func TestPushDropsEditedLegacyPinInBothStores(t *testing.T) { + pgURL := testPGURL(t) + cleanPGSchema(t, pgURL) + t.Cleanup(func() { cleanPGSchema(t, pgURL) }) + + local := testDB(t) + ps, err := New( + pgURL, "agentsview", local, + "curation-machine", true, + SyncOptions{}, + ) + require.NoError(t, err, "New sync") + defer ps.Close() + + ctx := context.Background() + require.NoError(t, ps.EnsureSchema(ctx), "EnsureSchema") + + seed := func(sessionID string) db.Session { + sess := db.Session{ + ID: sessionID, + Project: "proj-curation", + Machine: "local", + Agent: "claude", + MessageCount: 2, + CreatedAt: "2026-05-01T00:00:00Z", + } + require.NoError(t, local.UpsertSession(sess), + "UpsertSession %s", sessionID) + require.NoError(t, local.InsertMessages([]db.Message{ + { + SessionID: sessionID, Ordinal: 0, + Role: "user", Content: "question", + }, + { + SessionID: sessionID, Ordinal: 1, + Role: "assistant", Content: "draft answer", + }, + }), "InsertMessages %s", sessionID) + return sess + } + uploadSess := seed("pg-pin-upload-edit") + seed("pg-pin-reparse-edit") + _, err = ps.Push(ctx, false, nil) + require.NoError(t, err, "Push first") + + store, err := NewStore(pgURL, "agentsview", true) + require.NoError(t, err, "NewStore") + defer store.Close() + + pinBoth := func(sessionID string) { + msgs, err := local.GetAllMessages(ctx, sessionID) + require.NoError(t, err, "GetAllMessages %s", sessionID) + require.Len(t, msgs, 2, "seeded messages %s", sessionID) + _, err = local.PinMessage(sessionID, msgs[1].ID, nil) + require.NoError(t, err, "local PinMessage %s", sessionID) + note := "keep " + sessionID + _, err = store.PinMessage(sessionID, 1, ¬e) + require.NoError(t, err, "pg PinMessage %s", sessionID) + } + pinBoth("pg-pin-upload-edit") + pinBoth("pg-pin-reparse-edit") + + // Both replacement entry points edit the pinned message. + edited := func(sessionID string) []db.Message { + return []db.Message{ + { + SessionID: sessionID, Ordinal: 0, + Role: "user", Content: "question", + }, + { + SessionID: sessionID, Ordinal: 1, + Role: "assistant", Content: "edited answer", + }, + } + } + _, err = local.WriteSessionBatch([]db.SessionBatchWrite{{ + Session: uploadSess, + Messages: edited("pg-pin-upload-edit"), + DataVersion: db.CurrentDataVersion(), + ReplaceMessages: true, + }}) + require.NoError(t, err, "explicit re-upload") + require.NoError(t, local.ReplaceSessionMessages( + "pg-pin-reparse-edit", edited("pg-pin-reparse-edit"), + ), "reparse replacement") + + for _, sessionID := range []string{ + "pg-pin-upload-edit", "pg-pin-reparse-edit", + } { + localPins, err := local.ListPinnedMessages(ctx, sessionID, "") + require.NoError(t, err, "local pins %s", sessionID) + require.Empty(t, localPins, + "editing the pinned message drops the local pin (%s)", + sessionID) + } + + _, err = ps.Push(ctx, true, nil) + require.NoError(t, err, "Push rewrite") + + for _, sessionID := range []string{ + "pg-pin-upload-edit", "pg-pin-reparse-edit", + } { + pins, err := store.ListPinnedMessages(ctx, sessionID, "") + require.NoError(t, err, "pg pins %s", sessionID) + assert.Empty(t, pins, + "push must mirror the local drop (%s)", sessionID) + } +} + +func TestPushReconcilesPGPinsByPriorMessageIdentity(t *testing.T) { + pgURL := testPGURL(t) + + tests := []struct { + name string + oldMessages []db.Message + newMessages []db.Message + wantPin bool + wantOrdinal int + }{ + { + name: "duplicate UUID target removed", + oldMessages: []db.Message{ + { + Ordinal: 0, Role: "user", Content: "pinned", + SourceUUID: "duplicate", + }, + { + Ordinal: 1, Role: "assistant", Content: "other", + SourceUUID: "duplicate", + }, + { + Ordinal: 2, Role: "user", Content: "tail", + SourceUUID: "tail", + }, + }, + newMessages: []db.Message{ + { + Ordinal: 0, Role: "assistant", Content: "other", + SourceUUID: "duplicate", + }, + { + Ordinal: 1, Role: "user", Content: "tail", + SourceUUID: "tail", + }, + }, + }, + { + name: "UUID-less prompt split by IDE envelope", + oldMessages: []db.Message{ + { + Ordinal: 0, Role: "user", + Content: "ctxprompt", + }, + { + Ordinal: 1, Role: "assistant", Content: "answer", + SourceUUID: "answer", + }, + }, + newMessages: []db.Message{ + { + Ordinal: 0, Role: "user", + Content: "ctx", + SourceUUID: "entry:ide-context", IsSystem: true, + }, + { + Ordinal: 1, Role: "user", Content: "prompt", + SourceUUID: "entry", + }, + { + Ordinal: 2, Role: "assistant", Content: "answer", + SourceUUID: "answer", + }, + }, + }, + { + name: "UUID enrichment", + oldMessages: []db.Message{ + {Ordinal: 0, Role: "user", Content: "pinned"}, + {Ordinal: 1, Role: "assistant", Content: "answer"}, + }, + newMessages: []db.Message{ + { + Ordinal: 0, Role: "user", Content: "pinned", + SourceUUID: "new-provider-uuid", + }, + {Ordinal: 1, Role: "assistant", Content: "answer"}, + }, + wantPin: true, + wantOrdinal: 0, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cleanPGSchema(t, pgURL) + t.Cleanup(func() { cleanPGSchema(t, pgURL) }) + + local := testDB(t) + ps, err := New( + pgURL, "agentsview", local, + "curation-machine", true, SyncOptions{}, + ) + require.NoError(t, err, "New sync") + defer ps.Close() + + ctx := context.Background() + require.NoError(t, ps.EnsureSchema(ctx), "EnsureSchema") + + const sessionID = "pg-pin-prior-identity" + sess := db.Session{ + ID: sessionID, Project: "proj-curation", + Machine: "local", Agent: "claude", + MessageCount: len(tt.oldMessages), + CreatedAt: "2026-05-01T00:00:00Z", + } + require.NoError(t, local.UpsertSession(sess), "UpsertSession old") + oldMessages := append([]db.Message(nil), tt.oldMessages...) + for i := range oldMessages { + oldMessages[i].SessionID = sessionID + } + require.NoError(t, local.InsertMessages(oldMessages), + "InsertMessages old") + _, err = ps.Push(ctx, false, nil) + require.NoError(t, err, "Push old") + + store, err := NewStore(pgURL, "agentsview", true) + require.NoError(t, err, "NewStore") + defer store.Close() + _, err = store.PinMessage(sessionID, 0, nil) + require.NoError(t, err, "PinMessage") + + newMessages := append([]db.Message(nil), tt.newMessages...) + for i := range newMessages { + newMessages[i].SessionID = sessionID + } + sess.MessageCount = len(newMessages) + require.NoError(t, local.UpsertSession(sess), "UpsertSession new") + require.NoError(t, + local.ReplaceSessionMessages(sessionID, newMessages), + "ReplaceSessionMessages") + _, err = ps.Push(ctx, true, nil) + require.NoError(t, err, "Push new") + + pins, err := store.ListPinnedMessages(ctx, sessionID, "") + require.NoError(t, err, "ListPinnedMessages") + if !tt.wantPin { + assert.Empty(t, pins, + "an ambiguous old identity must not retain a pin") + return + } + require.Len(t, pins, 1, "pins = %v", pins) + assert.Equal(t, tt.wantOrdinal, pins[0].Ordinal) + }) + } +} + +func TestRestorePinnedMessagesPreservesPinCreatedAfterSnapshot(t *testing.T) { + pgURL := testPGURL(t) + + const schema = "agentsview_pin_snapshot_race_test" + pg, err := Open(pgURL, schema, true) + require.NoError(t, err, "Open") + defer pg.Close() + defer func() { + _, _ = pg.ExecContext( + context.Background(), + `DROP SCHEMA IF EXISTS `+schema+` CASCADE`, + ) + }() + + ctx := context.Background() + _, err = pg.ExecContext(ctx, `DROP SCHEMA IF EXISTS `+schema+` CASCADE`) + require.NoError(t, err, "drop schema") + require.NoError(t, EnsureSchema(ctx, pg, schema), "EnsureSchema") + + _, err = pg.ExecContext(ctx, ` + INSERT INTO sessions + (id, machine, project, agent, first_message, + started_at, message_count, user_message_count) + VALUES + ('pg-pin-snapshot-race', 'machine-a', 'proj-curation', + 'codex', 'snapshot race', + '2026-05-01T00:00:00Z'::timestamptz, 2, 1); + INSERT INTO messages + (session_id, ordinal, role, content, timestamp, + content_length, source_uuid) + VALUES + ('pg-pin-snapshot-race', 0, 'user', 'first', + '2026-05-01T00:00:00Z'::timestamptz, 5, 'uuid-first'), + ('pg-pin-snapshot-race', 1, 'assistant', 'second', + '2026-05-01T00:00:01Z'::timestamptz, 6, 'uuid-second'); + INSERT INTO pinned_messages + (session_id, message_id, ordinal, source_uuid, note) + VALUES + ('pg-pin-snapshot-race', 0, 0, 'uuid-first', 'old pin')`) + require.NoError(t, err, "seed session and old pin") + + tx, err := pg.BeginTx(ctx, nil) + require.NoError(t, err, "begin tx") + pins, err := snapshotPinnedMessages(ctx, tx, "pg-pin-snapshot-race") + if err != nil { + _ = tx.Rollback() + t.Fatalf("snapshotPinnedMessages: %v", err) + } + _, err = tx.ExecContext(ctx, ` + INSERT INTO pinned_messages + (session_id, message_id, ordinal, source_uuid, note) + VALUES + ('pg-pin-snapshot-race', 1, 1, 'uuid-second', 'new pin')`) + if err != nil { + _ = tx.Rollback() + t.Fatalf("insert post-snapshot pin: %v", err) + } + if err := restorePinnedMessages( + ctx, tx, "pg-pin-snapshot-race", pins, + ); err != nil { + _ = tx.Rollback() + t.Fatalf("restorePinnedMessages: %v", err) + } + require.NoError(t, tx.Commit(), "commit tx") + + store, err := NewStore(pgURL, schema, true) + require.NoError(t, err, "NewStore") + defer store.Close() + got, err := store.ListPinnedMessages(ctx, "pg-pin-snapshot-race", "") + require.NoError(t, err, "ListPinnedMessages") + require.Len(t, got, 2, "post-snapshot pin must survive: %v", got) + byOrdinal := make(map[int]db.PinnedMessage, len(got)) + for _, pin := range got { + byOrdinal[pin.Ordinal] = pin + } + assert.Contains(t, byOrdinal, 0, "snapshotted pin restored") + assert.Contains(t, byOrdinal, 1, "post-snapshot pin preserved") +} + +func TestPinMessageSerializesWithSessionReplacement(t *testing.T) { + pgURL := testPGURL(t) + + const schema = "agentsview_pin_session_lock_test" + pg, err := Open(pgURL, schema, true) + require.NoError(t, err, "Open") + defer pg.Close() + defer func() { + _, _ = pg.ExecContext( + context.Background(), + `DROP SCHEMA IF EXISTS `+schema+` CASCADE`, + ) + }() + + ctx := context.Background() + _, err = pg.ExecContext(ctx, `DROP SCHEMA IF EXISTS `+schema+` CASCADE`) + require.NoError(t, err, "drop schema") + require.NoError(t, EnsureSchema(ctx, pg, schema), "EnsureSchema") + _, err = pg.ExecContext(ctx, ` + INSERT INTO sessions + (id, machine, project, agent, first_message, + started_at, message_count, user_message_count) + VALUES + ('pg-pin-session-lock', 'machine-a', 'proj-curation', + 'codex', 'session lock', + '2026-05-01T00:00:00Z'::timestamptz, 1, 1); + INSERT INTO messages + (session_id, ordinal, role, content, timestamp, + content_length, source_uuid) + VALUES + ('pg-pin-session-lock', 0, 'user', 'message', + '2026-05-01T00:00:00Z'::timestamptz, 7, 'uuid-message')`) + require.NoError(t, err, "seed session") + + store, err := NewStore(pgURL, schema, true) + require.NoError(t, err, "NewStore") + defer store.Close() + store.pg.SetMaxOpenConns(1) + _, err = store.pg.ExecContext(ctx, `SET lock_timeout = '50ms'`) + require.NoError(t, err, "set lock timeout") + + lockTx, err := pg.BeginTx(ctx, nil) + require.NoError(t, err, "begin lock tx") + require.NoError(t, + lockPinnedMessagesSession(ctx, lockTx, "pg-pin-session-lock"), + "lock session pins") + + _, err = store.PinMessage("pg-pin-session-lock", 0, nil) + require.Error(t, err, + "pin mutation must wait while replacement owns the session lock") + assert.ErrorContains(t, err, "locking pg pins for session") + require.NoError(t, lockTx.Rollback(), "release session lock") + + _, err = store.pg.ExecContext(ctx, `SET lock_timeout = 0`) + require.NoError(t, err, "clear lock timeout") + pinID, err := store.PinMessage("pg-pin-session-lock", 0, nil) + require.NoError(t, err, "PinMessage after replacement lock") + assert.NotZero(t, pinID, "PinMessage after replacement lock") +} + func TestReconcilePinnedMessagesPrefersCurrentTargetPin(t *testing.T) { pgURL := testPGURL(t) @@ -297,7 +708,7 @@ func TestReconcilePinnedMessagesPrefersCurrentTargetPin(t *testing.T) { VALUES ('pg-pin-duplicate', 1, 1, 'uuid-answer', 'stale note', - '2026-05-01T00:01:00Z'::timestamptz), + '2026-05-01T00:03:00Z'::timestamptz), ('pg-pin-duplicate', 2, 2, 'uuid-answer', 'current note', '2026-05-01T00:02:00Z'::timestamptz)`) @@ -326,6 +737,404 @@ func TestReconcilePinnedMessagesPrefersCurrentTargetPin(t *testing.T) { assert.Equal(t, "current note", *pins[0].Note) } +func TestReconcilePinnedMessagesFollowsStoredUniqueSourceUUID(t *testing.T) { + pgURL := testPGURL(t) + + const schema = "agentsview_pin_shifted_uuid_test" + pg, err := Open(pgURL, schema, true) + require.NoError(t, err, "Open") + defer pg.Close() + defer func() { + _, _ = pg.ExecContext( + context.Background(), + `DROP SCHEMA IF EXISTS `+schema+` CASCADE`, + ) + }() + + ctx := context.Background() + _, err = pg.ExecContext(ctx, `DROP SCHEMA IF EXISTS `+schema+` CASCADE`) + require.NoError(t, err, "drop schema") + require.NoError(t, EnsureSchema(ctx, pg, schema), "EnsureSchema") + + _, err = pg.ExecContext(ctx, ` + INSERT INTO sessions + (id, machine, project, agent, first_message, + started_at, message_count, user_message_count) + VALUES + ('pg-pin-shifted-uuid', 'machine-a', 'proj-curation', + 'claude', 'shifted source uuid', + '2026-05-01T00:00:00Z'::timestamptz, 2, 1); + INSERT INTO messages + (session_id, ordinal, role, content, timestamp, + content_length, source_uuid) + VALUES + ('pg-pin-shifted-uuid', 0, 'user', '[context]', + '2026-05-01T00:00:00Z'::timestamptz, 9, + 'uuid-context'), + ('pg-pin-shifted-uuid', 1, 'user', 'question', + '2026-05-01T00:00:01Z'::timestamptz, 8, + 'uuid-question'); + INSERT INTO pinned_messages + (session_id, message_id, ordinal, source_uuid, + note, created_at) + VALUES + ('pg-pin-shifted-uuid', 0, 0, 'uuid-question', + 'keep shifted pin', + '2026-05-01T00:01:00Z'::timestamptz)`) + require.NoError(t, err, "seed shifted pin") + + tx, err := pg.BeginTx(ctx, nil) + require.NoError(t, err, "begin tx") + if err := reconcilePinnedMessages( + ctx, tx, "pg-pin-shifted-uuid", + ); err != nil { + _ = tx.Rollback() + t.Fatalf("reconcilePinnedMessages: %v", err) + } + require.NoError(t, tx.Commit(), "commit tx") + + store, err := NewStore(pgURL, schema, true) + require.NoError(t, err, "NewStore") + defer store.Close() + + pins, err := store.ListPinnedMessages(ctx, "pg-pin-shifted-uuid", "") + require.NoError(t, err, "ListPinnedMessages") + require.Len(t, pins, 1, "pins = %v", pins) + assert.Equal(t, int64(1), pins[0].MessageID) + assert.Equal(t, 1, pins[0].Ordinal) + require.NotNil(t, pins[0].Note) + assert.Equal(t, "keep shifted pin", *pins[0].Note) +} + +func TestRestorePinnedMessagesUsesResolvedAnchorOrdinalForNewDuplicateUUID(t *testing.T) { + pgURL := testPGURL(t) + + const schema = "agentsview_pin_shifted_duplicate_test" + pg, err := Open(pgURL, schema, true) + require.NoError(t, err, "Open") + defer pg.Close() + defer func() { + _, _ = pg.ExecContext( + context.Background(), + `DROP SCHEMA IF EXISTS `+schema+` CASCADE`, + ) + }() + + ctx := context.Background() + _, err = pg.ExecContext(ctx, `DROP SCHEMA IF EXISTS `+schema+` CASCADE`) + require.NoError(t, err, "drop schema") + require.NoError(t, EnsureSchema(ctx, pg, schema), "EnsureSchema") + + _, err = pg.ExecContext(ctx, ` + INSERT INTO sessions + (id, machine, project, agent, first_message, + started_at, message_count, user_message_count) + VALUES + ('pg-pin-shifted-duplicate', 'machine-a', 'proj-curation', + 'claude', 'shifted source becomes duplicate', + '2026-05-01T00:00:00Z'::timestamptz, 2, 1); + INSERT INTO messages + (session_id, ordinal, role, content, timestamp, + content_length, source_uuid) + VALUES + ('pg-pin-shifted-duplicate', 0, 'user', '[context]', + '2026-05-01T00:00:00Z'::timestamptz, 9, + 'uuid-context'), + ('pg-pin-shifted-duplicate', 1, 'assistant', 'answer', + '2026-05-01T00:00:01Z'::timestamptz, 6, + 'uuid-answer'); + INSERT INTO pinned_messages + (session_id, message_id, ordinal, source_uuid, + note, created_at) + VALUES + ('pg-pin-shifted-duplicate', 0, 0, 'uuid-answer', + 'keep shifted duplicate pin', + '2026-05-01T00:01:00Z'::timestamptz)`) + require.NoError(t, err, "seed shifted pin") + + tx, err := pg.BeginTx(ctx, nil) + require.NoError(t, err, "begin tx") + pins, err := snapshotPinnedMessages( + ctx, tx, "pg-pin-shifted-duplicate", + ) + if err != nil { + _ = tx.Rollback() + t.Fatalf("snapshotPinnedMessages: %v", err) + } + _, err = tx.ExecContext(ctx, ` + DELETE FROM messages + WHERE session_id = 'pg-pin-shifted-duplicate'; + INSERT INTO messages + (session_id, ordinal, role, content, timestamp, + content_length, source_uuid) + VALUES + ('pg-pin-shifted-duplicate', 0, 'assistant', 'retry', + '2026-05-01T00:00:00Z'::timestamptz, 5, + 'uuid-answer'), + ('pg-pin-shifted-duplicate', 1, 'assistant', 'answer', + '2026-05-01T00:00:01Z'::timestamptz, 6, + 'uuid-answer')`) + if err != nil { + _ = tx.Rollback() + t.Fatalf("replace messages: %v", err) + } + if err := restorePinnedMessages( + ctx, tx, "pg-pin-shifted-duplicate", pins, + ); err != nil { + _ = tx.Rollback() + t.Fatalf("restorePinnedMessages: %v", err) + } + require.NoError(t, tx.Commit(), "commit tx") + + store, err := NewStore(pgURL, schema, true) + require.NoError(t, err, "NewStore") + defer store.Close() + + got, err := store.ListPinnedMessages( + ctx, "pg-pin-shifted-duplicate", "", + ) + require.NoError(t, err, "ListPinnedMessages") + require.Len(t, got, 1, "pins = %v", got) + assert.Equal(t, 1, got[0].Ordinal) + require.NotNil(t, got[0].Note) + assert.Equal(t, "keep shifted duplicate pin", *got[0].Note) +} + +// restoreIdenticalDuplicatePins seeds a session holding two identical +// (source_uuid, role, content) messages with a pin on the second one, +// replaces the messages with replacementValues through the +// snapshot/restore cycle a push performs, and returns the surviving +// pins. +func restoreIdenticalDuplicatePins( + t *testing.T, schema, sessionID, replacementValues string, +) []db.PinnedMessage { + t.Helper() + pgURL := testPGURL(t) + + pg, err := Open(pgURL, schema, true) + require.NoError(t, err, "Open") + defer pg.Close() + defer func() { + _, _ = pg.ExecContext( + context.Background(), + `DROP SCHEMA IF EXISTS `+schema+` CASCADE`, + ) + }() + + ctx := context.Background() + _, err = pg.ExecContext(ctx, `DROP SCHEMA IF EXISTS `+schema+` CASCADE`) + require.NoError(t, err, "drop schema") + require.NoError(t, EnsureSchema(ctx, pg, schema), "EnsureSchema") + + _, err = pg.ExecContext(ctx, ` + INSERT INTO sessions + (id, machine, project, agent, first_message, + started_at, message_count, user_message_count) + VALUES + ('`+sessionID+`', 'machine-a', 'proj-curation', + 'claude', 'identical duplicates', + '2026-05-01T00:00:00Z'::timestamptz, 2, 2); + INSERT INTO messages + (session_id, ordinal, role, content, timestamp, + content_length, source_uuid) + VALUES + ('`+sessionID+`', 0, 'user', 'same', + '2026-05-01T00:00:00Z'::timestamptz, 4, 'dup'), + ('`+sessionID+`', 1, 'user', 'same', + '2026-05-01T00:00:01Z'::timestamptz, 4, 'dup'); + INSERT INTO pinned_messages + (session_id, message_id, ordinal, source_uuid, + note, created_at) + VALUES + ('`+sessionID+`', 1, 1, 'dup', + 'pinned duplicate', + '2026-05-01T00:01:00Z'::timestamptz)`) + require.NoError(t, err, "seed identical duplicate pin") + + tx, err := pg.BeginTx(ctx, nil) + require.NoError(t, err, "begin tx") + pins, err := snapshotPinnedMessages(ctx, tx, sessionID) + if err != nil { + _ = tx.Rollback() + t.Fatalf("snapshotPinnedMessages: %v", err) + } + _, err = tx.ExecContext(ctx, ` + DELETE FROM messages WHERE session_id = '`+sessionID+`'; + INSERT INTO messages + (session_id, ordinal, role, content, timestamp, + content_length, source_uuid) + VALUES `+replacementValues) + if err != nil { + _ = tx.Rollback() + t.Fatalf("replace messages: %v", err) + } + if err := restorePinnedMessages(ctx, tx, sessionID, pins); err != nil { + _ = tx.Rollback() + t.Fatalf("restorePinnedMessages: %v", err) + } + require.NoError(t, tx.Commit(), "commit tx") + + store, err := NewStore(pgURL, schema, true) + require.NoError(t, err, "NewStore") + defer store.Close() + + got, err := store.ListPinnedMessages(ctx, sessionID, "") + require.NoError(t, err, "ListPinnedMessages") + return got +} + +func TestRestorePinnedMessagesKeepsPinOnUnchangedIdenticalDuplicates( + t *testing.T, +) { + got := restoreIdenticalDuplicatePins(t, + "agentsview_pin_identical_dup_keep_test", + "pg-pin-identical-dup-keep", ` + ('pg-pin-identical-dup-keep', 0, 'user', 'same', + '2026-05-01T00:00:00Z'::timestamptz, 4, 'dup'), + ('pg-pin-identical-dup-keep', 1, 'user', 'same', + '2026-05-01T00:00:01Z'::timestamptz, 4, 'dup')`) + require.Len(t, got, 1, + "unchanged identical duplicates must keep the pin; pins = %v", got) + assert.Equal(t, 1, got[0].Ordinal, "pin stays at its saved ordinal") + require.NotNil(t, got[0].Note) + assert.Equal(t, "pinned duplicate", *got[0].Note) +} + +func TestRestorePinnedMessagesDropsPinOnChangedDuplicateMultiplicity( + t *testing.T, +) { + got := restoreIdenticalDuplicatePins(t, + "agentsview_pin_identical_dup_change_test", + "pg-pin-identical-dup-change", ` + ('pg-pin-identical-dup-change', 0, 'user', 'same', + '2026-05-01T00:00:00Z'::timestamptz, 4, 'dup'), + ('pg-pin-identical-dup-change', 1, 'user', 'same', + '2026-05-01T00:00:01Z'::timestamptz, 4, 'dup'), + ('pg-pin-identical-dup-change', 2, 'user', 'same', + '2026-05-01T00:00:02Z'::timestamptz, 4, 'dup')`) + assert.Empty(t, got, + "changed duplicate multiplicity must drop the ambiguous pin") +} + +func TestRestorePinnedMessagesFollowsShiftedIdenticalDuplicates( + t *testing.T, +) { + // A context row inserted before the duplicates shifts both while + // their multiplicity stays equal: the pin must follow its + // occurrence rank instead of staying on the saved ordinal where + // the first duplicate now sits. + got := restoreIdenticalDuplicatePins(t, + "agentsview_pin_identical_dup_shift_test", + "pg-pin-identical-dup-shift", ` + ('pg-pin-identical-dup-shift', 0, 'user', 'context', + '2026-05-01T00:00:00Z'::timestamptz, 7, 'ctx'), + ('pg-pin-identical-dup-shift', 1, 'user', 'same', + '2026-05-01T00:00:01Z'::timestamptz, 4, 'dup'), + ('pg-pin-identical-dup-shift', 2, 'user', 'same', + '2026-05-01T00:00:02Z'::timestamptz, 4, 'dup')`) + require.Len(t, got, 1, + "shifted duplicates must keep the pin; pins = %v", got) + assert.Equal(t, 2, got[0].Ordinal, + "pin follows the second occurrence, not the saved ordinal") +} + +func TestRestorePinnedMessagesFollowsShiftedEqualLegacyMessages( + t *testing.T, +) { + pgURL := testPGURL(t) + + const schema = "agentsview_pin_legacy_shift_test" + const sessionID = "pg-pin-legacy-shift" + pg, err := Open(pgURL, schema, true) + require.NoError(t, err, "Open") + defer pg.Close() + defer func() { + _, _ = pg.ExecContext( + context.Background(), + `DROP SCHEMA IF EXISTS `+schema+` CASCADE`, + ) + }() + + ctx := context.Background() + _, err = pg.ExecContext(ctx, `DROP SCHEMA IF EXISTS `+schema+` CASCADE`) + require.NoError(t, err, "drop schema") + require.NoError(t, EnsureSchema(ctx, pg, schema), "EnsureSchema") + + _, err = pg.ExecContext(ctx, ` + INSERT INTO sessions + (id, machine, project, agent, first_message, + started_at, message_count, user_message_count) + VALUES + ('`+sessionID+`', 'machine-a', 'proj-curation', + 'claude', 'equal legacy messages', + '2026-05-01T00:00:00Z'::timestamptz, 3, 3); + INSERT INTO messages + (session_id, ordinal, role, content, timestamp, + content_length, source_uuid) + VALUES + ('`+sessionID+`', 0, 'user', 'intro', + '2026-05-01T00:00:00Z'::timestamptz, 5, ''), + ('`+sessionID+`', 1, 'user', 'x', + '2026-05-01T00:00:01Z'::timestamptz, 1, ''), + ('`+sessionID+`', 2, 'user', 'x', + '2026-05-01T00:00:02Z'::timestamptz, 1, ''); + INSERT INTO pinned_messages + (session_id, message_id, ordinal, source_uuid, + note, created_at) + VALUES + ('`+sessionID+`', 2, 2, '', + 'legacy pin on second x', + '2026-05-01T00:01:00Z'::timestamptz)`) + require.NoError(t, err, "seed legacy pin") + + tx, err := pg.BeginTx(ctx, nil) + require.NoError(t, err, "begin tx") + pins, err := snapshotPinnedMessages(ctx, tx, sessionID) + if err != nil { + _ = tx.Rollback() + t.Fatalf("snapshotPinnedMessages: %v", err) + } + // A hidden row inserted at the front shifts two equal visible + // messages; the pin on the second "x" must follow its occurrence + // rank to the shifted ordinal. + _, err = tx.ExecContext(ctx, ` + DELETE FROM messages WHERE session_id = '`+sessionID+`'; + INSERT INTO messages + (session_id, ordinal, role, content, timestamp, + content_length, source_uuid, is_system) + VALUES + ('`+sessionID+`', 0, 'user', 'context', + '2026-05-01T00:00:00Z'::timestamptz, 7, '', TRUE), + ('`+sessionID+`', 1, 'user', 'intro', + '2026-05-01T00:00:01Z'::timestamptz, 5, '', FALSE), + ('`+sessionID+`', 2, 'user', 'x', + '2026-05-01T00:00:02Z'::timestamptz, 1, '', FALSE), + ('`+sessionID+`', 3, 'user', 'x', + '2026-05-01T00:00:03Z'::timestamptz, 1, '', FALSE)`) + if err != nil { + _ = tx.Rollback() + t.Fatalf("replace messages: %v", err) + } + if err := restorePinnedMessages(ctx, tx, sessionID, pins); err != nil { + _ = tx.Rollback() + t.Fatalf("restorePinnedMessages: %v", err) + } + require.NoError(t, tx.Commit(), "commit tx") + + store, err := NewStore(pgURL, schema, true) + require.NoError(t, err, "NewStore") + defer store.Close() + + got, err := store.ListPinnedMessages(ctx, sessionID, "") + require.NoError(t, err, "ListPinnedMessages") + require.Len(t, got, 1, + "shifted equal messages must keep the pin; pins = %v", got) + assert.Equal(t, 3, got[0].Ordinal, + "pin follows the second occurrence, not the saved ordinal") + require.NotNil(t, got[0].Note) + assert.Equal(t, "legacy pin on second x", *got[0].Note) +} + // TestReconcilePinnedMessagesPrunesPinWhenSourceUUIDGone covers the // case where a source-backed pin's source_uuid no longer exists in // the messages table, but a different message now occupies the diff --git a/internal/postgres/push.go b/internal/postgres/push.go index 80c8cb8ff..b1992e33f 100644 --- a/internal/postgres/push.go +++ b/internal/postgres/push.go @@ -2552,6 +2552,13 @@ func (s *Sync) pushMessages( ) } if localCount == 0 { + if err := lockPinnedMessagesSession(ctx, tx, sessionID); err != nil { + return 0, err + } + savedPins, err := snapshotPinnedMessages(ctx, tx, sessionID) + if err != nil { + return 0, err + } if _, err := tx.ExecContext(ctx, `DELETE FROM tool_result_events WHERE session_id = $1`, sessionID, @@ -2584,8 +2591,8 @@ func (s *Sync) pushMessages( if err := s.replaceUsageEvents(ctx, tx, sessionID); err != nil { return 0, err } - if err := reconcilePinnedMessages( - ctx, tx, sessionID, + if err := restorePinnedMessages( + ctx, tx, sessionID, savedPins, ); err != nil { return 0, err } @@ -2804,6 +2811,13 @@ func (s *Sync) pushMessages( } } + if err := lockPinnedMessagesSession(ctx, tx, sessionID); err != nil { + return 0, err + } + savedPins, err := snapshotPinnedMessages(ctx, tx, sessionID) + if err != nil { + return 0, err + } if _, err := tx.ExecContext(ctx, ` DELETE FROM tool_result_events WHERE session_id = $1 @@ -2877,7 +2891,9 @@ func (s *Sync) pushMessages( startOrdinal = nextOrdinal } - if err := reconcilePinnedMessages(ctx, tx, sessionID); err != nil { + if err := restorePinnedMessages( + ctx, tx, sessionID, savedPins, + ); err != nil { return count, err } @@ -2909,177 +2925,407 @@ func (s *Sync) replaceUsageEvents( return nil } -func reconcilePinnedMessages( +type savedPostgresPin struct { + id int64 + ordinal int + anchorOrdinal int + sourceUUID string + role string + content string + sourceUUIDCount int + sourceIdentityCount int + sourceIdentityRank int + legacyIdentityCount int + legacyIdentityRank int + messageFound bool + note sql.NullString + createdAt time.Time +} + +type resolvedPostgresPin struct { + saved savedPostgresPin + target int + sourceUUID string +} + +// snapshotPinnedMessagesQuery captures each pin plus the identity of +// the message it anchors, before that message is deleted. A populated +// pin source_uuid is the durable anchor and may legitimately disagree +// with message_id after an older ordinal-shifting reconciliation. +// Resolve it when unique and snapshot that resolved row's anchor +// ordinal; for duplicates, accept only the row still at the recorded +// ordinal. Keep the recorded ordinal separately so conflict resolution +// can distinguish a shifted stale pin from a pin already stored on the +// resolved target. UUID-less legacy pins continue to anchor by +// message_id. +const snapshotPinnedMessagesQuery = ` + SELECT p.id, p.message_id, + COALESCE(anchored.ordinal, p.message_id), p.note, p.created_at, + CASE WHEN p.source_uuid <> '' + THEN anchored.ordinal IS NOT NULL + ELSE current_message.ordinal IS NOT NULL + END, + CASE WHEN p.source_uuid <> '' + THEN p.source_uuid + ELSE COALESCE(current_message.source_uuid, '') + END, + COALESCE( + CASE WHEN p.source_uuid <> '' + THEN anchored.role + ELSE current_message.role + END, + '' + ), + COALESCE( + CASE WHEN p.source_uuid <> '' + THEN anchored.content + ELSE current_message.content + END, + '' + ), + CASE WHEN p.source_uuid <> '' THEN ( + SELECT COUNT(*) + FROM messages same_uuid + WHERE same_uuid.session_id = p.session_id + AND same_uuid.source_uuid = p.source_uuid + ) ELSE ( + SELECT COUNT(*) + FROM messages same_uuid + WHERE same_uuid.session_id = p.session_id + AND same_uuid.source_uuid = current_message.source_uuid + AND current_message.source_uuid <> '' + ) END, + CASE WHEN p.source_uuid <> '' THEN ( + SELECT COUNT(*) + FROM messages same_identity + WHERE same_identity.session_id = p.session_id + AND same_identity.source_uuid = p.source_uuid + AND same_identity.role = anchored.role + AND same_identity.content = anchored.content + ) ELSE ( + SELECT COUNT(*) + FROM messages same_identity + WHERE same_identity.session_id = p.session_id + AND same_identity.source_uuid = current_message.source_uuid + AND same_identity.role = current_message.role + AND same_identity.content = current_message.content + AND current_message.source_uuid <> '' + ) END, + CASE WHEN p.source_uuid <> '' THEN ( + SELECT COUNT(*) + FROM messages identity_rank + WHERE identity_rank.session_id = p.session_id + AND identity_rank.source_uuid = p.source_uuid + AND identity_rank.role = anchored.role + AND identity_rank.content = anchored.content + AND identity_rank.ordinal <= anchored.ordinal + ) ELSE ( + SELECT COUNT(*) + FROM messages identity_rank + WHERE identity_rank.session_id = p.session_id + AND identity_rank.source_uuid = current_message.source_uuid + AND identity_rank.role = current_message.role + AND identity_rank.content = current_message.content + AND identity_rank.ordinal <= current_message.ordinal + AND current_message.source_uuid <> '' + ) END, + ( + SELECT COUNT(*) + FROM messages legacy_identity + WHERE legacy_identity.session_id = p.session_id + AND legacy_identity.role = current_message.role + AND legacy_identity.content = current_message.content + AND NOT legacy_identity.is_system + ), + ( + SELECT COUNT(*) + FROM messages legacy_rank + WHERE legacy_rank.session_id = p.session_id + AND legacy_rank.role = current_message.role + AND legacy_rank.content = current_message.content + AND NOT legacy_rank.is_system + AND legacy_rank.ordinal <= current_message.ordinal + ) + FROM pinned_messages p + LEFT JOIN messages current_message + ON current_message.session_id = p.session_id + AND current_message.ordinal = p.message_id + LEFT JOIN messages anchored + ON anchored.session_id = p.session_id + AND p.source_uuid <> '' + AND anchored.source_uuid = p.source_uuid + AND ( + anchored.ordinal = p.message_id + OR ( + SELECT COUNT(*) + FROM messages anchor_count + WHERE anchor_count.session_id = p.session_id + AND anchor_count.source_uuid = p.source_uuid + ) = 1 + ) + WHERE p.session_id = $1 + ORDER BY p.id + FOR UPDATE OF p` + +func snapshotPinnedMessages( ctx context.Context, tx *sql.Tx, sessionID string, +) ([]savedPostgresPin, error) { + rows, err := tx.QueryContext( + ctx, snapshotPinnedMessagesQuery, sessionID, + ) + if err != nil { + return nil, fmt.Errorf("snapshotting pg pins: %w", err) + } + defer rows.Close() + + var pins []savedPostgresPin + for rows.Next() { + var pin savedPostgresPin + if err := rows.Scan( + &pin.id, &pin.ordinal, &pin.anchorOrdinal, + &pin.note, &pin.createdAt, + &pin.messageFound, &pin.sourceUUID, + &pin.role, &pin.content, + &pin.sourceUUIDCount, &pin.sourceIdentityCount, + &pin.sourceIdentityRank, + &pin.legacyIdentityCount, &pin.legacyIdentityRank, + ); err != nil { + return nil, fmt.Errorf("scanning pg pin snapshot: %w", err) + } + pins = append(pins, pin) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterating pg pin snapshots: %w", err) + } + return pins, nil +} + +// restorePinnedMessages re-attaches the snapshotted pins to the new +// message rows through the guarded identity rules; pins whose message +// can no longer be identified are dropped. +func restorePinnedMessages( + ctx context.Context, tx *sql.Tx, sessionID string, + pins []savedPostgresPin, ) error { - if _, err := tx.ExecContext(ctx, ` - UPDATE pinned_messages p - SET source_uuid = m.source_uuid - FROM messages m - WHERE p.session_id = $1 - AND m.session_id = p.session_id - AND m.ordinal = p.message_id - AND p.source_uuid = '' - AND m.source_uuid <> ''`, - sessionID, - ); err != nil { - return fmt.Errorf( - "backfilling pg pin source_uuid: %w", err, - ) + // Delete only rows captured and locked by the snapshot. The session + // row lock taken before the snapshot (lockPinnedMessagesSession) + // serializes PinMessage/UnpinMessage against this window, so no + // same-binary writer can commit a pin between snapshot and restore. + // The ON CONFLICT DO NOTHING below is defense-in-depth for writers + // that do not take that lock (e.g. an older binary sharing the same + // database): such a pin survives and wins any target conflict + // because it represents the newer user action. + for _, pin := range pins { + if _, err := tx.ExecContext(ctx, ` + DELETE FROM pinned_messages + WHERE session_id = $1 AND id = $2`, + sessionID, pin.id, + ); err != nil { + return fmt.Errorf( + "clearing snapshotted pg pin id=%d: %w", pin.id, err, + ) + } } - // Move shifted source-backed pins out of the real ordinal range - // first. Pins already on their resolved target stay in place so - // duplicate repairs prefer the current target row's metadata. - // When multiple messages share a source_uuid (the schema allows - // it), prefer the message at the pin's current message_id so a - // correctly-placed pin is not relocated to a different duplicate. - if _, err := tx.ExecContext(ctx, ` - WITH matched AS ( - SELECT DISTINCT ON (p.id) - p.id, p.message_id, p.ordinal, - m.ordinal AS target_ordinal - FROM pinned_messages p - JOIN messages m - ON m.session_id = p.session_id - AND m.source_uuid = p.source_uuid - WHERE p.session_id = $1 - AND p.source_uuid <> '' - ORDER BY p.id, - CASE WHEN m.ordinal = p.message_id THEN 0 ELSE 1 END, - m.ordinal - ), - numbered AS ( - SELECT id, - ROW_NUMBER() OVER (ORDER BY id) AS temp_ordinal - FROM matched - WHERE target_ordinal <> message_id - OR target_ordinal <> ordinal - ) - UPDATE pinned_messages p - SET message_id = (-2000000000 + numbered.temp_ordinal::INT), - ordinal = (-2000000000 + numbered.temp_ordinal::INT) - FROM numbered - WHERE p.id = numbered.id`, - sessionID, - ); err != nil { - return fmt.Errorf( - "staging pg pins for source_uuid realignment: %w", err, + resolved := make(map[int]resolvedPostgresPin) + for _, pin := range pins { + target, sourceUUID, ok, err := resolvePinnedMessageTarget( + ctx, tx, sessionID, pin, ) + if err != nil { + return err + } + if !ok { + continue + } + candidate := resolvedPostgresPin{ + saved: pin, target: target, sourceUUID: sourceUUID, + } + current, exists := resolved[target] + if !exists || preferResolvedPostgresPin(candidate, current) { + resolved[target] = candidate + } } - if _, err := tx.ExecContext(ctx, ` - WITH matched AS ( - SELECT DISTINCT ON (p.id) - p.id, p.message_id, p.created_at, - m.ordinal AS target_ordinal - FROM pinned_messages p - JOIN messages m - ON m.session_id = p.session_id - AND m.source_uuid = p.source_uuid - WHERE p.session_id = $1 - AND p.source_uuid <> '' - ORDER BY p.id, - CASE WHEN m.ordinal = p.message_id THEN 0 ELSE 1 END, - m.ordinal - ), - ranked AS ( - SELECT id, target_ordinal, - ROW_NUMBER() OVER ( - PARTITION BY target_ordinal - ORDER BY - (message_id = target_ordinal) DESC, - created_at DESC, - id DESC - ) AS target_rank - FROM matched - ) - DELETE FROM pinned_messages p - USING ranked r - WHERE p.session_id = $1 - AND r.target_rank = 1 - AND p.message_id = r.target_ordinal - AND p.id <> r.id`, - sessionID, - ); err != nil { - return fmt.Errorf( - "clearing pg pin target conflicts: %w", err, - ) + ordinals := make([]int, 0, len(resolved)) + for ordinal := range resolved { + ordinals = append(ordinals, ordinal) } + sort.Ints(ordinals) + for _, ordinal := range ordinals { + pin := resolved[ordinal] + var note any + if pin.saved.note.Valid { + note = pin.saved.note.String + } + if _, err := tx.ExecContext(ctx, ` + INSERT INTO pinned_messages ( + id, session_id, message_id, ordinal, + source_uuid, note, created_at + ) + VALUES ($1, $2, $3, $3, $4, $5, $6) + ON CONFLICT (session_id, message_id) DO NOTHING`, + pin.saved.id, sessionID, pin.target, + pin.sourceUUID, note, pin.saved.createdAt, + ); err != nil { + return fmt.Errorf( + "restoring pg pin ord=%d: %w", pin.target, err, + ) + } + } + return nil +} - if _, err := tx.ExecContext(ctx, ` - WITH matched AS ( - SELECT DISTINCT ON (p.id) - p.id, p.message_id, p.created_at, - m.ordinal AS target_ordinal - FROM pinned_messages p - JOIN messages m - ON m.session_id = p.session_id - AND m.source_uuid = p.source_uuid - WHERE p.session_id = $1 - AND p.source_uuid <> '' - ORDER BY p.id, - CASE WHEN m.ordinal = p.message_id THEN 0 ELSE 1 END, - m.ordinal +func resolvePinnedMessageTarget( + ctx context.Context, tx *sql.Tx, sessionID string, + pin savedPostgresPin, +) (int, string, bool, error) { + if !pin.messageFound { + return 0, "", false, nil + } + if pin.sourceUUID != "" { + if pin.sourceUUIDCount == 1 { + target, sourceUUID, ok, err := scanPinnedMessageTarget( + tx.QueryRowContext(ctx, ` + SELECT m.ordinal, m.source_uuid + FROM messages m + WHERE m.session_id = $1 + AND m.source_uuid = $2 + AND ( + SELECT COUNT(*) + FROM messages same_uuid + WHERE same_uuid.session_id = m.session_id + AND same_uuid.source_uuid = m.source_uuid + ) = 1`, + sessionID, pin.sourceUUID, + ), + ) + if err != nil { + return 0, "", false, fmt.Errorf( + "resolving unique pg pin uuid=%s: %w", + pin.sourceUUID, err, + ) + } + if ok { + return target, sourceUUID, true, nil + } + } + // Identical (uuid, role, content) rows are distinguishable only + // by position, so require the identity multiplicity to be + // unchanged and re-attach at the pin's occurrence rank inside + // the group. Rank, unlike the saved ordinal, follows the + // pinned occurrence across shifts caused by rows inserted + // before the group. A different count means duplicates were + // inserted or removed and the rank no longer identifies an + // occurrence, so the pin is dropped. + target, sourceUUID, ok, err := scanPinnedMessageTarget( + tx.QueryRowContext(ctx, ` + SELECT m.ordinal, m.source_uuid + FROM messages m + WHERE m.session_id = $1 + AND m.source_uuid = $2 + AND m.role = $3 + AND m.content = $4 + AND ( + SELECT COUNT(*) + FROM messages same_identity + WHERE same_identity.session_id = m.session_id + AND same_identity.source_uuid = m.source_uuid + AND same_identity.role = m.role + AND same_identity.content = m.content + ) = $5 + AND ( + SELECT COUNT(*) + FROM messages identity_rank + WHERE identity_rank.session_id = m.session_id + AND identity_rank.source_uuid = m.source_uuid + AND identity_rank.role = m.role + AND identity_rank.content = m.content + AND identity_rank.ordinal <= m.ordinal + ) = $6`, + sessionID, pin.sourceUUID, + pin.role, pin.content, + pin.sourceIdentityCount, pin.sourceIdentityRank, + ), + ) + if err != nil { + return 0, "", false, fmt.Errorf( + "resolving ambiguous pg pin uuid=%s ord=%d: %w", + pin.sourceUUID, pin.anchorOrdinal, err, + ) + } + return target, sourceUUID, ok, nil + } + + // A UUID-less pin re-attaches to the visible row holding its role, + // content, and occurrence rank within the visible (role, content) + // group, provided the group kept its size. Rank follows the pinned + // occurrence across ordinal shifts; matching the saved ordinal + // instead could attach the pin to an earlier equal message that + // shifted into its place. + target, sourceUUID, ok, err := scanPinnedMessageTarget( + tx.QueryRowContext(ctx, ` + SELECT m.ordinal, m.source_uuid + FROM messages m + WHERE m.session_id = $1 + AND m.role = $2 + AND m.content = $3 + AND NOT m.is_system + AND ( + SELECT COUNT(*) + FROM messages legacy_identity + WHERE legacy_identity.session_id = m.session_id + AND legacy_identity.role = m.role + AND legacy_identity.content = m.content + AND NOT legacy_identity.is_system + ) = $4 + AND ( + SELECT COUNT(*) + FROM messages legacy_rank + WHERE legacy_rank.session_id = m.session_id + AND legacy_rank.role = m.role + AND legacy_rank.content = m.content + AND NOT legacy_rank.is_system + AND legacy_rank.ordinal <= m.ordinal + ) = $5`, + sessionID, pin.role, pin.content, + pin.legacyIdentityCount, pin.legacyIdentityRank, ), - ranked AS ( - SELECT id, target_ordinal, - ROW_NUMBER() OVER ( - PARTITION BY target_ordinal - ORDER BY - (message_id = target_ordinal) DESC, - created_at DESC, - id DESC - ) AS target_rank - FROM matched - ) - UPDATE pinned_messages p - SET message_id = r.target_ordinal, - ordinal = r.target_ordinal - FROM ranked r - WHERE p.id = r.id - AND r.target_rank = 1`, - sessionID, - ); err != nil { - return fmt.Errorf( - "realigning pg pins by source_uuid: %w", err, + ) + if err != nil { + return 0, "", false, fmt.Errorf( + "resolving legacy pg pin ord=%d: %w", pin.ordinal, err, ) } + return target, sourceUUID, ok, nil +} - // Prune pins whose anchor no longer exists. For source-backed - // pins (source_uuid <> '') the canonical anchor is source_uuid, - // so a pin must be dropped when no message in this session has - // that source_uuid — otherwise a stale pin can survive on top - // of an unrelated message that now occupies the same ordinal. - // The ordinal-NOT-EXISTS clause additionally removes legacy - // pins (source_uuid = '') with a stale ordinal and clears any - // non-rank-1 duplicate left at the sentinel ordinal by step 2. - if _, err := tx.ExecContext(ctx, ` - DELETE FROM pinned_messages p - WHERE p.session_id = $1 - AND ( - ( - p.source_uuid <> '' - AND NOT EXISTS ( - SELECT 1 FROM messages m - WHERE m.session_id = p.session_id - AND m.source_uuid = p.source_uuid - ) - ) - OR NOT EXISTS ( - SELECT 1 FROM messages m - WHERE m.session_id = p.session_id - AND m.ordinal = p.message_id - ) - )`, - sessionID, - ); err != nil { - return fmt.Errorf( - "pruning stale pg pins: %w", err, - ) +func scanPinnedMessageTarget( + row *sql.Row, +) (int, string, bool, error) { + var ordinal int + var sourceUUID string + if err := row.Scan(&ordinal, &sourceUUID); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return 0, "", false, nil + } + return 0, "", false, err } + return ordinal, sourceUUID, true, nil +} - return nil +func preferResolvedPostgresPin( + candidate, current resolvedPostgresPin, +) bool { + candidateAtTarget := candidate.saved.ordinal == candidate.target + currentAtTarget := current.saved.ordinal == current.target + if candidateAtTarget != currentAtTarget { + return candidateAtTarget + } + if !candidate.saved.createdAt.Equal(current.saved.createdAt) { + return candidate.saved.createdAt.After(current.saved.createdAt) + } + return candidate.saved.id > current.saved.id } func pgMessageTokenFingerprint( diff --git a/internal/server/server_test.go b/internal/server/server_test.go index 06a2e2de5..f780c8f9b 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -3981,8 +3981,10 @@ func TestUploadSession_ReuploadPreservesPins(t *testing.T) { _, err = te.db.PinMessage("upload-pinned", msgs[0].ID, ¬e) require.NoError(t, err, "PinMessage") + // The pinned message is unchanged; only the reply was edited, so + // the pin re-attaches to its message through the identity rules. updated := testjsonl.NewSessionBuilder(). - AddClaudeUser(tsEarly, "updated upload"). + AddClaudeUser(tsEarly, "original upload"). AddClaudeAssistant(tsEarlyS5, "updated reply"). String() w = te.upload(t, "upload-pinned.jsonl", updated, @@ -4002,6 +4004,133 @@ func TestUploadSession_ReuploadPreservesPins(t *testing.T) { } } +// TestUploadSession_ReuploadDropsPinOnEditedMessage documents the +// intended limit: a re-upload that edits the pinned UUID-less message +// itself destroys the only identity the pin can follow, so the pin is +// dropped rather than guessed onto the edited row. Both stores apply +// the same rule, keeping SQLite and PostgreSQL consistent. +func TestUploadSession_ReuploadDropsPinOnEditedMessage(t *testing.T) { + te := setup(t) + + initial := testjsonl.NewSessionBuilder(). + AddClaudeUser(tsEarly, "original upload"). + AddClaudeAssistant(tsEarlyS5, "original reply"). + String() + w := te.upload(t, "upload-pin-edited.jsonl", initial, + "project=myproj&machine=remote") + assertStatus(t, w, http.StatusOK) + + msgs, err := te.db.GetAllMessages( + context.Background(), "upload-pin-edited", + ) + require.NoError(t, err, "GetAllMessages") + require.Len(t, msgs, 2, "initial messages") + _, err = te.db.PinMessage("upload-pin-edited", msgs[0].ID, nil) + require.NoError(t, err, "PinMessage") + + updated := testjsonl.NewSessionBuilder(). + AddClaudeUser(tsEarly, "edited upload"). + AddClaudeAssistant(tsEarlyS5, "original reply"). + String() + w = te.upload(t, "upload-pin-edited.jsonl", updated, + "project=myproj&machine=remote") + assertStatus(t, w, http.StatusOK) + + pins, err := te.db.ListPinnedMessages( + context.Background(), "upload-pin-edited", "", + ) + require.NoError(t, err, "ListPinnedMessages") + assert.Empty(t, pins, + "editing the pinned message drops the pin") +} + +func TestUploadSession_ReuploadDoesNotMoveLegacyPinToIDEEnvelope(t *testing.T) { + te := setup(t) + const sessionID = "upload-pinned-envelope" + const mixedPrompt = "The user opened /workspace/app/README.md. Explain this file." + + require.NoError(t, te.db.UpsertSession(db.Session{ + ID: sessionID, Project: "myproj", Machine: "remote", Agent: "claude", + }), "seed legacy uploaded session") + require.NoError(t, te.db.ReplaceSessionMessages(sessionID, []db.Message{{ + SessionID: sessionID, Ordinal: 0, Role: "user", + Content: mixedPrompt, ContentLength: len(mixedPrompt), + }}), "seed legacy mixed prompt") + msgs, err := te.db.GetAllMessages(context.Background(), sessionID) + require.NoError(t, err, "GetAllMessages before re-upload") + require.Len(t, msgs, 1, "legacy messages") + _, err = te.db.PinMessage(sessionID, msgs[0].ID, nil) + require.NoError(t, err, "PinMessage") + + updated := testjsonl.NewSessionBuilder(). + AddClaudeUser(tsEarly, mixedPrompt). + String() + w := te.upload(t, sessionID+".jsonl", updated, + "project=myproj&machine=remote") + assertStatus(t, w, http.StatusOK) + + pins, err := te.db.ListPinnedMessages(context.Background(), sessionID, "") + require.NoError(t, err, "ListPinnedMessages") + assert.Empty(t, pins, + "legacy pin must not move from the prompt to hidden IDE metadata") + + msgs, err = te.db.GetAllMessages(context.Background(), sessionID) + require.NoError(t, err, "GetAllMessages after re-upload") + require.Len(t, msgs, 2, "split messages") + assert.True(t, msgs[0].IsSystem, "IDE envelope must remain hidden") + assert.Equal(t, "Explain this file.", msgs[1].Content) +} + +func TestUploadSession_ReuploadFollowsLegacyPinAcrossIDEEnvelopeSplit(t *testing.T) { + te := setup(t) + const sessionID = "upload-pinned-after-envelope" + const mixedPrompt = "The user opened /workspace/app/README.md. Explain this file." + + require.NoError(t, te.db.UpsertSession(db.Session{ + ID: sessionID, Project: "myproj", Machine: "remote", Agent: "claude", + }), "seed legacy uploaded session") + require.NoError(t, te.db.ReplaceSessionMessages(sessionID, []db.Message{ + { + SessionID: sessionID, Ordinal: 0, Role: "user", + Content: mixedPrompt, ContentLength: len(mixedPrompt), + }, + { + SessionID: sessionID, Ordinal: 1, Role: "assistant", + Content: "Legacy reply", ContentLength: len("Legacy reply"), + }, + }), "seed legacy messages") + msgs, err := te.db.GetAllMessages(context.Background(), sessionID) + require.NoError(t, err, "GetAllMessages before re-upload") + require.Len(t, msgs, 2, "legacy messages") + _, err = te.db.PinMessage(sessionID, msgs[1].ID, nil) + require.NoError(t, err, "PinMessage") + + updated := testjsonl.NewSessionBuilder(). + AddClaudeUser(tsEarly, mixedPrompt). + AddClaudeAssistant(tsEarlyS5, "Legacy reply"). + String() + w := te.upload(t, sessionID+".jsonl", updated, + "project=myproj&machine=remote") + assertStatus(t, w, http.StatusOK) + + // The envelope split shifts the whole visible tail. The pin's + // role, content, and occurrence rank identify its message, so the + // pin follows "Legacy reply" to its shifted ordinal instead of + // re-attaching to the visible prompt at the saved ordinal. + pins, err := te.db.ListPinnedMessages(context.Background(), sessionID, "") + require.NoError(t, err, "ListPinnedMessages") + require.Len(t, pins, 1, "legacy pin must survive the envelope split") + assert.Equal(t, 2, pins[0].Ordinal, + "pin follows its message, not the saved ordinal") + + msgs, err = te.db.GetAllMessages(context.Background(), sessionID) + require.NoError(t, err, "GetAllMessages after re-upload") + require.Len(t, msgs, 3, "split messages") + assert.True(t, msgs[0].IsSystem, "IDE envelope must remain hidden") + assert.Equal(t, "Explain this file.", msgs[1].Content) + assert.Equal(t, "Legacy reply", msgs[2].Content) +} + func TestUploadSession_EmptyFile(t *testing.T) { te := setup(t) diff --git a/internal/server/upload.go b/internal/server/upload.go index 9db7526b6..5da3e6c5b 100644 --- a/internal/server/upload.go +++ b/internal/server/upload.go @@ -206,21 +206,28 @@ func sessionBatchWriteFromParsed( for i, m := range msgs { hasCtx, hasOut := m.TokenPresence() dbMsgs[i] = db.Message{ - SessionID: sess.ID, - Ordinal: m.Ordinal, - Role: string(m.Role), - Content: m.Content, - Timestamp: timeutil.Format(m.Timestamp), - HasThinking: m.HasThinking, - HasToolUse: m.HasToolUse, - ContentLength: m.ContentLength, - Model: m.Model, - TokenUsage: m.TokenUsage, - PromptSource: m.PromptSource, - ContextTokens: m.ContextTokens, - OutputTokens: m.OutputTokens, - HasContextTokens: hasCtx, - HasOutputTokens: hasOut, + SessionID: sess.ID, + Ordinal: m.Ordinal, + Role: string(m.Role), + Content: m.Content, + Timestamp: timeutil.Format(m.Timestamp), + HasThinking: m.HasThinking, + HasToolUse: m.HasToolUse, + ContentLength: m.ContentLength, + IsSystem: m.IsSystem, + IsCompactBoundary: m.IsCompactBoundary, + Model: m.Model, + TokenUsage: m.TokenUsage, + PromptSource: m.PromptSource, + SourceType: m.SourceType, + SourceSubtype: m.SourceSubtype, + SourceUUID: m.SourceUUID, + SourceParentUUID: m.SourceParentUUID, + IsSidechain: m.IsSidechain, + ContextTokens: m.ContextTokens, + OutputTokens: m.OutputTokens, + HasContextTokens: hasCtx, + HasOutputTokens: hasOut, } } diff --git a/internal/server/upload_internal_test.go b/internal/server/upload_internal_test.go index 4e54b0d0b..9f6809e03 100644 --- a/internal/server/upload_internal_test.go +++ b/internal/server/upload_internal_test.go @@ -68,3 +68,44 @@ func TestSessionBatchWriteFromParsedPreservesClaudeProvenance(t *testing.T) { require.Len(t, result.Messages, 1) assert.Equal(t, "queued", result.Messages[0].PromptSource) } + +func TestSessionBatchWriteFromParsedPreservesMessageIdentity(t *testing.T) { + sess := parser.ParsedSession{ID: "test-message-identity"} + msgs := []parser.ParsedMessage{{ + Ordinal: 1, + Role: parser.RoleUser, + Content: "hidden context", + IsSystem: true, + SourceType: "system", + SourceSubtype: "ide_opened_file", + SourceUUID: "entry-1:ide-context", + SourceParentUUID: "parent-1", + IsSidechain: true, + }} + + result := sessionBatchWriteFromParsed(sess, msgs) + + require.Len(t, result.Messages, 1) + assert.True(t, result.Messages[0].IsSystem) + assert.Equal(t, "system", result.Messages[0].SourceType) + assert.Equal(t, "ide_opened_file", result.Messages[0].SourceSubtype) + assert.Equal(t, "entry-1:ide-context", result.Messages[0].SourceUUID) + assert.Equal(t, "parent-1", result.Messages[0].SourceParentUUID) + assert.True(t, result.Messages[0].IsSidechain) +} + +func TestSessionBatchWriteFromParsedPreservesCompactBoundary(t *testing.T) { + sess := parser.ParsedSession{ID: "test-compact-boundary"} + msgs := []parser.ParsedMessage{{ + Ordinal: 1, + Role: parser.RoleSystem, + Content: "Conversation compacted", + IsSystem: true, + IsCompactBoundary: true, + }} + + result := sessionBatchWriteFromParsed(sess, msgs) + + require.Len(t, result.Messages, 1) + assert.True(t, result.Messages[0].IsCompactBoundary) +}