From e370e5af0da788eea30972ee176dd47b5f993451 Mon Sep 17 00:00:00 2001 From: Coder8124 Date: Sun, 8 Nov 2026 22:36:18 -0800 Subject: [PATCH] logos migrate sessions files every old checkpoint into its conversation and names conversation folders for what they were about --- cmd/logos/convo_resume_test.go | 6 +- cmd/logos/migrate.go | 55 +++- internal/session/auto.go | 3 +- internal/session/checkpoint.go | 166 +++++++++++- internal/session/convo_names_test.go | 365 +++++++++++++++++++++++++ internal/session/convos.go | 384 ++++++++++++++++++++++++--- internal/session/convos_test.go | 15 +- internal/session/transcript.go | 10 +- internal/transcript/receipts.go | 175 ++++++++++++ internal/transcript/receipts_test.go | 100 +++++++ 10 files changed, 1214 insertions(+), 65 deletions(-) create mode 100644 internal/session/convo_names_test.go create mode 100644 internal/transcript/receipts.go create mode 100644 internal/transcript/receipts_test.go diff --git a/cmd/logos/convo_resume_test.go b/cmd/logos/convo_resume_test.go index a3e9841..5150ce1 100644 --- a/cmd/logos/convo_resume_test.go +++ b/cmd/logos/convo_resume_test.go @@ -58,8 +58,10 @@ func TestASessionFolderSurvivesDeletingTheIndex(t *testing.T) { t.Errorf("resume %q after deleting the index is not that one conversation:\n%s", q, out) } } - if _, err := os.Stat(filepath.Join(vault, session.CheckpointDir, "kestrel", session.ConvosDir, "sess-checkout")); err != nil { - t.Errorf("the conversation folder is gone after reindexing: %v", err) + // The folder is named for its intent ("--sess-checkout"); the id + // after the "--" is what must survive. + if m, _ := filepath.Glob(filepath.Join(vault, session.CheckpointDir, "kestrel", session.ConvosDir, "*--sess-checkout")); len(m) != 1 { + t.Errorf("the conversation folder is gone after reindexing: %v", m) } } diff --git a/cmd/logos/migrate.go b/cmd/logos/migrate.go index d8582a2..d33354c 100644 --- a/cmd/logos/migrate.go +++ b/cmd/logos/migrate.go @@ -10,6 +10,8 @@ import ( "github.com/Coder8124/logos/internal/session" "github.com/Coder8124/logos/internal/setup" + "github.com/Coder8124/logos/internal/transcript" + "github.com/Coder8124/logos/internal/untrusted" "github.com/Coder8124/logos/internal/vault" ) @@ -407,9 +409,16 @@ func isBrainBinary(bin string) bool { return name == setup.OldName } -// migrateSessionsCmd moves a vault's flat checkpoints — every one written -// before conversations had folders — into the folder their own host_session -// names, and says how many it moved, skipped and could not move. +// migrateSessionsCmd files every checkpoint not yet in a conversation folder +// of its own — flat ones from before conversations had folders, and the +// _unattached pile — into one, names each folder after what the conversation +// was for, and says how many it moved, recovered, named, skipped and could not +// move. +// +// A checkpoint older than host_session cannot say which conversation wrote +// it, but that conversation's transcript still holds Logos's receipt for it; +// the receipts are read from this machine's Claude Code and Codex transcripts, +// and nothing leaves it. // // Explicit, never implied by reading: an unmigrated project resumes by the // same ids it will have afterwards (see session.Conversations), so nothing @@ -427,23 +436,55 @@ func migrateSessionsCmd(args []string) error { if _, err := os.Stat(v); err != nil { return missingVaultError(v) } - rep, err := session.MigrateFlat(v, dryRun) + // The transcripts are read only if some checkpoint has no host_session: + // on a vault written since that was recorded, scanning gigabytes of host + // history would buy nothing. + var unreadable []string + receipts := func() (map[string][]session.Receipt, error) { + found, skipped, err := transcript.CheckpointReceipts() + unreadable = skipped + if err != nil { + return nil, fmt.Errorf("reading host transcripts for checkpoint receipts: %w", err) + } + return found, nil + } + rep, err := session.MigrateFlat(v, dryRun, receipts) if err != nil { return err } - verb := "moved" + if len(unreadable) > 0 { + // Named, so a checkpoint left unattributed below can be traced to a + // transcript Logos could not open rather than read as having none. + fmt.Printf("%d host transcript%s could not be read and passed over:\n", len(unreadable), pluralS(len(unreadable))) + for _, p := range unreadable { + fmt.Printf(" %s\n", p) + } + } + verb, named := "moved", "named" if dryRun { - verb = "would move" + verb, named = "would move", "would name" } fmt.Printf("%s %d checkpoint%s into conversation folders, skipped %d, failed %d — in %s\n", verb, rep.Moved, pluralS(rep.Moved), len(rep.Skipped), len(rep.Failed), v) + if rep.Recovered > 0 { + fmt.Printf(" %d matched to their conversation by its receipt in a host transcript\n", rep.Recovered) + } + if rep.Unattributed > 0 { + fmt.Printf(" %d nothing ties to a conversation, each given a folder of its own\n", rep.Unattributed) + } + if rep.Named > 0 { + fmt.Printf(" %s %d conversation folder%s after what it was for\n", named, rep.Named, pluralS(rep.Named)) + } + for _, a := range rep.Ambiguous { + fmt.Println(" ambiguous " + untrusted.Inline(a) + " — filed on its own rather than under a guess") + } for _, s := range rep.Skipped { fmt.Println(" skipped " + s) } for _, f := range rep.Failed { fmt.Println(" failed " + f) } - if rep.Moved > 0 && !dryRun { + if (rep.Moved > 0 || rep.Named > 0) && !dryRun { // The vault moved under the cache, which still lists the old paths // until it is rebuilt. Said rather than done: an index run on a large // vault is minutes, and the user should choose when to spend them. diff --git a/internal/session/auto.go b/internal/session/auto.go index f3dd0d1..a9f458a 100644 --- a/internal/session/auto.go +++ b/internal/session/auto.go @@ -86,8 +86,7 @@ func WriteAuto(vaultDir string, c Checkpoint) (Checkpoint, error) { } else if prev, _ := Latest(vaultDir, asGiven); prev != nil { follows = prev.Session } - convo := convoFolder(c.HostSession) - id, path, err := claimCheckpoint(vaultDir, c.Project, c.Agent, idFor(c.Agent, time.Unix(c.TS, 0)), convo) + convo, id, path, err := claimInConvo(vaultDir, &c) if err != nil { return Checkpoint{}, err } diff --git a/internal/session/checkpoint.go b/internal/session/checkpoint.go index 5557cfb..f01c9cd 100644 --- a/internal/session/checkpoint.go +++ b/internal/session/checkpoint.go @@ -167,6 +167,123 @@ func convoFolder(hostSession string) string { return UnattachedFolder } +// folderNameSep joins a conversation folder's readable name to its id. A bare +// id is unreadable in a file browser — a UUID says nothing about which chat it +// was — so a folder is "--". dropSymbols never writes two +// dashes in a row, so neither half can contain the separator, and the id is +// always recoverable from the name alone (FolderID). +const folderNameSep = "--" + +// slugWords and slugMax keep the readable half to a phrase: the first words +// of an intent are its subject, and a folder name is a label, not the intent. +const ( + slugWords = 6 + slugMax = 48 +) + +// intentSlug is the readable half of a conversation folder's name. +func intentSlug(about string) string { + words := strings.Fields(about) + if len(words) > slugWords { + words = words[:slugWords] + } + return capAt(dropSymbols(strings.Join(words, " ")), slugMax) +} + +// namedConvoFolder is the folder name for a new conversation id, named after +// what it is about; the bare id when there is nothing to name it by. +func namedConvoFolder(about, id string) string { + slug := intentSlug(about) + if slug == "" || id == UnattachedFolder || len(slug)+len(folderNameSep)+len(id) > maxNameLen { + return id + } + return slug + folderNameSep + id +} + +// FolderID is the conversation id a folder name carries: everything after the +// first separator, or the whole name for a folder written before folders were +// named (and for UnattachedFolder). Every lookup of a conversation by id goes +// through this, never through the folder name, because the name's readable +// half is fixed when the folder is made and the id is what hooks, transcripts +// and `logos resume ` know. +func FolderID(name string) string { + if _, id, ok := strings.Cut(name, folderNameSep); ok && id != "" { + return id + } + return name +} + +// existingConvoFolder is the folder already holding conversation id in a +// scope, or "" when it has none yet. A scope with no _convos at all has none; +// any other failure to read it is an error, because answering "none" would +// have the caller make a second folder for a conversation that has one. +func existingConvoFolder(scopeDir, id string) (string, error) { + entries, err := os.ReadDir(filepath.Join(scopeDir, ConvosDir)) + if os.IsNotExist(err) { + return "", nil + } + if err != nil { + return "", err + } + for _, e := range entries { + if e.IsDir() && FolderID(e.Name()) == id { + return e.Name(), nil + } + } + return "", nil +} + +// convoFolderIn is the folder a checkpoint of hostSession goes to in a scope: +// the conversation's existing folder whatever it is called, so one +// conversation never splits across two names when its intent is restated; +// otherwise a new one named after about. A caller that will create the folder +// holds convoFolderLock across the choice and the creation. +func convoFolderIn(scopeDir, hostSession, about string) (string, error) { + id := convoFolder(hostSession) + if id == UnattachedFolder { + return id, nil + } + f, err := existingConvoFolder(scopeDir, id) + if err != nil || f != "" { + return f, err + } + return namedConvoFolder(about, id), nil +} + +// convoFolderLock is held from choosing a conversation's folder to creating +// it. While folders were named by id alone, two writers in a new conversation +// agreed on its folder without talking; now the name comes from what each one +// is about, so the Stop hook's auto record and an agent's checkpoint landing +// together would each find no folder and make one under a different name, +// splitting the conversation in two. +const convoFolderLock = "convo-folders" + +// claimInConvo picks c's conversation folder and claims its checkpoint file +// there, under convoFolderLock, so the folder exists before the next writer +// looks for it. +func claimInConvo(vaultDir string, c *Checkpoint) (convo, id, path string, err error) { + g, err := vault.Lock(vaultDir, convoFolderLock) + if err != nil { + return "", "", "", err + } + defer g.Unlock() + convo, err = convoFolderIn(filepath.Join(vaultDir, CheckpointDir, filepath.FromSlash(c.Project)), c.HostSession, c.about()) + if err != nil { + return "", "", "", err + } + id, path, err = claimCheckpoint(vaultDir, c.Project, c.Agent, idFor(c.Agent, time.Unix(c.TS, 0)), convo) + return convo, id, path, err +} + +// about is what a new conversation folder is named after: why the work +// matters if the checkpoint says, otherwise what it was doing. +func (c Checkpoint) about() string { + if strings.TrimSpace(c.Intent) != "" { + return c.Intent + } + return c.Task +} + func (c Checkpoint) Empty() bool { return strings.TrimSpace(c.Task) == "" && strings.TrimSpace(c.State) == "" && strings.TrimSpace(c.Next) == "" && len(c.Decisions) == 0 && @@ -302,8 +419,7 @@ func Commit(db *sql.DB, vaultDir string, c *Checkpoint) error { // which is when the session started: History sorts by name, so a long // session's checkpoint sorted behind a shorter one written before it, and // resume showed the older one as where work left off (#366). - convo := convoFolder(c.HostSession) - id, path, err := claimCheckpoint(vaultDir, c.Project, c.Agent, idFor(c.Agent, time.Unix(c.TS, 0)), convo) + convo, id, path, err := claimInConvo(vaultDir, c) if err != nil { return err } @@ -647,6 +763,13 @@ func ByHostSession(vaultDir, hostSession string) ([]Checkpoint, error) { // It also rejoins a checkpoint filed under a cut id (see rejoinCutIDs): the // second count is how many of the moved ones had their host_session rewritten. func Refile(vaultDir string) (refiled, rejoined int, err error) { + // Refile creates conversation folders too, so it takes the lock a + // checkpoint takes to make one (convoFolderLock). + g, err := vault.Lock(vaultDir, convoFolderLock) + if err != nil { + return 0, 0, err + } + defer g.Unlock() scopes, err := Scopes(vaultDir) if err != nil { return 0, 0, err @@ -681,10 +804,14 @@ func Refile(vaultDir string) (refiled, rejoined int, err error) { } raw, c.HostSession, wasCut = fixed, full, true } - wantRel := filepath.ToSlash(filepath.Join(ConvosDir, convoFolder(c.HostSession), en.Name)) - if en.Rel == wantRel { + if filedRight(en, c.HostSession) { continue // already filed where its own frontmatter says it belongs } + convo, err := convoFolderIn(dir, c.HostSession, c.about()) + if err != nil { + return refiled, rejoined, fmt.Errorf("re-filing %s: finding its conversation's folder: %w", en.Rel, err) + } + wantRel := filepath.ToSlash(filepath.Join(ConvosDir, convo, en.Name)) newPath := filepath.Join(dir, filepath.FromSlash(wantRel)) // Its sync copies go too, before the original leaves. A copy is // set aside only while its original is beside it, so one left @@ -728,6 +855,24 @@ func Refile(vaultDir string) (refiled, rejoined int, err error) { return refiled, rejoined, nil } +// filedRight says a checkpoint already sits in the conversation folder its +// own frontmatter names, by whatever readable name that folder was given. A +// checkpoint with no host session is filed right in UnattachedFolder or in a +// folder of its own — the one `logos migrate sessions` gives it, whose id is +// its own stem — and pulling it back out of the latter would undo the +// migration on the next index. +func filedRight(en CheckpointEntry, hostSession string) bool { + folder, ok := strings.CutPrefix(path.Dir(en.Rel), ConvosDir+"/") + if !ok || strings.Contains(folder, "/") { + return false + } + id := FolderID(folder) + if id == convoFolder(hostSession) { + return true + } + return convoFolder(hostSession) == UnattachedFolder && id == strings.TrimSuffix(en.Name, ".md") +} + // Before 0.5.8 the Stop hook's checkpoint-asked marker carried only the // activity log's 8-character cut of the host's id, and a checkpoint answering // it was filed under that cut, beside its conversation's real folder: one chat @@ -776,8 +921,8 @@ func fullIDsByCut(cps []Checkpoint) map[string]string { // only, so a body that happens to quote the line is left alone. func replaceHostSession(raw []byte, from, to string) ([]byte, bool) { s := string(raw) - end := strings.Index(s, "\n---") - if !strings.HasPrefix(s, "---\n") || end < 0 { + end, ok := frontmatterEnd(s) + if !ok { return nil, false } line := "\nhost_session: " + yamlStr(from) + "\n" @@ -788,6 +933,15 @@ func replaceHostSession(raw []byte, from, to string) ([]byte, bool) { return []byte(s[:i] + "\nhost_session: " + yamlStr(to) + "\n" + s[i+len(line):]), true } +// frontmatterEnd is the index of the newline before a checkpoint's closing +// "---", so s[:end+1] is its frontmatter and nothing of its body; ok is false +// for a file that does not open with frontmatter. Every in-place edit of a +// frontmatter line bounds its search with this. +func frontmatterEnd(s string) (end int, ok bool) { + end = strings.Index(s, "\n---") + return end, strings.HasPrefix(s, "---\n") && end >= 0 +} + // ResolvePath finds a checkpoint a stored citation names, wherever it // actually sits in the vault now. // diff --git a/internal/session/convo_names_test.go b/internal/session/convo_names_test.go new file mode 100644 index 0000000..a85cef8 --- /dev/null +++ b/internal/session/convo_names_test.go @@ -0,0 +1,365 @@ +package session + +import ( + "database/sql" + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" +) + +// convoDirOf is the conversation folder a checkpoint file sits in. +func convoDirOf(path string) string { return filepath.Base(filepath.Dir(path)) } + +func TestConversationFoldersAreNamedByIntent(t *testing.T) { + dir := t.TempDir() + first := commitIn(t, dir, &Checkpoint{Project: "kestrel", Agent: "a", Task: "fix checkout", + Intent: "Checkout crashes lose orders on every retry we make", HostSession: "sess-checkout"}) + if got := convoDirOf(first); got != "checkout-crashes-lose-orders-on-every--sess-checkout" { + t.Fatalf("folder is %q, want the intent's first words before the host id", got) + } + // The name is fixed when the folder is made: a restated intent later in + // the same conversation files beside the first, never into a second folder. + second := commitIn(t, dir, &Checkpoint{Project: "kestrel", Agent: "a", Task: "fix checkout", + Intent: "Now about refunds", HostSession: "sess-checkout"}) + if filepath.Dir(second) != filepath.Dir(first) { + t.Fatalf("a second checkpoint of one conversation went to %s, not %s", filepath.Dir(second), filepath.Dir(first)) + } + // No intent: the task names it. + byTask := commitIn(t, dir, &Checkpoint{Project: "kestrel", Agent: "a", Task: "Speed up search!", HostSession: "sess-search"}) + if got := convoDirOf(byTask); got != "speed-up-search--sess-search" { + t.Fatalf("folder is %q, want the task before the host id", got) + } + // Neither: the bare id, as before. + bare := commitIn(t, dir, &Checkpoint{Project: "kestrel", Agent: "a", State: "s", HostSession: "sess-bare"}) + if got := convoDirOf(bare); got != "sess-bare" { + t.Fatalf("folder is %q, want the bare host id", got) + } + + // The id a person resumes by is still the host's, not the folder's name. + convs, err := Conversations(dir, "kestrel") + if err != nil { + t.Fatal(err) + } + ids := map[string]int{} + for _, c := range convs { + ids[c.ID] = len(c.Chain) + } + if ids["sess-checkout"] != 2 || ids["sess-search"] != 1 || ids["sess-bare"] != 1 || len(ids) != 3 { + t.Fatalf("conversations by id: %v", ids) + } +} + +func TestRefileLeavesAnIntentNamedFolderAlone(t *testing.T) { + dir := t.TempDir() + path := commitIn(t, dir, &Checkpoint{Project: "kestrel", Agent: "a", Task: "fix checkout", HostSession: "sess-checkout"}) + moved, _, err := Refile(dir) + if err != nil || moved != 0 { + t.Fatalf("refile moved %d (%v) out of a folder already named for its conversation", moved, err) + } + if _, err := os.Stat(path); err != nil { + t.Fatal(err) + } +} + +func TestTranscriptDirFindsAnIntentNamedFolder(t *testing.T) { + dir := t.TempDir() + path := commitIn(t, dir, &Checkpoint{Project: "kestrel", Agent: "a", Task: "fix checkout", HostSession: "sess-checkout"}) + got, err := TranscriptDir(dir, "kestrel", "sess-checkout") + if err != nil || got != filepath.Dir(path) { + t.Fatalf("transcript would go to %s (%v), not beside its checkpoints in %s", got, err, filepath.Dir(path)) + } +} + +// flatNoHost writes a checkpoint the way every one before host_session was: +// flat in its project, naming no conversation. It returns the path and the +// receipts key for it. +func flatNoHost(t *testing.T, dir string, c *Checkpoint) (string, string, time.Time) { + t.Helper() + path := flatten(t, dir, c.Project, commitIn(t, dir, c)) + raw, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + stem := strings.TrimSuffix(filepath.Base(path), ".md") + return path, c.Project + "/" + stem, time.Unix(ParseCheckpoint(string(raw)).TS, 0) +} + +func TestMigrateFilesAFlatCheckpointIntoTheConversationItsReceiptNames(t *testing.T) { + dir := t.TempDir() + path, key, at := flatNoHost(t, dir, &Checkpoint{Project: "kestrel", Agent: "a", Task: "fix checkout", Intent: "Orders are lost"}) + // The same conversation's later checkpoint, already in its folder: the + // flat one joins it there, under the name that folder already has. + later := commitIn(t, dir, &Checkpoint{Project: "kestrel", Agent: "a", Task: "fix checkout", State: "done", HostSession: "sess-r"}) + + rep, err := MigrateFlat(dir, false, Receipts(map[string][]Receipt{key: {{HostSession: "sess-r", At: at.Add(2 * time.Second)}}})) + if err != nil || rep.Moved != 1 || rep.Recovered != 1 || len(rep.Failed) != 0 { + t.Fatalf("want 1 moved and recovered, got %+v %v", rep, err) + } + want := filepath.Join(filepath.Dir(later), filepath.Base(path)) + raw, err := os.ReadFile(want) + if err != nil { + t.Fatalf("not filed into its conversation's folder: %v", err) + } + if c := ParseCheckpoint(string(raw)); c.HostSession != "sess-r" || c.Intent != "Orders are lost" { + t.Fatalf("frontmatter after the move: host %q intent %q", c.HostSession, c.Intent) + } + if _, err := os.Stat(path); !os.IsNotExist(err) { + t.Fatal("the flat copy is still there") + } + convs, _ := Conversations(dir, "kestrel") + if len(convs) != 1 || convs[0].ID != "sess-r" || len(convs[0].Chain) != 2 { + t.Fatalf("want one conversation of two checkpoints, got %+v", convs) + } + // Run again: nothing left to do. + again, err := MigrateFlat(dir, false, nil) + if err != nil || again.Moved != 0 || again.Named != 0 { + t.Fatalf("a second run did more work: %+v %v", again, err) + } +} + +func TestAReceiptQuotedLongAfterTheCheckpointDoesNotClaimIt(t *testing.T) { + dir := t.TempDir() + path, key, at := flatNoHost(t, dir, &Checkpoint{Project: "kestrel", Agent: "a", Task: "tidy logs"}) + rep, err := MigrateFlat(dir, false, Receipts(map[string][]Receipt{key: {{HostSession: "sess-later", At: at.Add(2 * time.Hour)}}})) + if err != nil || rep.Recovered != 0 || rep.Unattributed != 1 { + t.Fatalf("want it unattributed, got %+v %v", rep, err) + } + stem := strings.TrimSuffix(filepath.Base(path), ".md") + own := filepath.Join(dir, CheckpointDir, "kestrel", ConvosDir, "tidy-logs--"+stem, filepath.Base(path)) + raw, err := os.ReadFile(own) + if err != nil { + t.Fatalf("not filed in a folder of its own: %v", err) + } + if ParseCheckpoint(string(raw)).HostSession != "" { + t.Fatal("a host session was written that nothing established") + } +} + +func TestACheckpointTwoConversationsClaimStaysUnattributedAndIsReported(t *testing.T) { + dir := t.TempDir() + path, key, at := flatNoHost(t, dir, &Checkpoint{Project: "kestrel", Agent: "a", Task: "tidy logs"}) + rep, err := MigrateFlat(dir, false, Receipts(map[string][]Receipt{key: { + {HostSession: "sess-1", At: at.Add(time.Second)}, + {HostSession: "sess-2", At: at.Add(time.Second)}, + }})) + // Reported once, as ambiguous: counting it unattributed as well would + // make the totals add up to more checkpoints than were moved. + if err != nil || rep.Recovered != 0 || rep.Unattributed != 0 || len(rep.Ambiguous) != 1 || !strings.Contains(rep.Ambiguous[0], filepath.Base(path)) { + t.Fatalf("want it reported ambiguous, got %+v %v", rep, err) + } + convs, _ := Conversations(dir, "kestrel") + if len(convs) != 1 || convs[0].Latest.HostSession != "" { + t.Fatalf("a guess was written: %+v", convs) + } +} + +func TestMigrateSplitsUnattachedIntoOwnFolders(t *testing.T) { + dir := t.TempDir() + a := commitIn(t, dir, &Checkpoint{Project: "kestrel", Agent: "a", Task: "first thing", TS: 1_700_000_000}) + b := commitIn(t, dir, &Checkpoint{Project: "kestrel", Agent: "b", Task: "second thing", TS: 1_700_000_100}) + if convoDirOf(a) != UnattachedFolder || convoDirOf(b) != UnattachedFolder { + t.Fatalf("setup: no-host checkpoints landed in %s and %s", convoDirOf(a), convoDirOf(b)) + } + rep, err := MigrateFlat(dir, false, nil) + if err != nil || rep.Moved != 2 || rep.Unattributed != 2 { + t.Fatalf("want 2 moved into their own folders, got %+v %v", rep, err) + } + for _, p := range []string{a, b} { + stem := strings.TrimSuffix(filepath.Base(p), ".md") + matches, _ := filepath.Glob(filepath.Join(dir, CheckpointDir, "kestrel", ConvosDir, "*--"+stem, filepath.Base(p))) + if len(matches) != 1 { + t.Fatalf("%s is not in a folder of its own", filepath.Base(p)) + } + } + if _, err := os.Stat(filepath.Dir(a)); !os.IsNotExist(err) { + t.Fatal("the emptied _unattached folder was left behind") + } + // Refiling afterwards must not pull them back into _unattached. + if moved, _, err := Refile(dir); err != nil || moved != 0 { + t.Fatalf("refile moved %d (%v) back out of their own folders", moved, err) + } +} + +func TestMigrateNamesABareConversationFolderByItsIntent(t *testing.T) { + dir := t.TempDir() + named := commitIn(t, dir, &Checkpoint{Project: "kestrel", Agent: "a", Task: "fix checkout", Intent: "Orders are lost", HostSession: "sess-old"}) + // The folder as 0.5.8 and earlier named it: the host id alone. + bareDir := filepath.Join(filepath.Dir(filepath.Dir(named)), "sess-old") + if err := os.Rename(filepath.Dir(named), bareDir); err != nil { + t.Fatal(err) + } + + dry, err := MigrateFlat(dir, true, nil) + if err != nil || dry.Named != 1 { + t.Fatalf("dry run: %+v %v", dry, err) + } + if _, err := os.Stat(bareDir); err != nil { + t.Fatal("a dry run renamed a folder") + } + rep, err := MigrateFlat(dir, false, nil) + if err != nil || rep.Named != 1 || rep.Moved != 0 { + t.Fatalf("want 1 folder named and nothing moved, got %+v %v", rep, err) + } + if _, err := os.Stat(named); err != nil { + t.Fatalf("the folder was not renamed to its intent: %v", err) + } + // A save into the conversation lands in the renamed folder. + next := commitIn(t, dir, &Checkpoint{Project: "kestrel", Agent: "a", Task: "fix checkout", State: "s", HostSession: "sess-old"}) + if filepath.Dir(next) != filepath.Dir(named) { + t.Fatalf("next save went to %s", filepath.Dir(next)) + } +} + +// Doctor and the status line read a checkpoint file's modification time as +// when it was written; a migration that rewrote it as "now" would make the +// oldest checkpoint in the vault look like the latest one. +func TestMigratingACheckpointKeepsWhenItWasWritten(t *testing.T) { + dir := t.TempDir() + path, key, at := flatNoHost(t, dir, &Checkpoint{Project: "kestrel", Agent: "a", Task: "old work"}) + written := time.Date(2026, 3, 1, 9, 0, 0, 0, time.UTC) + if err := os.Chtimes(path, written, written); err != nil { + t.Fatal(err) + } + rep, err := MigrateFlat(dir, false, Receipts(map[string][]Receipt{key: {{HostSession: "sess-o", At: at}}})) + if err != nil || rep.Moved != 1 { + t.Fatalf("%+v %v", rep, err) + } + moved, _ := filepath.Glob(filepath.Join(dir, CheckpointDir, "kestrel", ConvosDir, "*--sess-o", filepath.Base(path))) + if len(moved) != 1 { + t.Fatal("not moved into its conversation") + } + info, err := os.Stat(moved[0]) + if err != nil || !info.ModTime().Equal(written) { + t.Fatalf("the moved checkpoint reads as written %v, not %v", info.ModTime(), written) + } +} + +// A conversation whose transcript was saved before any checkpoint has a +// folder already, under its bare id. The migrated checkpoint must join it, +// not start a second folder for the same conversation beside it. +func TestMigrateJoinsAFolderHoldingOnlyATranscript(t *testing.T) { + dir := t.TempDir() + path, key, at := flatNoHost(t, dir, &Checkpoint{Project: "kestrel", Agent: "a", Task: "fix checkout"}) + bare := filepath.Join(dir, CheckpointDir, "kestrel", ConvosDir, "sess-t") + if err := os.MkdirAll(bare, 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(bare, "transcript.md"), []byte("talk\n"), 0o644); err != nil { + t.Fatal(err) + } + rep, err := MigrateFlat(dir, false, Receipts(map[string][]Receipt{key: {{HostSession: "sess-t", At: at}}})) + if err != nil || rep.Moved != 1 || len(rep.Failed) != 0 { + t.Fatalf("%+v %v", rep, err) + } + folders, _ := os.ReadDir(filepath.Join(dir, CheckpointDir, "kestrel", ConvosDir)) + var ids []string + for _, f := range folders { + if FolderID(f.Name()) == "sess-t" { + ids = append(ids, f.Name()) + } + } + if len(ids) != 1 { + t.Fatalf("one conversation has %d folders: %v", len(ids), ids) + } + conv := filepath.Join(dir, CheckpointDir, "kestrel", ConvosDir, ids[0]) + for _, f := range []string{filepath.Base(path), "transcript.md"} { + if _, err := os.Stat(filepath.Join(conv, f)); err != nil { + t.Fatalf("%s is not in the conversation's folder: %v", f, err) + } + } +} + +// A sync client's conflict copy is the same checkpoint as another device +// saw it. Left behind, it would keep the old flat place (or _unattached) +// alive with a stray that no conversation owns. +func TestSyncCopiesFollowAMigratedCheckpoint(t *testing.T) { + dir := t.TempDir() + a := commitIn(t, dir, &Checkpoint{Project: "kestrel", Agent: "a", Task: "first thing"}) + stem := strings.TrimSuffix(filepath.Base(a), ".md") + copyName := stem + " (conflicted copy).md" + raw, _ := os.ReadFile(a) + if err := os.WriteFile(filepath.Join(filepath.Dir(a), copyName), raw, 0o644); err != nil { + t.Fatal(err) + } + rep, err := MigrateFlat(dir, false, nil) + if err != nil || rep.Moved != 1 || len(rep.Failed) != 0 || len(rep.Skipped) != 0 { + t.Fatalf("%+v %v", rep, err) + } + moved, _ := filepath.Glob(filepath.Join(dir, CheckpointDir, "kestrel", ConvosDir, "*--"+stem, copyName)) + if len(moved) != 1 { + t.Fatal("the sync copy did not follow its checkpoint") + } + if _, err := os.Stat(filepath.Dir(a)); !os.IsNotExist(err) { + t.Fatalf("_unattached was left behind holding the copy: %v", err) + } +} + +// Some writers leave the key with no value. Adding a second host_session +// beside it makes the frontmatter undecodable, and the checkpoint reads as +// empty everywhere afterwards. +func TestAnEmptyHostSessionKeyIsFilledNotDuplicated(t *testing.T) { + for _, empty := range []string{"host_session:", `host_session: ""`, "host_session: ''"} { + raw := "---\nsession: \"x\"\n" + empty + "\nagent: a\n---\nbody host_session: no\n" + got, ok := withHostSession([]byte(raw), "sess-1") + if !ok || strings.Count(string(got), "host_session:") != 2 || !strings.Contains(string(got), "host_session: "+yamlStr("sess-1")+"\n") { + t.Fatalf("%q became:\n%s", empty, got) + } + if c := ParseCheckpoint(string(got)); c.HostSession != "sess-1" { + t.Fatalf("%q: the result does not decode to its host: %+v", empty, c) + } + } +} + +// Reading every host transcript costs seconds on a machine with years of +// them; a vault whose checkpoints all name their conversation needs none. +func TestTheReceiptSourceIsNotReadWhenEveryCheckpointHasAHost(t *testing.T) { + dir := t.TempDir() + flatten(t, dir, "kestrel", commitIn(t, dir, &Checkpoint{Project: "kestrel", Agent: "a", Task: "fix checkout", HostSession: "sess-h"})) + read := false + rep, err := MigrateFlat(dir, false, func() (map[string][]Receipt, error) { + read = true + return nil, nil + }) + if err != nil || rep.Moved != 1 { + t.Fatalf("%+v %v", rep, err) + } + if read { + t.Fatal("the host transcripts were read for a vault that needed none") + } +} + +// The Stop hook's automatic record and an agent's checkpoint can land in the +// same new conversation at once, with different intents. Each deciding the +// folder name alone would give the conversation two folders. +func TestTwoWritersInANewConversationShareOneFolder(t *testing.T) { + dir := t.TempDir() + const n = 8 + dbs := make([]*sql.DB, n) + for i := range dbs { + dbs[i] = testDB(t) + } + var wg sync.WaitGroup + errs := make(chan error, n) + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + errs <- Commit(dbs[i], dir, &Checkpoint{Project: "kestrel", Agent: fmt.Sprintf("a%d", i), Task: fmt.Sprintf("task %d", i), Intent: fmt.Sprintf("Reason number %d", i), HostSession: "sess-race"}) + }(i) + } + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Fatal(err) + } + } + folders, _ := filepath.Glob(filepath.Join(dir, CheckpointDir, "kestrel", ConvosDir, "*sess-race")) + if len(folders) != 1 { + t.Fatalf("one conversation got %d folders: %v", len(folders), folders) + } +} diff --git a/internal/session/convos.go b/internal/session/convos.go index 319572d..ee9cf7e 100644 --- a/internal/session/convos.go +++ b/internal/session/convos.go @@ -5,11 +5,14 @@ import ( "fmt" "os" "path/filepath" + "regexp" "sort" "strings" + "time" "unicode" "github.com/Coder8124/logos/internal/project" + "github.com/Coder8124/logos/internal/transcript" "github.com/Coder8124/logos/internal/untrusted" "github.com/Coder8124/logos/internal/vault" ) @@ -85,7 +88,7 @@ func Conversations(vaultDir, projectName string) ([]Conversation, error) { flat := en.Rel == en.Name id := convoFolder(c.HostSession) if !flat { - id = strings.Split(en.Rel, "/")[1] + id = FolderID(strings.Split(en.Rel, "/")[1]) } if id == UnattachedFolder { id = stem @@ -302,24 +305,83 @@ func RenderConversation(projectName string, c Conversation) string { // so — a migration that moved files without a number reads as one that might // have lost some. type MigrateReport struct { - Moved int + Moved int + // Recovered is how many of the moved checkpoints had no host_session and + // were given one from their conversation's own receipt. + Recovered int + // Ambiguous names each checkpoint more than one conversation's receipt + // claimed. It is filed on its own rather than under a guess. + Ambiguous []string + // Unattributed is how many moved checkpoints nothing tied to a + // conversation, each given a folder of its own. + Unattributed int + // Named is how many conversation folders were given a readable name. + Named int Skipped []string // ": ", left exactly where they were Failed []string // ": ", the old file still readable } +// Receipt is one host conversation's record of a checkpoint save: Logos's +// answer, naming the file, in that conversation's transcript. +type Receipt = transcript.Receipt + +// ReceiptSource reads every checkpoint receipt on the machine, keyed +// "/". MigrateFlat calls it at most once, and only when a +// checkpoint actually lacks a host_session: reading the host transcripts is +// a walk of gigabytes, and a vault already migrated has nothing to ask it. +type ReceiptSource func() (map[string][]Receipt, error) + +// Receipts is a ReceiptSource for receipts already in hand. +func Receipts(m map[string][]Receipt) ReceiptSource { + return func() (map[string][]Receipt, error) { return m, nil } +} + +// receiptsOnce memoises a ReceiptSource for one migration, error included, +// so a failure is reported against every checkpoint it leaves unattributed +// rather than retried for each. +type receiptsOnce struct { + src ReceiptSource + done bool + m map[string][]Receipt + err error +} + +func (r *receiptsOnce) get() (map[string][]Receipt, error) { + if !r.done && r.src != nil { + r.m, r.err = r.src() + } + r.done = true + return r.m, r.err +} + +// receiptWindow is how far a receipt may sit from its checkpoint's own +// timestamp and still be the save that wrote it. The real gap is seconds; a +// receipt hours away is the file being quoted later by a conversation that +// read it, which is not the one that wrote it. +const receiptWindow = 10 * time.Minute + // removeFlat is os.Remove, swapped in tests: the half of a move that can fail // after the new copy exists is otherwise impossible to arrange. var removeFlat = os.Remove -// MigrateFlat moves every checkpoint still sitting directly in a scope -// directory — the pre-C5 layout — into the conversation folder its own -// host_session names. +// MigrateFlat files every checkpoint not yet in a conversation of its own — +// one still flat in its scope (the pre-C5 layout) or one parked in +// UnattachedFolder — into the conversation folder it belongs to, and names +// every conversation folder after what it was for. +// +// A checkpoint whose frontmatter has no host_session is matched against +// receipts, keyed "/": when exactly one conversation's receipt +// for it sits within receiptWindow of when it was written, that conversation +// wrote it, and its host_session is written into the frontmatter — the +// frontmatter is the truth, so the folder follows from it and survives a +// rebuilt index. Two conversations claiming it is reported, not decided. A +// checkpoint nothing claims gets a folder of its own, named by its stem: one +// shared "_unattached" pile would present unrelated sittings as one. // -// Refile already does this on every `logos index`, but it stops at the first -// error and returns a bare count. This is the explicit, reported version: it -// carries on past a file it cannot move, and names every one it skipped or -// failed, because one unwritable folder must not leave the rest of a vault -// half-migrated without saying which half. +// Refile does the frontmatter-led part on every `logos index`, but stops at +// the first error and returns a bare count. This is the explicit, reported +// version: it carries on past a file it cannot move, and names every one it +// skipped or failed. // // The new copy is written before the old one is removed, so a crash or a // failed remove leaves the checkpoint readable twice — Conversations folds @@ -327,8 +389,18 @@ var removeFlat = os.Remove // bytes is that interrupted move, finished here; a target holding different // bytes is a real collision, and both files are left for a person to look at // rather than one overwriting the other. -func MigrateFlat(vaultDir string, dryRun bool) (MigrateReport, error) { +func MigrateFlat(vaultDir string, dryRun bool, receipts ReceiptSource) (MigrateReport, error) { var rep MigrateReport + if !dryRun { + // Migration makes and renames conversation folders, so it holds the + // lock every writer takes to choose one (convoFolderLock). + g, err := vault.Lock(vaultDir, convoFolderLock) + if err != nil { + return rep, err + } + defer g.Unlock() + } + src := &receiptsOnce{src: receipts} scopes, err := Scopes(vaultDir) if err != nil { return rep, err @@ -340,43 +412,279 @@ func MigrateFlat(vaultDir string, dryRun bool) (MigrateReport, error) { rep.Failed = append(rep.Failed, fmt.Sprintf("%s: %v", scope, err)) continue } - for _, en := range entries { - if en.Rel != en.Name { - continue // already in a folder: Refile's business, not migration's + migrateScope(&rep, dir, scope, entries, dryRun, src) + } + return rep, nil +} + +type migrateMove struct { + en CheckpointEntry + display string + raw []byte + written time.Time // the old file's modification time, kept on the new one + id string + recovered bool + ambiguous bool +} + +type stemmed struct { + stem string + c Checkpoint +} + +func migrateScope(rep *MigrateReport, dir, scope string, entries []CheckpointEntry, dryRun bool, receipts *receiptsOnce) { + var moves []migrateMove + byID := map[string][]stemmed{} // every checkpoint each conversation will hold, for its name + folders := map[string]string{} // id -> the folder it already has + for _, en := range entries { + stem := strings.TrimSuffix(en.Name, ".md") + display := filepath.ToSlash(filepath.Join(CheckpointDir, scope, en.Rel)) + folder := "" + if en.Rel != en.Name { + folder = strings.Split(en.Rel, "/")[1] + } + path := filepath.Join(dir, filepath.FromSlash(en.Rel)) + raw, err := os.ReadFile(path) + var info os.FileInfo + if err == nil { + info, err = os.Stat(path) + } + if err != nil { + rep.Failed = append(rep.Failed, fmt.Sprintf("%s: %v", display, err)) + continue + } + c := ParseCheckpoint(string(raw)) + if folder != "" && folder != UnattachedFolder { + // Already in a conversation's folder: it stays, and counts toward + // the folder's name. + id := FolderID(folder) + if _, ok := folders[id]; !ok { + folders[id] = folder } - display := filepath.ToSlash(filepath.Join(CheckpointDir, scope, en.Name)) - oldPath := filepath.Join(dir, en.Name) - raw, err := os.ReadFile(oldPath) - if err != nil { - rep.Failed = append(rep.Failed, fmt.Sprintf("%s: %v", display, err)) - continue + if !c.Empty() { + byID[id] = append(byID[id], stemmed{stem, c}) } - c := ParseCheckpoint(string(raw)) - if c.Empty() { - rep.Skipped = append(rep.Skipped, display+": empty or torn, left in place") + continue + } + if c.Empty() { + rep.Skipped = append(rep.Skipped, display+": empty or torn, left in place") + continue + } + mv := migrateMove{en: en, display: display, raw: raw, written: info.ModTime()} + if c.HostSession == "" { + all, err := receipts.get() + if err != nil { + // Filed on its own now, it would never be looked at again: + // a rerun finds it in a conversation folder and moves on. + rep.Failed = append(rep.Failed, fmt.Sprintf("%s: left in place, because the host transcripts that say which conversation saved it could not be read: %v", display, err)) continue } - newPath := filepath.Join(dir, ConvosDir, convoFolder(c.HostSession), en.Name) - if existing, err := os.ReadFile(newPath); err == nil && !bytes.Equal(existing, raw) { - rep.Skipped = append(rep.Skipped, display+": a different checkpoint already has its name in "+convoFolder(c.HostSession)+", both kept") - continue - } else if err != nil && !os.IsNotExist(err) { - rep.Failed = append(rep.Failed, fmt.Sprintf("%s: %v", display, err)) - continue - } else if err != nil && !dryRun { - if err := vault.WriteAtomic(newPath, raw); err != nil { - rep.Failed = append(rep.Failed, fmt.Sprintf("%s: writing the new copy: %v", display, err)) + host, claimants := hostFromReceipts(all[scope+"/"+stem], c.TS) + switch { + case claimants > 1: + mv.ambiguous = true + rep.Ambiguous = append(rep.Ambiguous, fmt.Sprintf("%s: %d conversations saved it", display, claimants)) + case host != "": + fixed, ok := withHostSession(raw, host) + if !ok { + rep.Failed = append(rep.Failed, display+": its receipt names a conversation, but its frontmatter could not be found to record it") continue } + mv.raw, c.HostSession, mv.recovered = fixed, host, true + } + } + mv.id = convoFolder(c.HostSession) + if mv.id == UnattachedFolder { + mv.id = stem + } + moves = append(moves, mv) + byID[mv.id] = append(byID[mv.id], stemmed{stem, c}) + } + + // Name every conversation: an existing named folder keeps its name; a + // bare-id one is renamed to carry what it was for; a new one is made so. + names := map[string]string{} + for id, cs := range byID { + about := conversationAbout(cs) + have := folders[id] + if have == "" { + // A folder holding only a harvested transcript has no checkpoint + // to have shown it above; missing it would make a second folder + // for the same conversation. + f, err := existingConvoFolder(dir, id) + if err != nil { + rep.Failed = append(rep.Failed, fmt.Sprintf("%s: reading its conversation folders: %v", filepath.ToSlash(filepath.Join(CheckpointDir, scope)), err)) + return + } + have = f + } + switch { + case have == "": + names[id] = namedConvoFolder(about, id) + case have == id: + want := namedConvoFolder(about, id) + names[id] = have + if want == have { + break + } + if _, err := os.Lstat(filepath.Join(dir, ConvosDir, want)); err == nil { + rep.Skipped = append(rep.Skipped, fmt.Sprintf("%s: %s already exists, so it kept its name", + filepath.ToSlash(filepath.Join(CheckpointDir, scope, ConvosDir, have)), want)) + break } if !dryRun { - if err := removeFlat(oldPath); err != nil { - rep.Failed = append(rep.Failed, fmt.Sprintf("%s: the new copy is written but the old one could not be removed (both are readable; run again to finish): %v", display, err)) - continue + if err := os.Rename(filepath.Join(dir, ConvosDir, have), filepath.Join(dir, ConvosDir, want)); err != nil { + rep.Failed = append(rep.Failed, fmt.Sprintf("%s: naming it %s: %v", + filepath.ToSlash(filepath.Join(CheckpointDir, scope, ConvosDir, have)), want, err)) + break } } - rep.Moved++ + names[id] = want + rep.Named++ + default: + names[id] = have } } - return rep, nil + + unattachedEmptied := false + for _, mv := range moves { + target := names[mv.id] + oldPath := filepath.Join(dir, filepath.FromSlash(mv.en.Rel)) + newPath := filepath.Join(dir, ConvosDir, target, mv.en.Name) + // Its sync copies go with it. A copy is set aside only while its + // original sits beside it, so one left behind would become a second + // checkpoint in every listing — and keep _unattached from emptying. + // Checked before anything is written: one copy must not land on + // another. + copies, err := vault.CopiesOf(filepath.Dir(oldPath), mv.en.Name) + if err != nil { + rep.Failed = append(rep.Failed, fmt.Sprintf("%s: listing its sync copies: %v", mv.display, err)) + continue + } + clash := "" + for _, cp := range copies { + if _, err := os.Lstat(filepath.Join(filepath.Dir(newPath), cp)); err == nil { + clash = cp + break + } + } + if clash != "" { + rep.Skipped = append(rep.Skipped, fmt.Sprintf("%s: its sync copy %s is already in %s, so neither was moved", mv.display, clash, target)) + continue + } + if existing, err := os.ReadFile(newPath); err == nil && !bytes.Equal(existing, mv.raw) { + rep.Skipped = append(rep.Skipped, mv.display+": a different checkpoint already has its name in "+target+", both kept") + continue + } else if err != nil && !os.IsNotExist(err) { + rep.Failed = append(rep.Failed, fmt.Sprintf("%s: %v", mv.display, err)) + continue + } else if err != nil && !dryRun { + if err := vault.WriteAtomic(newPath, mv.raw); err != nil { + rep.Failed = append(rep.Failed, fmt.Sprintf("%s: writing the new copy: %v", mv.display, err)) + continue + } + } + if !dryRun { + // Doctor and the status line read a checkpoint's modification time + // as when it was written; a copy stamped now would make the oldest + // checkpoint in the vault the latest. Kept before the old file goes, + // so a failure here leaves the move for a rerun to finish. + if err := os.Chtimes(newPath, mv.written, mv.written); err != nil { + rep.Failed = append(rep.Failed, fmt.Sprintf("%s: the new copy is written but its time could not be kept (both are readable; run again to finish): %v", mv.display, err)) + continue + } + moved := true + for _, cp := range copies { + if err := os.Rename(filepath.Join(filepath.Dir(oldPath), cp), filepath.Join(filepath.Dir(newPath), cp)); err != nil { + rep.Failed = append(rep.Failed, fmt.Sprintf("%s: the new copy is written but its sync copy %s could not follow (run again to finish): %v", mv.display, cp, err)) + moved = false + break + } + } + if !moved { + continue + } + if err := removeFlat(oldPath); err != nil { + rep.Failed = append(rep.Failed, fmt.Sprintf("%s: the new copy is written but the old one could not be removed (both are readable; run again to finish): %v", mv.display, err)) + continue + } + } + rep.Moved++ + if mv.recovered { + rep.Recovered++ + } else if !mv.ambiguous && mv.id == strings.TrimSuffix(mv.en.Name, ".md") { + rep.Unattributed++ + } + unattachedEmptied = unattachedEmptied || mv.en.Rel != mv.en.Name + } + // An emptied _unattached would sit there as a stray; os.Remove refuses a + // folder that still holds anything, so this cannot take a checkpoint. + if unattachedEmptied && !dryRun { + _ = os.Remove(filepath.Join(dir, ConvosDir, UnattachedFolder)) + } +} + +// conversationAbout is what a conversation's folder is named after: its +// latest checkpoint's intent, inherited the way resume inherits it, else that +// checkpoint's task. +func conversationAbout(cs []stemmed) string { + sort.Slice(cs, func(i, j int) bool { return cs[i].stem > cs[j].stem }) + chain := make([]Checkpoint, len(cs)) + for i := range cs { + chain[i] = cs[i].c + } + if why, _ := IntentFor(chain[0], chain[1:]); strings.TrimSpace(why) != "" { + return why + } + return chain[0].Task +} + +// hostFromReceipts is the one conversation whose receipt for a checkpoint +// written at ts falls within receiptWindow, and how many distinct +// conversations did. A checkpoint with no timestamp takes every receipt. +func hostFromReceipts(rs []Receipt, ts int64) (string, int) { + hosts := map[string]bool{} + host := "" + for _, r := range rs { + if ts != 0 { + gap := r.At.Sub(time.Unix(ts, 0)) + if gap < -receiptWindow || gap > receiptWindow { + continue + } + } + if !hosts[r.HostSession] { + hosts[r.HostSession] = true + host = r.HostSession + } + } + if len(hosts) != 1 { + host = "" + } + return host, len(hosts) +} + +// emptyHostSession is a host_session key with no value, as a hand-edited +// checkpoint may carry it. +var emptyHostSession = regexp.MustCompile(`(?m)^host_session:[ \t]*(?:""|'')?[ \t]*$`) + +// withHostSession gives a checkpoint's frontmatter a host_session line: in +// place of an empty one, since a second key would make the YAML undecodable +// and the checkpoint read as empty; else beside its session line where +// Markdown writes it; else first. Only the frontmatter is searched: the body +// is someone's text and may say anything. +func withHostSession(raw []byte, host string) ([]byte, bool) { + s := string(raw) + end, ok := frontmatterEnd(s) + if !ok { + return nil, false + } + if loc := emptyHostSession.FindStringIndex(s[:end+1]); loc != nil { + return []byte(s[:loc[0]] + "host_session: " + yamlStr(host) + s[loc[1]:]), true + } + line := "host_session: " + yamlStr(host) + "\n" + at := len("---\n") + if i := strings.Index(s[:end+1], "\nsession: "); i >= 0 { + at = i + 1 + strings.Index(s[i+1:], "\n") + 1 + } + return []byte(s[:at] + line + s[at:]), true } diff --git a/internal/session/convos_test.go b/internal/session/convos_test.go index 7f0be2c..bd9059e 100644 --- a/internal/session/convos_test.go +++ b/internal/session/convos_test.go @@ -117,7 +117,7 @@ func TestACrashMidMigrationLeavesTheOldFlatFileReadable(t *testing.T) { flat := flatten(t, dir, "kestrel", commitIn(t, dir, &Checkpoint{Project: "kestrel", Agent: "a", Task: "keep me", HostSession: "sess-k"})) removeFlat = func(string) error { return errors.New("disk went away") } - rep, err := MigrateFlat(dir, false) + rep, err := MigrateFlat(dir, false, nil) removeFlat = os.Remove if err != nil { t.Fatal(err) @@ -133,14 +133,14 @@ func TestACrashMidMigrationLeavesTheOldFlatFileReadable(t *testing.T) { t.Fatalf("a half-moved checkpoint should read as exactly one checkpoint, got %+v", convs) } - rep, err = MigrateFlat(dir, false) + rep, err = MigrateFlat(dir, false, nil) if err != nil || rep.Moved != 1 || len(rep.Failed) != 0 { t.Fatalf("running again did not finish the move: %+v %v", rep, err) } if _, err := os.Stat(flat); !os.IsNotExist(err) { t.Fatal("the stale flat copy survived the rerun") } - if _, err := os.Stat(filepath.Join(dir, CheckpointDir, "kestrel", ConvosDir, "sess-k", filepath.Base(flat))); err != nil { + if _, err := os.Stat(filepath.Join(dir, CheckpointDir, "kestrel", ConvosDir, "keep-me--sess-k", filepath.Base(flat))); err != nil { t.Fatalf("the checkpoint is not in its folder: %v", err) } } @@ -158,14 +158,14 @@ func TestMigratingSessionsCountsWhatItMovedAndKeepsACollidingFile(t *testing.T) torn := filepath.Join(dir, CheckpointDir, "kestrel", "20200101-000000-a.md") os.WriteFile(torn, nil, 0o600) - dry, _ := MigrateFlat(dir, true) + dry, _ := MigrateFlat(dir, true, nil) if dry.Moved != 1 { t.Fatalf("dry run counted %d", dry.Moved) } if _, err := os.Stat(moved); err != nil { t.Fatal("a dry run moved a file") } - rep, err := MigrateFlat(dir, false) + rep, err := MigrateFlat(dir, false, nil) if err != nil || rep.Moved != 1 || len(rep.Skipped) != 2 || len(rep.Failed) != 0 { t.Fatalf("want 1 moved, 2 skipped (collision, torn), got %+v %v", rep, err) } @@ -174,8 +174,9 @@ func TestMigratingSessionsCountsWhatItMovedAndKeepsACollidingFile(t *testing.T) t.Fatalf("%s should have been left alone: %v", p, err) } } - if _, err := os.Stat(filepath.Join(dir, CheckpointDir, "kestrel", ConvosDir, UnattachedFolder, filepath.Base(moved))); err != nil { - t.Fatal("a checkpoint with no host session did not move to _unattached") + own := "plain--" + strings.TrimSuffix(filepath.Base(moved), ".md") + if _, err := os.Stat(filepath.Join(dir, CheckpointDir, "kestrel", ConvosDir, own, filepath.Base(moved))); err != nil { + t.Fatal("a checkpoint with no host session did not move to a folder of its own") } } diff --git a/internal/session/transcript.go b/internal/session/transcript.go index 5f53460..80ba980 100644 --- a/internal/session/transcript.go +++ b/internal/session/transcript.go @@ -96,10 +96,14 @@ func convoDirsFor(vaultDir, folder string) []string { return nil } var out []string + // A folder is matched by the id its name carries, not by the name: the + // readable half before the id is the conversation's intent, which the + // transcript's caller does not know. try := func(scopeDir string) { - d := filepath.Join(scopeDir, ConvosDir, folder) - if fi, err := os.Stat(d); err == nil && fi.IsDir() { - out = append(out, d) + // An unreadable _convos is passed over like an unreadable project + // below: a read here must not fail for one scope's sake. + if f, _ := existingConvoFolder(scopeDir, folder); f != "" { + out = append(out, filepath.Join(scopeDir, ConvosDir, f)) } } for _, p := range projects { diff --git a/internal/transcript/receipts.go b/internal/transcript/receipts.go new file mode 100644 index 0000000..d7aa089 --- /dev/null +++ b/internal/transcript/receipts.go @@ -0,0 +1,175 @@ +package transcript + +import ( + "bufio" + "bytes" + "encoding/json" + "io" + "os" + "regexp" + "strings" + "time" +) + +// Receipt is one checkpoint save as the host saw it happen: which of its +// conversations called Logos, and when the answer came back. +type Receipt struct { + HostSession string + At time.Time +} + +// receiptLine is the first line of every checkpoint receipt Logos has ever +// returned, once the optional "✓ Logos · " marker is off (LOGOS_ANNOUNCE=off +// drops it): "checkpoint saved to logos|brain — sessions/….md", "Checkpoint +// written to sessions/….md in the vault.", the "already saved" of an +// identical retry, and the CLI's "checkpoint written: sessions/….md". It is +// anchored at the start of the line, unlike ingest's handoffReceipt, because +// here the same words later in a line are someone quoting a receipt. +var receiptLine = regexp.MustCompile(`(?i)^checkpoint (?:already )?(?:saved|written)\b[^\x60]*?\x60?sessions/([^\s\x60]+\.md)`) + +// CheckpointReceipts maps "/" to every receipt for that +// checkpoint in the Claude Code and Codex transcripts on this machine. +// +// It exists for checkpoints written before host_session was recorded: the +// frontmatter cannot say which conversation wrote them, but that +// conversation's transcript still holds Logos's answer naming the file. Only +// the answer itself counts — the first line of a Logos tool result. The same +// text in a grep's output, a shell command or an agent's reply is someone +// quoting a receipt, usually from another conversation, and would file the +// checkpoint under the conversation that happened to read it. +// +// The folder part of the path is dropped from the key, because a checkpoint's +// folder is exactly what is being decided; scope and stem name it on their +// own, since claimCheckpoint keeps a stem unique within a scope. +// +// A transcript that cannot be read is passed over and its path returned in +// unreadable, not made fatal: one file locked by a running host, or left +// unreadable by a permissions slip, would otherwise stop every other +// checkpoint from finding its conversation. The caller reports the paths, so +// a checkpoint that stays unattributed because of one is not silent. +func CheckpointReceipts() (out map[string][]Receipt, unreadable []string, err error) { + out = map[string][]Receipt{} + add := func(text, host, ts string) { + key := receiptKey(text) + at, err := time.Parse(time.RFC3339Nano, ts) + if key == "" || host == "" || err != nil { + return + } + out[key] = append(out[key], Receipt{HostSession: host, At: at}) + } + cc, err := claudeCodeReader{}.discover() + if err != nil { + return nil, nil, err + } + for _, path := range cc { + if err := eachReceiptLine(path, func(line []byte) { + var l claudeLine + var m claudeMessage + if json.Unmarshal(line, &l) != nil || l.Type != "user" || json.Unmarshal(l.Message, &m) != nil { + return + } + var blocks []claudeBlock + if json.Unmarshal(m.Content, &blocks) != nil { + return + } + for _, b := range blocks { + if b.Type == "tool_result" { + add(flattenContent(b.Content), l.SessionID, l.Timestamp) + } + } + }); err != nil { + unreadable = append(unreadable, path) + } + } + cx, err := codexReader{}.discover() + if err != nil { + return nil, nil, err + } + for _, path := range cx { + if err := eachReceiptLine(path, func(line []byte) { + var l struct { + Timestamp string `json:"timestamp"` + Type string `json:"type"` + Payload struct { + Type string `json:"type"` + ThreadID string `json:"thread_id"` + Item struct { + Type string `json:"type"` + Result struct { + Content []codexContentItem `json:"content"` + } `json:"result"` + } `json:"item"` + } `json:"payload"` + } + // Codex records an MCP call twice, as an event and as a response + // item; the event alone carries the thread id, so it is the one read. + if json.Unmarshal(line, &l) != nil || l.Type != "event_msg" || l.Payload.Type != "item_completed" || l.Payload.Item.Type != "McpToolCall" { + return + } + if c := l.Payload.Item.Result.Content; len(c) > 0 { + add(c[0].Text, l.Payload.ThreadID, l.Timestamp) + } + }); err != nil { + unreadable = append(unreadable, path) + } + } + return out, unreadable, nil +} + +// receiptKey is "/" for a tool result whose first line is a +// checkpoint receipt, or "" for anything else. +func receiptKey(text string) string { + first, _, _ := strings.Cut(strings.TrimSpace(text), "\n") + m := receiptLine.FindStringSubmatch(strings.TrimPrefix(first, "✓ Logos · ")) + if m == nil { + return "" + } + rel := strings.TrimSuffix(m[1], ".md") + i := strings.LastIndex(rel, "/") + if i <= 0 { + return "" + } + scope, stem := rel[:i], rel[i+1:] + if j := strings.Index(scope, "/_convos/"); j >= 0 { + scope = scope[:j] + } + return scope + "/" + stem +} + +// eachReceiptLine hands fn each line that mentions a receipt at all. The +// substring test runs before any JSON is decoded: a machine's transcripts +// run to gigabytes, and nearly every line is something else. +func eachReceiptLine(path string, fn func([]byte)) error { + f, err := os.Open(path) + if err != nil { + if os.IsNotExist(err) { + return nil // removed between discovery and here + } + return err + } + defer f.Close() + // Lower-case "heckpoint" catches "Checkpoint" and "checkpoint" alike. + needles := [][]byte{[]byte("heckpoint saved"), []byte("heckpoint written"), []byte("heckpoint already saved")} + r := bufio.NewReaderSize(f, 1<<20) + for { + line, err := r.ReadBytes('\n') + if len(line) > 0 && containsAny(line, needles) { + fn(line) + } + if err == io.EOF { + return nil + } + if err != nil { + return err + } + } +} + +func containsAny(line []byte, needles [][]byte) bool { + for _, n := range needles { + if bytes.Contains(line, n) { + return true + } + } + return false +} diff --git a/internal/transcript/receipts_test.go b/internal/transcript/receipts_test.go new file mode 100644 index 0000000..beac880 --- /dev/null +++ b/internal/transcript/receipts_test.go @@ -0,0 +1,100 @@ +package transcript + +import ( + "os" + "path/filepath" + "testing" + "time" +) + +func writeLines(t *testing.T, path string, lines ...string) { + t.Helper() + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + t.Fatal(err) + } + var b []byte + for _, l := range lines { + b = append(b, l...) + b = append(b, '\n') + } + if err := os.WriteFile(path, b, 0o600); err != nil { + t.Fatal(err) + } +} + +func TestCheckpointReceiptsAreReadOnlyFromLogosOwnToolResults(t *testing.T) { + claude, codex := t.TempDir(), t.TempDir() + t.Setenv(LogosClaudeProjectsEnv, claude) + t.Setenv(LogosCodexSessionsEnv, codex) + + writeLines(t, filepath.Join(claude, "-Users-x-kestrel", "sess-a.jsonl"), + // The receipt itself: the first line of a Logos tool result. + `{"type":"user","sessionId":"sess-a","timestamp":"2026-09-01T10:00:01Z","message":{"role":"user","content":[{"type":"tool_result","tool_use_id":"t1","content":[{"type":"text","text":"✓ Logos · checkpoint saved to logos — sessions/kestrel/20260901-100000-claude-code.md"}]}]}}`, + // Announcements off: the same receipt without its marker still counts. + `{"type":"user","sessionId":"sess-a","timestamp":"2026-09-01T11:00:01Z","message":{"role":"user","content":[{"type":"tool_result","tool_use_id":"t2","content":"checkpoint saved to logos — sessions/kestrel/wt/_convos/_unattached/20260901-110000-claude-code.md"}]}}`, + // Before the rename the receipt said brain, sometimes with the path quoted. + `{"type":"user","sessionId":"sess-b","timestamp":"2026-05-01T10:00:01Z","message":{"role":"user","content":[{"type":"tool_result","tool_use_id":"t4","content":[{"type":"text","text":"✓ Logos · checkpoint saved to brain — `+"`"+`sessions/kestrel/20260501-100000-claude-code.md`+"`"+`"}]}]}}`, + // The CLI's own answer, in the output of the shell command that ran it. + `{"type":"user","sessionId":"sess-b","timestamp":"2026-05-02T10:00:01Z","message":{"role":"user","content":[{"type":"tool_result","tool_use_id":"t5","content":"checkpoint written: sessions/kestrel/20260502-100000-cli.md"}]}}`, + // A grep whose output quotes another conversation's receipt is not one. + `{"type":"user","sessionId":"sess-a","timestamp":"2026-09-02T10:00:00Z","message":{"role":"user","content":[{"type":"tool_result","tool_use_id":"t3","content":"log.txt:3: ✓ Logos · checkpoint saved to logos — sessions/kestrel/20260830-090000-claude-code.md"}]}}`, + // Nor is an agent repeating a receipt to the user in its own words. + `{"type":"assistant","sessionId":"sess-a","timestamp":"2026-09-02T10:00:01Z","message":{"role":"assistant","content":[{"type":"text","text":"checkpoint saved to logos — sessions/kestrel/20260830-090000-claude-code.md"}]}}`, + ) + writeLines(t, filepath.Join(codex, "2026", "09", "01", "rollout-2026-09-01T10-00-00-cdx.jsonl"), + `{"timestamp":"2026-09-01T12:00:00.500Z","type":"event_msg","payload":{"type":"item_completed","thread_id":"cdx-1","item":{"type":"McpToolCall","server":"logos","tool":"checkpoint","result":{"content":[{"type":"text","text":"✓ Logos · checkpoint saved to logos — sessions/acme/20260901-120000-codex.md"}]}}}}`, + // A shell command printing the same text is not Logos speaking. + `{"timestamp":"2026-09-01T12:05:00Z","type":"event_msg","payload":{"type":"item_completed","thread_id":"cdx-1","item":{"type":"CommandExecution","aggregated_output":"checkpoint saved to logos — sessions/acme/20260830-000000-codex.md"}}}`, + ) + + got, unreadable, err := CheckpointReceipts() + if err != nil || len(unreadable) != 0 { + t.Fatal(err, unreadable) + } + want := map[string]Receipt{ + "kestrel/20260901-100000-claude-code": {HostSession: "sess-a", At: time.Date(2026, 9, 1, 10, 0, 1, 0, time.UTC)}, + "kestrel/wt/20260901-110000-claude-code": {HostSession: "sess-a", At: time.Date(2026, 9, 1, 11, 0, 1, 0, time.UTC)}, + "kestrel/20260501-100000-claude-code": {HostSession: "sess-b", At: time.Date(2026, 5, 1, 10, 0, 1, 0, time.UTC)}, + "kestrel/20260502-100000-cli": {HostSession: "sess-b", At: time.Date(2026, 5, 2, 10, 0, 1, 0, time.UTC)}, + "acme/20260901-120000-codex": {HostSession: "cdx-1", At: time.Date(2026, 9, 1, 12, 0, 0, 5e8, time.UTC)}, + } + if len(got) != len(want) { + t.Fatalf("want %d receipts, got %d: %+v", len(want), len(got), got) + } + for key, w := range want { + rs := got[key] + if len(rs) != 1 || rs[0].HostSession != w.HostSession || !rs[0].At.Equal(w.At) { + t.Fatalf("%s: want %+v, got %+v", key, w, rs) + } + } +} + +// One transcript Logos cannot open must not cost every other checkpoint its +// conversation: it is passed over and named, and the rest are still read. +func TestAnUnreadableTranscriptIsPassedOverAndNamed(t *testing.T) { + if os.Getuid() == 0 { + t.Skip("root reads a mode-000 file") + } + claude := t.TempDir() + t.Setenv(LogosClaudeProjectsEnv, claude) + t.Setenv(LogosCodexSessionsEnv, t.TempDir()) + writeLines(t, filepath.Join(claude, "-Users-x-kestrel", "sess-a.jsonl"), + `{"type":"user","sessionId":"sess-a","timestamp":"2026-09-01T10:00:01Z","message":{"role":"user","content":[{"type":"tool_result","tool_use_id":"t1","content":"checkpoint saved to logos — sessions/kestrel/20260901-100000-claude-code.md"}]}}`) + locked := filepath.Join(claude, "-Users-x-kestrel", "sess-b.jsonl") + writeLines(t, locked, `{"type":"user","sessionId":"sess-b","timestamp":"2026-09-01T11:00:01Z","message":{"role":"user","content":[{"type":"tool_result","tool_use_id":"t1","content":"checkpoint saved to logos — sessions/kestrel/20260901-110000-claude-code.md"}]}}`) + if err := os.Chmod(locked, 0); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.Chmod(locked, 0o600) }) + + got, unreadable, err := CheckpointReceipts() + if err != nil { + t.Fatalf("one unreadable transcript stopped the scan: %v", err) + } + if len(unreadable) != 1 || unreadable[0] != locked { + t.Fatalf("want the locked transcript named, got %v", unreadable) + } + if len(got["kestrel/20260901-100000-claude-code"]) != 1 { + t.Fatalf("the readable transcript's receipt is lost: %+v", got) + } +}