From 5f7b1279592a2cd743043c54e9e0ab2bdf3219cb Mon Sep 17 00:00:00 2001 From: Wu Sheng Date: Sat, 17 Oct 2026 17:37:58 +0800 Subject: [PATCH] Write nothing of a retired session again A retired session is one whose marker _retired/ is in the root: an empty file named by its id. Nothing makes one yet. The level that retires a session comes in its own change, with its setting and its docs. This change makes every writer leave a retired session alone first, so a session retired once is never collected, derived, parsed or sent again, whatever is left of it. Each writer checks the marker before it creates a session or a chain directory, or takes a lock that would create one. A marker that cannot be read is an error, never taken for no marker, and the writer changes nothing of that session in that pass. - Claude Code, the change recorder and Kiro pass over a retired source session, and lost-source marking skips it. - Provider bodies of a retired session are marked retired in seen.state and never read again. A marker that cannot be read leaves seen.state as it was. - LangSmith: a run whose conversation is decided and retired never waits. It is remembered in that conversation, so its later updates, and runs inside it, are known to be retired too, also after a restart and after the run above it is forgotten. - parse, the metrics deriver and the SD scenario build do nothing for a retired id. A marker the deriver cannot read keeps the first pass's look-back. - push sends no landed file of a retired session. It still reads each round under a retired session's directory by the session the round's header names, and spooled metrics requests still go. - repack and the scenario runner refuse a root holding a retired session, and a destination where the session is retired. The runner copies only into a new directory. - The pass log says each retired session once per process, asz sources names them for every adapter, and asz verify counts them. --- cmd/asz/changes.go | 8 +- cmd/asz/kiro.go | 8 +- cmd/asz/main.go | 41 +++ cmd/asz/provider.go | 4 +- cmd/asz/refresh.go | 29 ++ cmd/asz/retention_test.go | 163 +++++++++++ internal/adapters/claudecode/collect.go | 12 + internal/adapters/claudecode/collect_test.go | 50 ++++ internal/adapters/claudecode/gone.go | 8 + .../adapters/claudecodechanges/collect.go | 13 + .../adapters/claudecodeprovider/collect.go | 75 ++++- .../claudecodeprovider/collect_test.go | 157 ++++++++++ internal/adapters/claudecodeprovider/seen.go | 4 + internal/adapters/kiro/collect.go | 13 + internal/adapters/kiro/kiro_test.go | 37 +++ internal/adapters/langsmith/collect.go | 39 ++- internal/adapters/langsmith/retired_test.go | 266 +++++++++++++++++ internal/export/otlp/push.go | 54 +++- internal/metrics/completion_test.go | 34 +++ internal/metrics/mcp_test.go | 34 +++ internal/metrics/metrics.go | 13 + internal/parse/parse.go | 9 + internal/repack/repack.go | 31 ++ internal/scenario/marker_test.go | 34 +++ internal/scenario/run/retired_test.go | 82 ++++++ internal/scenario/run/run.go | 22 +- internal/scenario/write_sd.go | 8 + internal/storage/retired.go | 79 +++++ internal/storage/retired_test.go | 72 +++++ tests/chain/repack_test.go | 41 +++ tests/chain/retired_test.go | 273 ++++++++++++++++++ 31 files changed, 1694 insertions(+), 19 deletions(-) create mode 100644 internal/adapters/langsmith/retired_test.go create mode 100644 internal/scenario/run/retired_test.go create mode 100644 internal/storage/retired.go create mode 100644 internal/storage/retired_test.go create mode 100644 tests/chain/retired_test.go diff --git a/cmd/asz/changes.go b/cmd/asz/changes.go index d0d2c81..b24bc4c 100644 --- a/cmd/asz/changes.go +++ b/cmd/asz/changes.go @@ -36,7 +36,7 @@ func changesMatch(ad config.Adapter) func(claudecodechanges.Session) bool { } // cmdSourcesChanges lists the sessions the changes adapter can see. -func cmdSourcesChanges(_ *config.Config, ad config.Adapter, _ bool) error { +func cmdSourcesChanges(cfg *config.Config, ad config.Adapter, _ bool) error { root, err := claudecodechanges.ResolveSourceRoot(ad.SourceRoot) if err != nil { return err @@ -70,5 +70,9 @@ func cmdSourcesChanges(_ *config.Config, ad config.Adapter, _ bool) error { return err } fmt.Printf("\n%d session(s) with change records\n", len(sessions)) - return nil + ids := make([]string, 0, len(sessions)) + for _, s := range sessions { + ids = append(ids, s.ID) + } + return sayRetired(cfg, ids) } diff --git a/cmd/asz/kiro.go b/cmd/asz/kiro.go index a759ae9..7e68593 100644 --- a/cmd/asz/kiro.go +++ b/cmd/asz/kiro.go @@ -30,7 +30,7 @@ import ( // cmdSourcesKiro lists the Kiro sessions the adapter can see, and what it // sees and does not collect. -func cmdSourcesKiro(_ *config.Config, ad config.Adapter, _ bool) error { +func cmdSourcesKiro(cfg *config.Config, ad config.Adapter, _ bool) error { root, err := kiro.ResolveSourceRoot(ad.SourceRoot) if err != nil { return err @@ -76,7 +76,11 @@ func cmdSourcesKiro(_ *config.Config, ad config.Adapter, _ bool) error { return err } fmt.Printf("\n%d V3 session(s)\n", len(sessions)) - return nil + ids := make([]string, 0, len(sessions)) + for _, s := range sessions { + ids = append(ids, s.ID) + } + return sayRetired(cfg, ids) } // notCollected says, one line each, what Kiro wrote that this adapter does diff --git a/cmd/asz/main.go b/cmd/asz/main.go index f6174c9..237f384 100644 --- a/cmd/asz/main.go +++ b/cmd/asz/main.go @@ -356,6 +356,38 @@ func cmdSources(cfg *config.Config, ad config.Adapter, once bool) error { fmt.Println(m) } } + ids := make([]string, 0, len(sessions)) + for _, s := range sessions { + ids = append(ids, s.ID) + } + return sayRetired(cfg, ids) +} + +// sayRetired lists the sessions a source holds that are retired in the +// storage root. They are not collected again, whatever the source holds of +// them, and a person reading the source's list would not otherwise know. +func sayRetired(cfg *config.Config, ids []string) error { + zoneRoot, err := cfg.ResolvedRoot() + if err != nil { + return err + } + zone := storage.NewZone(zoneRoot) + var retired []string + for _, id := range ids { + if ok, err := storage.Retired(zone, id); err != nil { + return err + } else if ok { + retired = append(retired, id) + } + } + if len(retired) == 0 { + return nil + } + sort.Strings(retired) + fmt.Printf("\n%d session(s) are retired in %s and are not collected again:\n", len(retired), zoneRoot) + for _, id := range retired { + fmt.Printf(" %s\n", id) + } return nil } @@ -601,6 +633,15 @@ func cmdVerify(cfg *config.Config, _ config.Adapter, _ bool) error { fmt.Printf("%d provider body(ies) of a removal in progress no longer rebuild: the removal deleted a body they are built from first, and the next pass deletes them too\n", removing) } + // A retired session keeps nothing to verify, only its marker. What is + // left of one whose retirement has not finished is checked above, as + // any session is. + if retired, err := storage.RetiredSessions(storage.NewZone(zoneRoot)); err != nil { + return err + } else if len(retired) > 0 { + fmt.Printf("%d session(s) are retired: nothing of them is kept, and none is collected again\n", len(retired)) + } + // A conversation's rounds are checked too. They are a digest chain, so a // missing or edited round is detectable in a way landed files are not, and // the check costs one pass over a directory that is far smaller than the data. diff --git a/cmd/asz/provider.go b/cmd/asz/provider.go index 373413f..bf599c2 100644 --- a/cmd/asz/provider.go +++ b/cmd/asz/provider.go @@ -118,7 +118,7 @@ func landedSession(zone *storage.Zone, id string) (claudecode.Session, bool) { // cmdSourcesProvider says what the provider adapter can see: the directory, // how many bodies it holds, and which sessions the requests name. -func cmdSourcesProvider(_ *config.Config, ad config.Adapter, _ bool) error { +func cmdSourcesProvider(cfg *config.Config, ad config.Adapter, _ bool) error { root, err := claudecodeprovider.ResolveSourceRoot(ad.SourceRoot) if err != nil { return err @@ -158,7 +158,7 @@ func cmdSourcesProvider(_ *config.Config, ad config.Adapter, _ bool) error { if n := len(ids); n > 0 { fmt.Printf("\n%d session(s) named by requests; a response is joined to its session when collected\n", n) } - return nil + return sayRetired(cfg, ids) } // sessionOf reads the session a request body names, or nothing. diff --git a/cmd/asz/refresh.go b/cmd/asz/refresh.go index 18d600c..4cea624 100644 --- a/cmd/asz/refresh.go +++ b/cmd/asz/refresh.go @@ -90,6 +90,9 @@ func sameDir(a, b string) bool { // next; with none, asz collect prints the same numbers and nothing else // changes. The page itself never writes. type refresher struct { + // saidRetired names the retired sessions this process has said it found + // evidence of, so each is said once. + saidRetired map[string]bool // srv is the page to keep current, or nil when nothing is served. srv *view.Server pusher *otlp.Pusher @@ -523,6 +526,8 @@ func (r *refresher) pass() error { // dropped counts the LangSmith runs not collected this pass because their // conversation could not be told. var dropped int + // retired names the retired sessions a collector found evidence of. + var retired []string changed := map[string]bool{} // A removal an earlier pass left in progress is finished first, before @@ -561,6 +566,7 @@ func (r *refresher) pass() error { for _, id := range cs.Changed { changed[id] = true } + retired = append(retired, cs.Retired...) } // Each recorder directory is collected on its own: one that is absent, // or fails, leaves the others collected. @@ -588,6 +594,7 @@ func (r *refresher) pass() error { for _, id := range ch.Changed { changed[id] = true } + retired = append(retired, ch.Retired...) } // Kiro may not be installed, or not have run on its V3 engine yet, and @@ -616,6 +623,7 @@ func (r *refresher) pass() error { for _, id := range ks.Changed { changed[id] = true } + retired = append(retired, ks.Retired...) // Sessions that are there and are not read are said once, and // again when the count changes, not on every pass. if said := strings.Join(notCollected(ks.Skipped), "\n"); said != r.kiroSkipped { @@ -654,6 +662,7 @@ func (r *refresher) pass() error { conflicts += ps.Conflicts busy += ps.Busy waiting, unreadable = ps.Waiting, ps.Unreadable + retired = append(retired, ps.Retired...) for _, id := range ps.Changed { changed[id] = true } @@ -672,7 +681,9 @@ func (r *refresher) pass() error { for _, id := range landedRuns.Sessions { changed[id] = true } + retired = append(retired, landedRuns.Retired...) } + r.sayRetired(retired) // What this pass landed, plus whatever an earlier pass could not take // the lock for or failed to parse. @@ -959,3 +970,21 @@ func sayUnread(unread map[string]int) { id, unread[id]) } } + +// sayRetired says, once a process for each retired session, that a collector +// found evidence of it and did not collect it. A source the runtime still +// holds is found on every pass, and the same line on every pass would make +// the other lines of the log hard to find. +func (r *refresher) sayRetired(retired []string) { + sort.Strings(retired) + for _, id := range retired { + if r.saidRetired[id] { + continue + } + if r.saidRetired == nil { + r.saidRetired = map[string]bool{} + } + r.saidRetired[id] = true + fmt.Fprintf(os.Stderr, " retired: %s: found and not collected; a retired session is never collected again\n", id) + } +} diff --git a/cmd/asz/retention_test.go b/cmd/asz/retention_test.go index 5fa1f39..7547a7d 100644 --- a/cmd/asz/retention_test.go +++ b/cmd/asz/retention_test.go @@ -24,8 +24,10 @@ import ( "testing" "time" + "github.com/apache/skywalking-ai-sessionizer/internal/adapters/kiro" "github.com/apache/skywalking-ai-sessionizer/internal/config" "github.com/apache/skywalking-ai-sessionizer/internal/export/otlp" + "github.com/apache/skywalking-ai-sessionizer/internal/scenario" "github.com/apache/skywalking-ai-sessionizer/internal/storage" "github.com/apache/skywalking-ai-sessionizer/internal/view" "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata" @@ -342,3 +344,164 @@ func TestLateMetricsAreSaidByThePipelineAndByPush(t *testing.T) { t.Fatalf("asz push did not say it:\n%s", out) } } + +// retiredRoot is a root whose growth conversation a pipeline collected and +// parsed, then retired, with nothing left of it but its marker. Its source +// is still there. +func retiredRoot(t *testing.T) (string, string, *refresher) { + t.Helper() + root, b := scenarioRoot(t, "growth.yaml") + if err := os.RemoveAll(filepath.Join(root, storage.ScenarioDir)); err != nil { + t.Fatal(err) + } + ref := pipelineOver(t, root, bothAdapters(root, "")) + stdoutOf(t, func() { + if err := ref.pass(); err != nil { + t.Fatal(err) + } + }) + zone := storage.NewZone(root) + if err := os.MkdirAll(filepath.Join(root, storage.RetiredDir), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(zone.RetiredPath(b.Session), nil, 0o444); err != nil { + t.Fatal(err) + } + for _, dir := range []string{zone.SessionDir(b.Session), filepath.Join(root, "_conversations", b.Session)} { + if err := storage.DeleteTree(dir, nil); err != nil { + t.Fatal(err) + } + } + return root, b.Session, ref +} + +// A pass says once a process that it found a retired session in its sources +// and did not collect it, and the next pass does not say it again. +func TestAPassSaysARetiredSessionOnce(t *testing.T) { + root, session, ref := retiredRoot(t) + line := " retired: " + session + ": found and not collected" + var out string + stdoutOf(t, func() { + out = stderrOf(t, func() { + if err := ref.pass(); err != nil { + t.Fatal(err) + } + }) + }) + if n := strings.Count(out, line); n != 1 { + t.Fatalf("the pass said it %d times, want once:\n%s", n, out) + } + stdoutOf(t, func() { + out = stderrOf(t, func() { + if err := ref.pass(); err != nil { + t.Fatal(err) + } + }) + }) + if strings.Contains(out, line) { + t.Fatalf("the next pass said it again:\n%s", out) + } + if _, err := os.Stat(storage.NewZone(root).SessionDir(session)); !os.IsNotExist(err) { + t.Fatalf("the pass made the retired session's directory again: %v", err) + } +} + +// asz verify says how many sessions are retired, asz sources names a source +// session that is retired, and asz repack refuses a root that holds one, +// whichever session it is asked for, before it makes anything. +func TestCommandsSayWhatIsRetired(t *testing.T) { + root, session, _ := retiredRoot(t) + saved := positional + t.Cleanup(func() { positional = saved }) + cfg := config.Default() + cfg.Storage.Root = root + + positional = nil + out := stdoutOf(t, func() { + if err := cmdVerify(cfg, config.Adapter{}, false); err != nil { + t.Fatalf("verify: %v", err) + } + }) + if !strings.Contains(out, "1 session(s) are retired: nothing of them is kept, and none is collected again") { + t.Errorf("verify does not say the session is retired:\n%s", out) + } + + ad := config.Adapter{Name: config.AdapterClaudeCodeLocal, Enabled: true, SourceRoot: filepath.Join(root, "_source")} + out = stdoutOf(t, func() { + if err := cmdSources(cfg, ad, false); err != nil { + t.Fatalf("sources: %v", err) + } + }) + if !strings.Contains(out, "1 session(s) are retired in "+root+" and are not collected again:\n "+session) { + t.Errorf("sources does not name the retired session:\n%s", out) + } + + for _, want := range []string{"", "another-session"} { + dest := filepath.Join(t.TempDir(), "dest") + positional = []string{dest} + if want != "" { + positional = append(positional, want) + } + var err error + stdoutOf(t, func() { err = cmdRepack(cfg, config.Adapter{}, false) }) + if err == nil || !strings.Contains(err.Error(), "retired session") { + t.Errorf("a repack of a root holding a retired session, asked for %q, returned %v", want, err) + } + if pathExists(dest) { + t.Errorf("a refused repack, asked for %q, made its destination", want) + } + } +} + +// Each adapter's asz sources names a session of its source that is retired. +func TestEverySourcesListNamesARetiredSession(t *testing.T) { + saved := positional + t.Cleanup(func() { positional = saved }) + kiroSource := filepath.Join("..", "..", "internal", "adapters", "kiro", "testdata") + kiroSessions, _, _, err := kiro.Discover(kiroSource) + if err != nil || len(kiroSessions) == 0 { + t.Fatalf("%v: no Kiro fixture to list", err) + } + for _, c := range []struct { + name string + scenario string + adapter func(root string) config.Adapter + session func(b string) string + }{ + {"changes", "all-kinds.yaml", func(root string) config.Adapter { + return config.Adapter{Name: config.AdapterChanges, Enabled: true, SourceRoot: filepath.Join(root, "_source", "plugins", "data")} + }, func(b string) string { return b }}, + {"provider", "provider-bodies-long.yaml", func(root string) config.Adapter { + return config.Adapter{Name: config.AdapterClaudeCodeProvider, Enabled: true, SourceRoot: filepath.Join(root, "_source", scenario.ProviderBodyDir)} + }, func(b string) string { return b }}, + {"kiro", "", func(string) config.Adapter { + return config.Adapter{Name: config.AdapterKiroLocal, Enabled: true, SourceRoot: kiroSource} + }, func(string) string { return kiroSessions[0].ID }}, + } { + t.Run(c.name, func(t *testing.T) { + root, built := t.TempDir(), "" + if c.scenario != "" { + r, b := scenarioRoot(t, c.scenario) + root, built = r, b.Session + } + session := c.session(built) + zone := storage.NewZone(root) + if err := os.MkdirAll(filepath.Join(root, storage.RetiredDir), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(zone.RetiredPath(session), nil, 0o444); err != nil { + t.Fatal(err) + } + cfg := config.Default() + cfg.Storage.Root = root + out := stdoutOf(t, func() { + if err := cmdSources(cfg, c.adapter(root), false); err != nil { + t.Fatalf("sources: %v", err) + } + }) + if !strings.Contains(out, "are retired in "+root+" and are not collected again:\n "+session) { + t.Errorf("sources does not name the retired session %s:\n%s", session, out) + } + }) + } +} diff --git a/internal/adapters/claudecode/collect.go b/internal/adapters/claudecode/collect.go index e286f00..0442da6 100644 --- a/internal/adapters/claudecode/collect.go +++ b/internal/adapters/claudecode/collect.go @@ -91,6 +91,9 @@ type Stats struct { // interrupted after landing but before the index was written. Reindexed int Errors []error + // Retired names the sessions discovery found that are retired. Their + // sources are not read, and nothing of them is written. + Retired []string // Changed lists the sessions whose landed data or index moved this pass. // A caller that assembles structure afterwards needs only these; parsing // every session again measured 3.3 s on a 43-session corpus with nothing @@ -130,6 +133,15 @@ func (c *Collector) CollectAll(filter func(Session) bool) (*Stats, error) { if filter != nil && !filter(s) { continue } + // Before its lock, which would make its directory again. + if retired, err := storage.Retired(c.Zone, s.ID); err != nil || retired { + if err != nil { + st.Errors = append(st.Errors, fmt.Errorf("session %s: %w", s.ID, err)) + } else { + st.Retired = append(st.Retired, s.ID) + } + continue + } st.Sessions++ if err := c.collectSession(s, st); err != nil { st.Errors = append(st.Errors, fmt.Errorf("session %s: %w", s.ID, err)) diff --git a/internal/adapters/claudecode/collect_test.go b/internal/adapters/claudecode/collect_test.go index dcb43bd..d5309c6 100644 --- a/internal/adapters/claudecode/collect_test.go +++ b/internal/adapters/claudecode/collect_test.go @@ -321,3 +321,53 @@ func TestSourceUnderNewInodeResumes(t *testing.T) { t.Fatalf("third pass: landed=%d records=%d err=%v", st.SourcesLanded, st.Records, err) } } + +// TestARetiredSessionIsNotCollectedOrMarked. A session that was retired +// while what is left of it is still there grows in its source, as a resumed +// session does, and is not collected. Its source then goes, and its cursor +// is not moved to say so: nothing of a retired session is written. +func TestARetiredSessionIsNotCollectedOrMarked(t *testing.T) { + src, root := t.TempDir(), t.TempDir() + zone := storage.NewZone(root) + main := filepath.Join(src, "-proj-a", tSess+".jsonl") + mk(t, main, "{\"uuid\":\"m1\"}\n") + col := claudecode.New(src, zone, 0) + if _, err := col.CollectAll(nil); err != nil { + t.Fatal(err) + } + if err := os.MkdirAll(filepath.Join(root, storage.RetiredDir), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(zone.RetiredPath(tSess), nil, 0o444); err != nil { + t.Fatal(err) + } + cursor := filepath.Join(root, tSess, "streams", "main", "transcript.cursor") + before, err := os.ReadFile(cursor) + if err != nil { + t.Fatal(err) + } + f, err := os.OpenFile(main, os.O_APPEND|os.O_WRONLY, 0) + if err != nil { + t.Fatal(err) + } + if _, err := f.WriteString("{\"uuid\":\"m2\"}\n"); err != nil { + t.Fatal(err) + } + f.Close() + st, err := claudecode.New(src, zone, 0).CollectAll(nil) + if err != nil || len(st.Errors) > 0 { + t.Fatalf("%v %v", err, st.Errors) + } + if st.SourcesLanded != 0 || len(st.Retired) != 1 || st.Retired[0] != tSess { + t.Fatalf("landed %d, retired %v; want nothing landed and %s retired", st.SourcesLanded, st.Retired, tSess) + } + if err := os.Remove(main); err != nil { + t.Fatal(err) + } + if _, err := claudecode.New(src, zone, 0).CollectAll(nil); err != nil { + t.Fatal(err) + } + if after, err := os.ReadFile(cursor); err != nil || string(after) != string(before) { + t.Fatalf("the retired session's cursor changed: %v\nbefore %s\nafter %s", err, before, after) + } +} diff --git a/internal/adapters/claudecode/gone.go b/internal/adapters/claudecode/gone.go index 9a946bd..5e371b1 100644 --- a/internal/adapters/claudecode/gone.go +++ b/internal/adapters/claudecode/gone.go @@ -199,6 +199,14 @@ func (c *Collector) markLostSessions(found map[string]bool, filter func(Session) if !it.IsDir() || found[id] || !IsSessionID(id) || c.wasChecked(id) { continue } + // What is left of a retired session while its retirement finishes is + // not written to. + if retired, err := storage.Retired(c.Zone, id); err != nil || retired { + if err != nil { + st.Errors = append(st.Errors, fmt.Errorf("session %s: %w", id, err)) + } + continue + } done, err := c.markLostSession(id, filter, st, now) if err != nil { st.Errors = append(st.Errors, fmt.Errorf("session %s: %w", id, err)) diff --git a/internal/adapters/claudecodechanges/collect.go b/internal/adapters/claudecodechanges/collect.go index 7de37d5..a7f3ed0 100644 --- a/internal/adapters/claudecodechanges/collect.go +++ b/internal/adapters/claudecodechanges/collect.go @@ -57,6 +57,9 @@ type Stats struct { Indexed int Reindexed int Errors []error + // Retired names the sessions found that are retired. Their records are + // not read, and nothing of them is written, a cursor included. + Retired []string // Changed lists the sessions whose landed data or index moved this pass. Changed []string } @@ -89,6 +92,16 @@ func (c *Collector) CollectAll(filter func(Session) bool) (*Stats, error) { if filter != nil && !filter(s) { continue } + // Before anything: claiming an older cursor saves it without + // landing a record, and taking the lock makes the directory. + if retired, err := storage.Retired(c.Zone, s.ID); err != nil || retired { + if err != nil { + st.Errors = append(st.Errors, fmt.Errorf("session %s: %w", s.ID, err)) + } else { + st.Retired = append(st.Retired, s.ID) + } + continue + } st.Sessions++ if err := c.collectSession(s, st); err != nil { st.Errors = append(st.Errors, fmt.Errorf("session %s: %w", s.ID, err)) diff --git a/internal/adapters/claudecodeprovider/collect.go b/internal/adapters/claudecodeprovider/collect.go index 0d8bb42..4e5bb13 100644 --- a/internal/adapters/claudecodeprovider/collect.go +++ b/internal/adapters/claudecodeprovider/collect.go @@ -69,6 +69,10 @@ type Collector struct { Zone *storage.Zone MaxDelta int64 Now func() time.Time + + // retired holds, for one pass, what the marker of each session asked + // about said. + retired map[string]retirement } // Stats summarises one collection pass. The fields the local adapter reports @@ -92,6 +96,9 @@ type Stats struct { // that names a session, is never claimed. Waiting int Unreadable int + // Retired names the retired sessions bodies were found for. None of + // them lands. + Retired []string } // Complete reports whether the pass collected everything it could. @@ -146,6 +153,7 @@ func (c *Collector) CollectAll(filter Filter) (*Stats, error) { return nil, err } now := c.Now() + c.retired = map[string]retirement{} // A line whose file is gone and whose session is gone describes nothing // any more: a body pruned from the directory after its session was @@ -178,6 +186,21 @@ func (c *Collector) CollectAll(filter Filter) (*Stats, error) { st.SourcesSeen++ f := &file{name: it.Name(), path: filepath.Join(c.SourceRoot, it.Name()), size: info.Size(), mtime: info.ModTime().UnixNano()} f.e = sn.entries[f.name] + // A body of a retired session is not read again, whatever became of + // its file: its session's directory is gone, so a landed line would + // otherwise land it again. + if f.e != nil && f.e.session != "" { + switch c.retiredOf(f.e.session, st) { + case retired: + if f.e.state != stateRetired { + f.e.state = stateRetired + sn.dirty = true + } + continue + case unknown: + continue + } + } if f.e != nil && f.e.size == f.size && f.e.mtime == f.mtime { switch f.e.state { case stateLanded: @@ -215,7 +238,7 @@ func (c *Collector) CollectAll(filter Filter) (*Stats, error) { continue } checked[e.session] = true - if _, fresh := bySession[e.session]; !fresh && c.indexBehind(e.session) { + if _, fresh := bySession[e.session]; !fresh && c.retiredOf(e.session, st) == live && c.indexBehind(e.session) { bySession[e.session] = nil } } @@ -238,6 +261,40 @@ func (c *Collector) CollectAll(filter Filter) (*Stats, error) { return st, nil } +// retirement is what a session's retired marker says. +type retirement int + +const ( + live retirement = iota + retired + // unknown is a marker that could not be read. Nothing of the session + // is written that pass, and nothing is recorded of it either: taking it + // for retired would mark its bodies retired, and they would land no + // more once the marker reads again. + unknown +) + +// retiredOf says whether a session is retired, asking its marker once a pass. +func (c *Collector) retiredOf(session string, st *Stats) retirement { + if r, asked := c.retired[session]; asked { + return r + } + if c.retired == nil { + c.retired = map[string]retirement{} + } + r := live + switch is, err := storage.Retired(c.Zone, session); { + case err != nil: + st.Errors = append(st.Errors, fmt.Errorf("session %s: %w", session, err)) + r = unknown + case is: + st.Retired = append(st.Retired, session) + r = retired + } + c.retired[session] = r + return r +} + // indexBehind reports whether a session's index covers less than its landed // provider files. func (c *Collector) indexBehind(session string) bool { @@ -342,7 +399,9 @@ func (c *Collector) attribute(fresh []*file, sn *seen, filter Filter, st *Stats) } var requests []near for _, e := range sn.entries { - if e.state == stateLanded && RoleOf(e.name) == providerbody.RoleRequest { + // A retired request still names its session for the response it + // was answered with, so the response is known to be retired too. + if (e.state == stateLanded || e.state == stateRetired) && RoleOf(e.name) == providerbody.RoleRequest { addNamed(e.keys.PreviousRequest, e.session) requests = append(requests, near{e.session, e.mtime}) } @@ -401,6 +460,18 @@ func (c *Collector) attribute(fresh []*file, sn *seen, filter Filter, st *Stats) } } state := stateWaiting + if session != "" { + switch c.retiredOf(session, st) { + case retired: + sn.entries[f.name] = &entry{name: f.name, size: f.size, mtime: f.mtime, sha: providerbody.Digest(f.body)[:12], + state: stateRetired, session: session, keys: f.keys} + sn.dirty = true + continue + case unknown: + // Left as it was, and read again next pass. + continue + } + } if session != "" { switch filter(session) { case Collect: diff --git a/internal/adapters/claudecodeprovider/collect_test.go b/internal/adapters/claudecodeprovider/collect_test.go index bcadc6e..4032229 100644 --- a/internal/adapters/claudecodeprovider/collect_test.go +++ b/internal/adapters/claudecodeprovider/collect_test.go @@ -25,6 +25,7 @@ import ( "io" "os" "path/filepath" + "runtime" "strings" "testing" "time" @@ -560,3 +561,159 @@ func TestAnIndexAheadOfItsStateIsBuiltAgain(t *testing.T) { t.Fatalf("%d records indexed, want the 3 bodies landed", len(seen)) } } + +// retireAll marks a session retired and takes its directory away, as a +// finished retirement leaves it. +func (f *fixture) retireAll(session string) { + f.t.Helper() + if err := os.MkdirAll(filepath.Join(f.zone.Root(), storage.RetiredDir), 0o755); err != nil { + f.t.Fatal(err) + } + if err := os.WriteFile(f.zone.RetiredPath(session), nil, 0o444); err != nil { + f.t.Fatal(err) + } + if err := storage.DeleteTree(f.zone.SessionDir(session), nil); err != nil { + f.t.Fatal(err) + } +} + +// TestARetiredSessionsBodiesAreNotReadAgain. Claude Code never deletes a +// body file, so a retired session's bodies stay in the source directory. A +// pass marks their lines retired without reading them, every pass after it +// passes them over the same way, and a new request naming the session is +// retired too. None of them lands, and the session's directory is not made +// again. +func TestARetiredSessionsBodiesAreNotReadAgain(t *testing.T) { + if runtime.GOOS == "windows" || os.Geteuid() == 0 { + t.Skip("a file that cannot be read needs a user that is not root, and not Windows") + } + f := newFixture(t) + old := f.request(sessionA, "", 2) + f.pass(all) + f.retireAll(sessionA) + // Not to be read: a pass that read it would fail. + if err := os.Chmod(filepath.Join(f.src, old), 0); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = os.Chmod(filepath.Join(f.src, old), 0o644) }) + for pass := 0; pass < 2; pass++ { + st := f.pass(all) + if st.SourcesLanded != 0 || len(st.Retired) != 1 || st.Retired[0] != sessionA { + t.Fatalf("pass %d: landed %d, retired %v; want nothing landed and %s retired", pass, st.SourcesLanded, st.Retired, sessionA) + } + } + added := f.request(sessionA, "", 3) + if st := f.pass(all); st.SourcesLanded != 0 { + t.Fatalf("a new request of a retired session landed %d", st.SourcesLanded) + } + if _, err := os.Stat(f.zone.SessionDir(sessionA)); !os.IsNotExist(err) { + t.Fatalf("the retired session's directory was made again: %v", err) + } + seen, err := os.ReadFile(f.zone.ProviderStatePath()) + if err != nil { + t.Fatal(err) + } + for _, name := range []string{old, added} { + if !strings.Contains(string(seen), name+"\t") || !strings.Contains(lineOf(string(seen), name), "\tretired\t") { + t.Errorf("%s: its line is %q, want it retired", name, lineOf(string(seen), name)) + } + } +} + +func lineOf(table, name string) string { + for _, line := range strings.Split(table, "\n") { + if strings.HasPrefix(line, name+"\t") { + return line + } + } + return "" +} + +// TestWhatIsLeftOfARetiredSessionIsNotWritten. A retirement that has not +// finished leaves the session's directory, here with no index, which a pass +// would build again for a session it lands into. It is not built. A response +// that comes after the retirement, for a request the retired session's +// request names, is known to be retired by that request, and does not wait. +func TestWhatIsLeftOfARetiredSessionIsNotWritten(t *testing.T) { + f := newFixture(t) + f.request(sessionA, "req_9", 2) + f.pass(all) + if err := os.MkdirAll(filepath.Join(f.zone.Root(), storage.RetiredDir), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(f.zone.RetiredPath(sessionA), nil, 0o444); err != nil { + t.Fatal(err) + } + if err := storage.DeleteTree(f.zone.IndexDir(sessionA), nil); err != nil { + t.Fatal(err) + } + f.pass(all) + if _, err := os.Stat(f.zone.IndexDir(sessionA)); !os.IsNotExist(err) { + t.Fatalf("the retired session's index was built again: %v", err) + } + resp := f.response("req_9", "msg_9") + if st := f.pass(all); st.SourcesLanded != 0 || st.Waiting != 0 { + t.Fatalf("the response landed %d and waits %d; want it retired", st.SourcesLanded, st.Waiting) + } + seen, err := os.ReadFile(f.zone.ProviderStatePath()) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(lineOf(string(seen), resp), "\tretired\t") { + t.Errorf("the response's line is %q, want it retired", lineOf(string(seen), resp)) + } +} + +// TestAMarkerThatCannotBeReadChangesNothing. While a session's marker cannot +// be read, the session is not written, and nothing is recorded of it: its +// landed body keeps its line, and a new body waits unrecorded. Once the +// marker reads again, the new body lands, and the landed one still reports a +// change to its bytes, as a live session's body does. +func TestAMarkerThatCannotBeReadChangesNothing(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("a file where a directory should be reads as not there on Windows") + } + f := newFixture(t) + old := f.request(sessionA, "", 2) + f.pass(all) + before, err := os.ReadFile(f.zone.ProviderStatePath()) + if err != nil { + t.Fatal(err) + } + blocked := filepath.Join(f.zone.Root(), storage.RetiredDir) + if err := os.WriteFile(blocked, nil, 0o644); err != nil { + t.Fatal(err) + } + added := f.request(sessionA, "", 3) + st, err := f.col.CollectAll(all) + if err != nil { + t.Fatal(err) + } + if len(st.Errors) == 0 || st.SourcesLanded != 0 { + t.Fatalf("errors %v, landed %d: a marker that cannot be read must be an error, and nothing land", st.Errors, st.SourcesLanded) + } + after, err := os.ReadFile(f.zone.ProviderStatePath()) + if err != nil { + t.Fatal(err) + } + if string(after) != string(before) { + t.Fatalf("the table changed while the marker could not be read:\nbefore %s\nafter %s", before, after) + } + if err := os.Remove(blocked); err != nil { + t.Fatal(err) + } + if st := f.pass(all); st.SourcesLanded != 1 { + t.Fatalf("landed %d once the marker read again, want the new body", st.SourcesLanded) + } + if _, ok := f.landed(sessionA)[added]; !ok { + t.Fatal("the new body did not land") + } + f.write(old, []byte(`{"changed":true}`)) + st, err = f.col.CollectAll(all) + if err != nil { + t.Fatal(err) + } + if st.Conflicts == 0 { + t.Fatal("a landed body changed after the marker could not be read is no longer reported") + } +} diff --git a/internal/adapters/claudecodeprovider/seen.go b/internal/adapters/claudecodeprovider/seen.go index 1f5bef0..29b88fe 100644 --- a/internal/adapters/claudecodeprovider/seen.go +++ b/internal/adapters/claudecodeprovider/seen.go @@ -51,6 +51,10 @@ const ( // stateChanged: the file changed after it landed. The line keeps the // digest that landed, so putting the bytes back ends it. stateChanged = "changed" + // stateRetired: the body's session is retired. Nothing of it lands + // again, and Claude Code never deletes a body file, so the line is what + // keeps each pass from reading the file again. It goes with its file. + stateRetired = "retired" ) // seenSchema is the version of the state file. diff --git a/internal/adapters/kiro/collect.go b/internal/adapters/kiro/collect.go index cf8a429..931117e 100644 --- a/internal/adapters/kiro/collect.go +++ b/internal/adapters/kiro/collect.go @@ -85,6 +85,9 @@ type Stats struct { // Skipped is what discovery found and does not collect. Skipped Skipped Errors []error + // Retired names the sessions found that are retired. Their files, the + // sub-executions included, are not read, and nothing of them is written. + Retired []string // Changed lists the sessions whose landed data or index moved this pass. Changed []string } @@ -116,6 +119,16 @@ func (c *Collector) CollectAll(filter func(Session) bool) (*Stats, error) { if filter != nil && !filter(s) { continue } + // Before its lock, which would make its directory again. A + // sub-execution carries its parent's id, so it goes with it. + if retired, err := storage.Retired(c.Zone, s.ID); err != nil || retired { + if err != nil { + st.Errors = append(st.Errors, fmt.Errorf("session %s: %w", s.ID, err)) + } else { + st.Retired = append(st.Retired, s.ID) + } + continue + } st.Sessions++ if err := c.collectSession(s, st); err != nil { st.Errors = append(st.Errors, fmt.Errorf("session %s: %w", s.ID, err)) diff --git a/internal/adapters/kiro/kiro_test.go b/internal/adapters/kiro/kiro_test.go index 6e5cdc1..42bdc29 100644 --- a/internal/adapters/kiro/kiro_test.go +++ b/internal/adapters/kiro/kiro_test.go @@ -955,3 +955,40 @@ func TestImagesTooLargeToLandAreLeftOut(t *testing.T) { t.Fatal("no step says too_large") } } + +// TestARetiredSessionIsNotCollected. A Kiro session is retired and its +// directory is gone, while its files, its sub-executions included, are still +// in Kiro's. A pass lands nothing of it and does not make its directory +// again. The other sessions are collected as before. +func TestARetiredSessionIsNotCollected(t *testing.T) { + r := newRoot(t, fixtures) + st := r.collect() + id := "" + for _, s := range st.Changed { + if strings.HasPrefix(s, subagents) { + id = s + } + } + if id == "" { + t.Fatalf("the subagents fixture did not land: %v", st.Changed) + } + if err := storage.DeleteTree(r.zone.SessionDir(id), nil); err != nil { + t.Fatal(err) + } + if err := os.MkdirAll(filepath.Join(r.zone.Root(), storage.RetiredDir), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(r.zone.RetiredPath(id), nil, 0o444); err != nil { + t.Fatal(err) + } + again := r.collect() + if len(again.Retired) != 1 || again.Retired[0] != id { + t.Fatalf("retired %v, want %s", again.Retired, id) + } + if _, err := os.Stat(r.zone.SessionDir(id)); !os.IsNotExist(err) { + t.Fatalf("the retired session's directory was made again: %v", err) + } + if again.Sessions != st.Sessions-1 { + t.Errorf("%d sessions collected, want the other %d", again.Sessions, st.Sessions-1) + } +} diff --git a/internal/adapters/langsmith/collect.go b/internal/adapters/langsmith/collect.go index 0b553f7..c515aa5 100644 --- a/internal/adapters/langsmith/collect.go +++ b/internal/adapters/langsmith/collect.go @@ -97,6 +97,9 @@ type Landed struct { // run whose conversation would come from a run above it that is not known. // Two updates of one run count twice. Dropped int + // Retired names the retired conversations runs arrived for. None of + // them lands. + Retired []string } // Collect converts every waiting request, oldest first. @@ -229,6 +232,11 @@ func (c *Collector) Collect() (Landed, error) { out.Records += landed.Records out.Unassigned += landed.Unassigned out.Dropped += landed.Dropped + for _, s := range landed.Retired { + if !slices.Contains(out.Retired, s) { + out.Retired = append(out.Retired, s) + } + } out.Bodies += landed.Bodies out.BodyConflicts += landed.BodyConflicts for _, s := range landed.Sessions { @@ -444,9 +452,29 @@ func (c *Collector) convertWith(request Waiting, open *pending, wait bool) (Land sessions[i], undecided[i] = sessions[j], undecided[j] } } + // A run whose conversation is decided and retired is not collected. It + // is decided before the wait, so it never holds back a run that is: only + // a decided conversation is asked about, never one a run fell back to + // while it waits. + retired := make([]bool, len(operations)) + asked := map[string]bool{} + for i := range operations { + if undecided[i] { + continue + } + known, ok := asked[sessions[i]] + if !ok { + var err error + if known, err = storage.Retired(c.Zone, sessions[i]); err != nil { + return Landed{}, err + } + asked[sessions[i]] = known + } + retired[i] = known + } if wait { for i := range operations { - if undecided[i] || !open.placeable(runs[i]) { + if !retired[i] && (undecided[i] || !open.placeable(runs[i])) { return Landed{}, &unplaceable{run: runs[i].ID} } } @@ -482,6 +510,15 @@ func (c *Collector) convertWith(request Waiting, open *pending, wait bool) (Land out.Dropped++ continue } + if retired[i] { + // Remembered in its conversation, so a later arrival of it, or + // of a run inside it, is known to be retired too. + open.started(envelope, methodOf(op), session, c.Now()) + if !slices.Contains(out.Retired, session) { + out.Retired = append(out.Retired, session) + } + continue + } // A run may be updated after it ends as well as before, and that // update carries none of this. So a run is remembered whether or not // it has finished, and is forgotten only by age. Forgetting it here diff --git a/internal/adapters/langsmith/retired_test.go b/internal/adapters/langsmith/retired_test.go new file mode 100644 index 0000000..edb3923 --- /dev/null +++ b/internal/adapters/langsmith/retired_test.go @@ -0,0 +1,266 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package langsmith + +import ( + "os" + "path/filepath" + "runtime" + "strings" + "testing" + "time" + + "github.com/apache/skywalking-ai-sessionizer/internal/storage" +) + +// retireSession marks a conversation retired and takes its directory away, +// as a finished retirement leaves it. +func retireSession(t *testing.T, z *storage.Zone, session string) { + t.Helper() + if err := os.MkdirAll(filepath.Join(z.Root(), storage.RetiredDir), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(z.RetiredPath(session), nil, 0o444); err != nil { + t.Fatal(err) + } + if err := storage.DeleteTree(z.SessionDir(session), nil); err != nil { + t.Fatal(err) + } +} + +// TestARetiredConversationsRunsAreNotCollected. A conversation is retired +// after its function and its call landed. Later arrivals of it are not +// collected, and none waits: a run that names it, inside a run that never +// came; a call inside its function that names no thread; an update of its +// call. A run of another conversation in the same request lands in the same +// pass. +func TestARetiredConversationsRunsAreNotCollected(t *testing.T) { + _, zone, base := started(t, "") + c := collectorOver(zone) + start := strings.Replace(inheritCall(``), `"end_time":"2026-09-20T10:00:02Z",`, ``, 1) + postBatch(t, base, `{"post":[`+inheritFunction("t-retired")+`,`+start+`]}`) + passes(t, c, 1) + retired := threadOf("t-retired") + if !mentions(t, zone, retired, `"FakeListChatModel"`) { + t.Fatal("the call did not land in its function's conversation") + } + retireSession(t, zone, retired) + + const call = "16161616-1616-7616-8616-161616161612" + callOrder := inheritOrder + ".20260920T100002000000Z" + call + another := strings.ReplaceAll(strings.ReplaceAll(inheritCall(``), inheritChild, call), + inheritOrder+".20260920T100001000000Z"+call, callOrder) + // It ran inside a run that never came, so as a live run it would wait. + const missing = "16161616-1616-7616-8616-161616161613" + named := `{"id":"` + lostStep + `","trace_id":"` + missing + `","parent_run_id":"` + missing + `",` + + `"dotted_order":"20260920T100000000000Z` + missing + `.20260920T100001000000Z` + lostStep + `",` + + `"run_type":"chain","name":"named","start_time":"2026-09-20T10:00:01Z","end_time":"2026-09-20T10:00:02Z",` + + `"session_name":"asz","extra":{"metadata":{"thread_id":"t-retired"},"runtime":{"library":"langsmith"}},` + + `"outputs":{"answer":"named again"}}` + live := rootRun(lostBelow, "asz", `"thread_id":"t-live"`, "a live one") + postBatch(t, base, `{"post":[`+named+`,`+another+`,`+live+`],"patch":[`+ + sparse(inheritChild, inheritRoot, lostCallOrder, "the call's end")+`]}`) + landed, err := c.Collect() + if err != nil { + t.Fatal(err) + } + if landed.Waiting != 0 || landed.Dropped != 0 || len(landed.Retired) != 1 || landed.Retired[0] != retired { + t.Fatalf("%+v: the retired runs must be left out at once, and only them", landed) + } + if !mentions(t, zone, threadOf("t-live"), "a live one") { + t.Error("the run of another conversation did not land") + } + if _, err := os.Stat(zone.SessionDir(retired)); !os.IsNotExist(err) { + t.Fatalf("the retired conversation's directory was made again: %v", err) + } + open := loadPending(filepath.Join(zone.Root(), InboxDir, pendingFile)) + for _, id := range []string{lostStep, call, inheritChild} { + if open.Runs[id].Session != retired { + t.Errorf("run %s is remembered in %q, want the retired conversation", id, open.Runs[id].Session) + } + } +} + +// TestAnUpdateUnderARetiredRunWaitsForItsStart. A call's update, naming no +// conversation, comes before its start, inside a function whose +// conversation is retired. It is not taken for the function's while it +// waits: its start names a thread of its own, and both land there. +func TestAnUpdateUnderARetiredRunWaitsForItsStart(t *testing.T) { + _, zone, base := started(t, "") + c := collectorOver(zone) + postBatch(t, base, `{"post":[`+inheritFunction("t-retired")+`]}`) + passes(t, c, 1) + retireSession(t, zone, threadOf("t-retired")) + postBatch(t, base, `{"patch":[`+sparse(inheritChild, inheritRoot, lostCallOrder, "its own end")+`]}`) + if landed, err := c.Collect(); err != nil || landed.Waiting != 1 || len(landed.Retired) != 0 { + t.Fatalf("%v, %+v: the update must wait for its start, not be retired", err, landed) + } + start := strings.Replace(inheritCall(`"thread_id":"t-own"`), `"end_time":"2026-09-20T10:00:02Z",`, ``, 1) + postBatch(t, base, `{"post":[`+start+`]}`) + passes(t, c, 1) + if !mentions(t, zone, threadOf("t-own"), "its own end") || !mentions(t, zone, threadOf("t-own"), `"FakeListChatModel"`) { + t.Error("the call and its update are not in the call's own conversation") + } +} + +// TestARetiredUnassignedConversationStaysRetired. A trace with no thread +// anywhere landed in its own conversation, unassigned, which is then +// retired. A later update of its run that carries a thread stays in it, and +// is not collected. +func TestARetiredUnassignedConversationStaysRetired(t *testing.T) { + _, zone, base := started(t, "") + c := collectorOver(zone) + postBatch(t, base, `{"post":[`+rootRun(lostStep, "asz", ``, "no thread anywhere")+`]}`) + passes(t, c, 1) + retireSession(t, zone, UnassignedID(lostStep)) + postBatch(t, base, `{"patch":[`+rootRun(lostStep, "asz", `"thread_id":"t-later"`, "a thread too late")+`]}`) + landed, err := c.Collect() + if err != nil || len(landed.Retired) != 1 || landed.Retired[0] != UnassignedID(lostStep) { + t.Fatalf("%v, %+v: the update must stay in its retired conversation", err, landed) + } + if landedAnywhere(t, zone, "a thread too late") { + t.Error("the update of a retired conversation's run was collected") + } +} + +// TestARetiredConversationAcrossRestartsAndWaits. A conversation is retired +// after its function and its call landed. +// - After a restart, an update of its call is still retired: what is +// remembered of the call names it. +// - A request that holds an update of the retired call and an update whose +// start never came waits as long as it may for the second. Then the first +// is left out as retired and the second dropped, and nothing lands. +// - Once the function is forgotten, a call inside it that names no thread +// cannot be told and is dropped: it lands neither in the retired +// conversation nor anywhere else. +func TestARetiredConversationAcrossRestartsAndWaits(t *testing.T) { + _, zone, base := started(t, "") + c, clock := clockedOver(zone) + start := strings.Replace(inheritCall(``), `"end_time":"2026-09-20T10:00:02Z",`, ``, 1) + postBatch(t, base, `{"post":[`+inheritFunction("t-retired")+`,`+start+`]}`) + passes(t, c, 1) + retired := threadOf("t-retired") + retireSession(t, zone, retired) + + restarted := &Collector{Zone: zone, Now: clock.now} + postBatch(t, base, `{"patch":[`+sparse(inheritChild, inheritRoot, lostCallOrder, "after a restart")+`]}`) + if landed, err := restarted.Collect(); err != nil || len(landed.Retired) != 1 || landed.Retired[0] != retired { + t.Fatalf("%v, %+v: the call's update is retired after a restart", err, landed) + } + + order := "20260920T100000000000Z" + lostStep + postBatch(t, base, `{"patch":[`+sparse(inheritChild, inheritRoot, lostCallOrder, "beside one that waits")+`,`+ + sparse(lostStep, lostStep, order, "never started")+`]}`) + dropped, retiredSeen := 0, false + for pass := 0; pass <= deferLimit; pass++ { + landed, err := restarted.Collect() + if err != nil { + t.Fatal(err) + } + dropped += landed.Dropped + retiredSeen = retiredSeen || len(landed.Retired) == 1 + } + if dropped != 1 || !retiredSeen || !inboxEmpty(t, zone) { + t.Fatalf("%d dropped, retired seen %v: once the wait ends, the retired update is left out and the other dropped", dropped, retiredSeen) + } + + clock.at = clock.at.Add(25 * time.Hour) + postBatch(t, base, `{"post":[`+otherRun("t-other")+`]}`) + const call = "16161616-1616-7616-8616-161616161614" + late := strings.ReplaceAll(inheritCall(``), inheritChild, call) + postBatch(t, base, `{"post":[`+late+`]}`) + if dropped := passes(t, restarted, deferLimit+1); dropped != 1 { + t.Fatalf("%d dropped: a call under a forgotten retired function cannot be told", dropped) + } + for _, text := range []string{"after a restart", "beside one that waits", "never started", call} { + if landedAnywhere(t, zone, text) { + t.Errorf("%q was collected", text) + } + } +} + +// TestARetiredCallOutlivesItsFunction. A conversation is retired after its +// function and its call landed. The call is updated within the day, so it +// is still remembered when its function is forgotten. Its next update is +// still retired, by what is remembered of the call itself: it does not wait +// for the function, is not dropped, and lands nowhere. +func TestARetiredCallOutlivesItsFunction(t *testing.T) { + _, zone, base := started(t, "") + c, clock := clockedOver(zone) + start := strings.Replace(inheritCall(``), `"end_time":"2026-09-20T10:00:02Z",`, ``, 1) + postBatch(t, base, `{"post":[`+inheritFunction("t-retired")+`,`+start+`]}`) + passes(t, c, 1) + retired := threadOf("t-retired") + retireSession(t, zone, retired) + + clock.at = clock.at.Add(12 * time.Hour) + postBatch(t, base, `{"patch":[`+sparse(inheritChild, inheritRoot, lostCallOrder, "within the day")+`]}`) + if landed, err := c.Collect(); err != nil || len(landed.Retired) != 1 || landed.Retired[0] != retired { + t.Fatalf("%v, %+v: the call's update is retired", err, landed) + } + clock.at = clock.at.Add(13 * time.Hour) + postBatch(t, base, `{"post":[`+otherRun("t-other")+`]}`) + passes(t, c, 1) + open := loadPending(filepath.Join(zone.Root(), InboxDir, pendingFile)) + if _, ok := open.Runs[inheritRoot]; ok { + t.Fatal("the function is still remembered; the test needs it forgotten") + } + if open.Runs[inheritChild].Session != retired { + t.Fatalf("the call is remembered in %q, want the retired conversation", open.Runs[inheritChild].Session) + } + + postBatch(t, base, `{"patch":[`+sparse(inheritChild, inheritRoot, lostCallOrder, "after its function")+`]}`) + landed, err := c.Collect() + if err != nil || landed.Waiting != 0 || landed.Dropped != 0 || len(landed.Retired) != 1 || landed.Retired[0] != retired { + t.Fatalf("%v, %+v: the call's update is still retired, at once", err, landed) + } + for _, text := range []string{"within the day", "after its function"} { + if landedAnywhere(t, zone, text) { + t.Errorf("%q was collected", text) + } + } +} + +// TestAMarkerThatCannotBeReadKeepsTheRequest. While the markers cannot be +// read, no run's conversation can be told retired or not, so the request is +// kept, not landed and not dropped. Once they read, it lands. +func TestAMarkerThatCannotBeReadKeepsTheRequest(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("a file where a directory should be reads as not there on Windows") + } + _, zone, base := started(t, "") + blocked := filepath.Join(zone.Root(), storage.RetiredDir) + if err := os.WriteFile(blocked, nil, 0o644); err != nil { + t.Fatal(err) + } + postBatch(t, base, `{"post":[`+rootRun(lostStep, "asz", `"thread_id":"t-live"`, "kept for later")+`]}`) + c := collectorOver(zone) + if _, err := c.Collect(); err == nil { + t.Fatal("a pass that cannot read the markers succeeded") + } + if inboxEmpty(t, zone) || landedAnywhere(t, zone, "kept for later") { + t.Fatal("the request was not kept for a later pass") + } + if err := os.Remove(blocked); err != nil { + t.Fatal(err) + } + passes(t, c, 1) + if !mentions(t, zone, threadOf("t-live"), "kept for later") { + t.Error("the request did not land once the markers read") + } +} diff --git a/internal/export/otlp/push.go b/internal/export/otlp/push.go index e0de630..0657cfa 100644 --- a/internal/export/otlp/push.go +++ b/internal/export/otlp/push.go @@ -327,13 +327,21 @@ func (p *Pusher) Pass() (*Stats, error) { if b.stop != nil { break } + retired, err := b.isRetired(s.id) + if err != nil { + st.Errors = append(st.Errors, fmt.Errorf("%s: %w", s.id, err)) + sent[s.id] = true + continue + } // A session's records are attributed to the runtime that produced // them, which its landed headers name. - b.services[s.id] = p.serviceOf(s.id, s.files) - // A held session's files wait. Its conversation's rounds are asked - // one by one, by the session each header names, which need not be - // this one. - held := !b.claim(s.id) + if !retired { + b.services[s.id] = p.serviceOf(s.id, s.files) + } + // A held session's files wait, and a retired one's never go. Its + // conversation's rounds are asked one by one, by the session each + // header names, which need not be this one. + held := retired || !b.claim(s.id) for _, lf := range s.files { if b.stop != nil || held { break @@ -556,8 +564,9 @@ func (b *batch) addRounds(conv string) { b.st.Deferred++ continue } - // A held session is listed once, in Held, not once per round. - if errors.Is(err, errSessionHeld) { + // A held session is listed once, in Held, not once per round, + // and a retired one is not sent at all. + if errors.Is(err, errSessionHeld) || errors.Is(err, errSessionRetired) { continue } b.st.Errors = append(b.st.Errors, fmt.Errorf("%s: %w", rel, err)) @@ -630,6 +639,29 @@ type batch struct { // latest is the latest record time of each session, read once per pass // when a file without timed records needs a timestamp. latest map[string]uint64 + + // retired is what each session's retired marker said, asked once a + // pass. + retired map[string]bool +} + +// isRetired reports whether a session is retired. None of its landed files +// and none of its rounds goes: a receiver that still holds its first chain +// must not be sent a second. Its metrics requests still go, to its owner, as +// any other session's do. +func (b *batch) isRetired(session string) (bool, error) { + if retired, asked := b.retired[session]; asked { + return retired, nil + } + retired, err := storage.Retired(b.p.Zone, session) + if err != nil { + return false, err + } + if b.retired == nil { + b.retired = map[string]bool{} + } + b.retired[session] = retired + return retired, nil } type pendingFile struct{ rel, digest, session string } @@ -918,6 +950,11 @@ func (b *batch) addRound(rel, path, conv string) error { } // A round goes under the session its header names, which need not be // the conversation's directory, so that session is the one asked. + if retired, err := b.isRetired(session); err != nil { + return err + } else if retired { + return errSessionRetired + } if !b.claim(session) { return errSessionHeld } @@ -1176,6 +1213,9 @@ var errRoundUnfinished = errors.New("round is still being written") // errSessionHeld is a round of a session this pass holds. var errSessionHeld = errors.New("the round's session is held") +// errSessionRetired is a round of a retired session, which is never sent. +var errSessionRetired = errors.New("the round's session is retired") + // roundIsFinished reports whether a round file holds every byte it will. // // The test is its last frame: a finished round ends with a commit carrying diff --git a/internal/metrics/completion_test.go b/internal/metrics/completion_test.go index ceb3b3c..b689f3d 100644 --- a/internal/metrics/completion_test.go +++ b/internal/metrics/completion_test.go @@ -20,6 +20,7 @@ package metrics_test import ( "os" "path/filepath" + "runtime" "sort" "testing" "time" @@ -400,3 +401,36 @@ func TestFailedSpoolWriteDoesNotRecordLaterReceipts(t *testing.T) { t.Fatalf("spool recovery counted one completed call as %v tokens, want 60", n) } } + +// A marker that cannot be read on the first pass leaves the session for the +// next, and keeps the look-back it would have had: the retry derives the +// recent minute only, not the history the first pass was asked to leave out. +func TestAMarkerThatCannotBeReadKeepsTheLookback(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("a file where a directory should be reads as not there on Windows") + } + z := storage.NewZone(t.TempDir()) + land(t, z, "s1", "main", 1, []call{ + {id: "old", model: "m", at: base.Add(-48 * time.Hour), frags: 1, in: 100, out: 100}, + {id: "recent", model: "m", at: base, frags: 1, in: 4, out: 60}, + }) + blocked := filepath.Join(z.Root(), storage.RetiredDir) + if err := os.WriteFile(blocked, nil, 0o644); err != nil { + t.Fatal(err) + } + d := deriver(z, base.Add(time.Hour), metrics.Options{Lookback: 24 * time.Hour, Grace: -1}) + st, err := d.Pass(nil) + if err != nil || len(st.Errors) != 1 || len(st.PendingSessions) != 1 || st.Files != 0 { + t.Fatalf("the first pass must leave the session it cannot tell: %v %+v", err, st) + } + if err := os.Remove(blocked); err != nil { + t.Fatal(err) + } + if st, err = d.Pass(st.PendingSessions); err != nil || len(st.Errors) > 0 || st.Files != 1 { + t.Fatalf("the retry must derive the file: %v %+v", err, st) + } + got, _ := points(t, z) + if n := total(got["main/output/m#s1"]); n != 60 { + t.Fatalf("the retry bypassed the first pass's look-back: got %v output tokens, want 60", n) + } +} diff --git a/internal/metrics/mcp_test.go b/internal/metrics/mcp_test.go index ae0ef13..0db3a10 100644 --- a/internal/metrics/mcp_test.go +++ b/internal/metrics/mcp_test.go @@ -25,6 +25,7 @@ import ( "runtime" "slices" "strconv" + "strings" "testing" "time" @@ -409,3 +410,36 @@ func TestTheCountReadsWhatItShould(t *testing.T) { t.Fatalf("unread %v, want 1: the observation landed twice, and nothing else", st.Unread) } } + +// TestNothingOfARetiredSessionIsDerived. A session is retired while a file of +// it is still there, not yet derived. A pass over every session, and one +// that names it, derive nothing of it and keep no state of it. +func TestNothingOfARetiredSessionIsDerived(t *testing.T) { + z := storage.NewZone(t.TempDir()) + landExecutions(t, z, 1, observed("toolu_a", "mcp__status__lookup", "status", 100, execution.OutcomeReturned)) + if err := os.MkdirAll(filepath.Join(z.Root(), storage.RetiredDir), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(z.RetiredPath("s1"), nil, 0o444); err != nil { + t.Fatal(err) + } + for _, scope := range [][]string{nil, {"s1"}} { + st, err := unreadDeriver(z).Pass(scope) + if err != nil || len(st.Errors) > 0 { + t.Fatalf("%v %v", err, st) + } + if st.Files != 0 || st.Requests != 0 { + t.Fatalf("scope %v: derived %d file(s) and %d request(s) of a retired session", scope, st.Files, st.Requests) + } + } + if _, err := os.Stat(filepath.Join(z.SessionDir("s1"), "metrics")); !os.IsNotExist(err) { + t.Fatalf("a receipt of a retired session was written: %v", err) + } + state, err := os.ReadFile(filepath.Join(z.Root(), storage.SpoolDir, metrics.StateFile)) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(state), "s1") { + t.Fatalf("the state keeps the retired session: %s", state) + } +} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 5435ad5..7e6d770 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -235,6 +235,12 @@ func (d *Deriver) Pass(sessions []string) (*Stats, error) { } var scanned []string for _, session := range sessions { + // Nothing of a retired session is derived, nor any state of it kept, + // a session named by the caller included. + retired, rerr := storage.Retired(d.Zone, session) + if rerr == nil && retired { + continue + } // A session the first pass did not finish keeps that pass's look-back // until it is finished. It is kept per session, not per file: a file // the pass never reached, after an error, must not export the old @@ -243,6 +249,13 @@ func (d *Deriver) Pass(sessions []string) (*Stats, error) { if !since.IsZero() { state.PendingSince[session] = since } + // A marker that cannot be read leaves the session for another pass, + // with the look-back it would have had kept for it. + if rerr != nil { + st.Errors = append(st.Errors, fmt.Errorf("%s: %w", session, rerr)) + st.PendingSessions = append(st.PendingSessions, session) + continue + } sessionSince := state.PendingSince[session] files, err := storage.LandedFiles(d.Zone, session) if err != nil { diff --git a/internal/parse/parse.go b/internal/parse/parse.go index 276f483..6f3ac61 100644 --- a/internal/parse/parse.go +++ b/internal/parse/parse.go @@ -159,6 +159,15 @@ func Session(z *storage.Zone, opt Options) (*Round, error) { chain := sessionflow.OpenChain(z.Root(), opt.Conversation) nothing := &Round{Session: opt.Session, Conversation: opt.Conversation} + // A retired session is never written again: no index is built or saved + // for it, and no chain made, while what is left of it waits for its + // retirement to finish. + if retired, err := storage.Retired(z, opt.Session); err != nil { + return nil, err + } else if retired { + return nothing, nil + } + // A session with no landed files and no chain has nothing to parse. Taking // the lock would create the chain's directory, so the parse returns first // and creates nothing. Otherwise a parse that ran after a scenario root diff --git a/internal/repack/repack.go b/internal/repack/repack.go index 93307e1..00fbe9c 100644 --- a/internal/repack/repack.go +++ b/internal/repack/repack.go @@ -59,7 +59,14 @@ type slot struct { // any session is written. A skeleton keeps no records to re-cut, and a chain // built again in the new root would have nothing to bind them to. A removal // in progress is refused too: the next pipeline pass finishes it. +// +// A root that holds a retired session is refused whole, whichever sessions +// are named: a new root without its marker would collect it again under the +// id a receiver may still hold. func CheckSource(src *storage.Zone, sessions []string) error { + if err := checkRetired(src); err != nil { + return err + } for _, session := range sessions { files, err := storage.LandedFiles(src, session) if err != nil { @@ -72,6 +79,19 @@ func CheckSource(src *storage.Zone, sessions []string) error { return nil } +// checkRetired refuses a root that holds a retired session. +func checkRetired(src *storage.Zone) error { + retired, err := storage.RetiredSessions(src) + if err != nil { + return err + } + if len(retired) > 0 { + return fmt.Errorf("repack: the root holds %d retired session(s) under %s; a new root without their markers would collect them again, so this root cannot be repacked", + len(retired), storage.RetiredDir) + } + return nil +} + func checkRemoved(src *storage.Zone, session string, files []storage.LandedFile) error { if r, err := storage.LoadRemoving(src.RemovingPath(session)); err != nil { return err @@ -128,6 +148,17 @@ func Session(src, dst *storage.Zone, session string, budget int64, now time.Time if err := checkRemoved(src, session, files); err != nil { return nil, err } + // Also when Session is called without the command's whole-root check, + // as the scenario runner calls it. A destination that retired the + // session never writes it again either. + if err := checkRetired(src); err != nil { + return nil, err + } + if retired, err := storage.Retired(dst, session); err != nil { + return nil, err + } else if retired { + return nil, fmt.Errorf("repack: session %s is retired in the destination, and is never written again", session) + } srcDir, dstDir := src.SessionDir(session), dst.SessionDir(session) // Reserve a new session directory even when Session is called without // the command's whole-root check. A retry must never rewrite its files. diff --git a/internal/scenario/marker_test.go b/internal/scenario/marker_test.go index fe01165..9798bd9 100644 --- a/internal/scenario/marker_test.go +++ b/internal/scenario/marker_test.go @@ -27,6 +27,8 @@ import ( "strings" "testing" "time" + + "github.com/apache/skywalking-ai-sessionizer/internal/storage" ) // everyKindOfFile is one session holding every kind of file a removal @@ -270,6 +272,38 @@ func TestAnSDBuildWritesNoMarker(t *testing.T) { } } +// TestAnSDBuildRefusesARetiredSession. An sd root has no _scenario/, so +// retention runs over it as over a real root. A build of a session retired +// there would start it again at sequence 1, under an id a receiver may still +// hold, so it is refused, and nothing of the session is made. +func TestAnSDBuildRefusesARetiredSession(t *testing.T) { + sc, err := Load(filepath.Join("..", "..", "tests", "scenarios", "assembly.yaml")) + if err != nil { + t.Fatal(err) + } + out := t.TempDir() + b, err := Build(sc, FormatSD, out, Options{At: markerAt}) + if err != nil { + t.Fatal(err) + } + zone := storage.NewZone(b.Out) + if err := storage.DeleteTree(zone.SessionDir(b.Session), nil); err != nil { + t.Fatal(err) + } + if err := os.MkdirAll(filepath.Join(b.Out, storage.RetiredDir), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(zone.RetiredPath(b.Session), nil, 0o444); err != nil { + t.Fatal(err) + } + if _, err := Build(sc, FormatSD, out, Options{At: markerAt}); err == nil || !strings.Contains(err.Error(), "retired") { + t.Fatalf("a build of a retired session returned %v", err) + } + if _, err := os.Stat(zone.SessionDir(b.Session)); !os.IsNotExist(err) { + t.Fatalf("the refused build made the session's directory: %v", err) + } +} + // TestParseRemove holds the syntax of --remove. A retention of zero or less // is refused rather than read as immediately, since a person who wrote a // duration meant one. diff --git a/internal/scenario/run/retired_test.go b/internal/scenario/run/retired_test.go new file mode 100644 index 0000000..60db0db --- /dev/null +++ b/internal/scenario/run/retired_test.go @@ -0,0 +1,82 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package run + +import ( + "os" + "path/filepath" + "testing" + + "github.com/apache/skywalking-ai-sessionizer/internal/storage" +) + +// A copy of a root that holds a retired session is refused, as repack +// refuses it, and makes nothing: it would carry what is left of the session +// into a root without its marker, beside a live session's files. +func TestACopyOfARootWithARetiredSessionIsRefused(t *testing.T) { + from, to := t.TempDir(), filepath.Join(t.TempDir(), "twin") + z := storage.NewZone(from) + for _, session := range []string{"retired", "live"} { + dir := z.StreamDir(session, storage.StreamMain) + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, "transcript-x-000001.sd"), []byte("{}\n"), 0o444); err != nil { + t.Fatal(err) + } + } + if err := os.MkdirAll(filepath.Join(from, storage.RetiredDir), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(z.RetiredPath("retired"), nil, 0o444); err != nil { + t.Fatal(err) + } + if err := copyLanded(from, to); err == nil { + t.Fatal("a root holding a retired session was copied") + } + if _, err := os.Stat(to); !os.IsNotExist(err) { + t.Fatalf("the refused copy made its destination: %v", err) + } +} + +// A copy goes to a new directory only. A destination already there may hold +// the marker of the very session copied, which the copy would collect again; +// one that is there is refused, and nothing of the copy is made in it. +func TestACopyIntoARootThatIsThereIsRefused(t *testing.T) { + from, to := t.TempDir(), t.TempDir() + dir := storage.NewZone(from).StreamDir("s1", storage.StreamMain) + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, "transcript-x-000001.sd"), []byte("{}\n"), 0o444); err != nil { + t.Fatal(err) + } + dst := storage.NewZone(to) + if err := os.MkdirAll(filepath.Join(to, storage.RetiredDir), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(dst.RetiredPath("s1"), nil, 0o444); err != nil { + t.Fatal(err) + } + if err := copyLanded(from, to); err == nil { + t.Fatal("a copy went into a root that is already there") + } + if _, err := os.Stat(dst.SessionDir("s1")); !os.IsNotExist(err) { + t.Fatalf("the refused copy made the retired session again: %v", err) + } +} diff --git a/internal/scenario/run/run.go b/internal/scenario/run/run.go index 1eda260..424aab4 100644 --- a/internal/scenario/run/run.go +++ b/internal/scenario/run/run.go @@ -296,9 +296,12 @@ func checkFormat(sc *scenario.Scenario, ex *expect.File, f scenario.Format, out // Two parses of identical landed evidence must produce identical // rounds. The chain in out was cut at checkpoints, so it is not // the comparison; two fresh copies of the landed files are. - a, b := out+"-twin-a", out+"-twin-b" - defer os.RemoveAll(a) - defer os.RemoveAll(b) + twins, err := os.MkdirTemp(filepath.Dir(out), filepath.Base(out)+"-twins-") + if err != nil { + return nil, err + } + defer os.RemoveAll(twins) + a, b := filepath.Join(twins, "a"), filepath.Join(twins, "b") for _, twin := range []string{a, b} { if err := copyLanded(out, twin); err != nil { return nil, err @@ -559,7 +562,20 @@ func parseAll(root, session string, maxRound int64) (*parse.Round, error) { // copyLanded copies a session's landed files, and nothing else: no index, no // state, no chain. What a bundle would carry. +// +// A root that holds a retired session is refused, as repack refuses it: the +// copy would carry what is left of it into a root without its marker. The +// destination must not be there yet, so the copy never adds a session to a +// root whose marker says it is retired. func copyLanded(from, to string) error { + if retired, err := storage.RetiredSessions(storage.NewZone(from)); err != nil { + return err + } else if len(retired) > 0 { + return fmt.Errorf("%s holds %d retired session(s), and a copy of it would collect them again", from, len(retired)) + } + if err := os.Mkdir(to, 0o755); err != nil { + return fmt.Errorf("a copy goes to a new directory: %w", err) + } return filepath.WalkDir(from, func(path string, d os.DirEntry, err error) error { if err != nil { return err diff --git a/internal/scenario/write_sd.go b/internal/scenario/write_sd.go index 01ee3bc..103079f 100644 --- a/internal/scenario/write_sd.go +++ b/internal/scenario/write_sd.go @@ -46,6 +46,14 @@ import ( // parser reads it like any other root. func writeSD(p *Plan, root string, now time.Time, maxDelta int64) ([]string, error) { z := storage.NewZone(root) + // An sd root has no _scenario/, so retention runs over it as over a + // real one. A build of a retired id would start it again at sequence 1, + // a second chain under an id the receiver may still hold. + if retired, err := storage.Retired(z, p.Session); err != nil { + return nil, err + } else if retired { + return nil, fmt.Errorf("scenario: session %s is retired in %s, and is never written again", p.Session, root) + } sessionDir := z.SessionDir(p.Session) if err := mkdirAll(sessionDir); err != nil { return nil, err diff --git a/internal/storage/retired.go b/internal/storage/retired.go new file mode 100644 index 0000000..458a4f7 --- /dev/null +++ b/internal/storage/retired.go @@ -0,0 +1,79 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package storage + +import ( + "errors" + "io/fs" + "os" + "path/filepath" + "strings" +) + +// RetiredDir holds one empty file for each retired session, named by its id. +// +// A retired session is never written again: nothing of it is landed, +// indexed, parsed, derived or sent, and no directory of it is made. Its +// receiver may still hold its first chain, for longer than this root does, +// and a second chain under the same id would show the receiver two +// conversations as one. The marker is what keeps it out, so it is never +// removed, and it holds nothing but its name: nothing reads more. +const RetiredDir = "_retired" + +// RetiredPath is where a session's retired marker is. +func (z *Zone) RetiredPath(session string) string { + return filepath.Join(z.root, RetiredDir, session) +} + +// Retired reports whether a session is retired. Every writer that can make a +// session's directory, its chain or any state of it asks before it does, +// since taking a session's lock makes its directory. An error other than the +// marker not being there is returned, and the caller writes nothing of the +// session that pass: a marker it cannot see may still be there. +func Retired(z *Zone, session string) (bool, error) { + _, err := os.Lstat(z.RetiredPath(session)) + switch { + case err == nil: + return true, nil + case errors.Is(err, fs.ErrNotExist): + return false, nil + } + return false, err +} + +// RetiredSessions lists the retired sessions of a root, by their markers. A +// name that cannot be a session's, such as a marker's temporary file, is not +// one. +func RetiredSessions(z *Zone) ([]string, error) { + items, err := os.ReadDir(filepath.Join(z.root, RetiredDir)) + if err != nil { + if errors.Is(err, fs.ErrNotExist) { + return nil, nil + } + return nil, err + } + var out []string + for _, it := range items { + name := it.Name() + if it.IsDir() || strings.HasPrefix(name, ".") { + continue + } + out = append(out, name) + } + return out, nil +} diff --git a/internal/storage/retired_test.go b/internal/storage/retired_test.go new file mode 100644 index 0000000..157f731 --- /dev/null +++ b/internal/storage/retired_test.go @@ -0,0 +1,72 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package storage_test + +import ( + "os" + "path/filepath" + "runtime" + "strings" + "testing" + + "github.com/apache/skywalking-ai-sessionizer/internal/storage" +) + +// A session is retired when its marker is there. A root with no _retired +// retires nothing, a temporary file left by an interrupted write is no +// marker, and a marker that cannot be looked at is an error rather than +// taken for none. +func TestARetiredSessionIsKnownByItsMarker(t *testing.T) { + root := t.TempDir() + z := storage.NewZone(root) + if retired, err := storage.Retired(z, "s1"); err != nil || retired { + t.Fatalf("a root with no %s says %v, %v", storage.RetiredDir, retired, err) + } + if ids, err := storage.RetiredSessions(z); err != nil || len(ids) != 0 { + t.Fatalf("a root with no %s lists %v, %v", storage.RetiredDir, ids, err) + } + dir := filepath.Join(root, storage.RetiredDir) + if err := os.MkdirAll(filepath.Join(dir, "a-directory"), 0o755); err != nil { + t.Fatal(err) + } + for _, name := range []string{"s1", ".tmp-123"} { + if err := os.WriteFile(filepath.Join(dir, name), nil, 0o444); err != nil { + t.Fatal(err) + } + } + if retired, err := storage.Retired(z, "s1"); err != nil || !retired { + t.Fatalf("s1 with its marker says %v, %v", retired, err) + } + if retired, err := storage.Retired(z, "s2"); err != nil || retired { + t.Fatalf("s2 with no marker says %v, %v", retired, err) + } + if ids, err := storage.RetiredSessions(z); err != nil || strings.Join(ids, ",") != "s1" { + t.Fatalf("lists %v, %v; want s1 alone", ids, err) + } + + if runtime.GOOS == "windows" { + return // A file where a directory should be reads as not there. + } + other := storage.NewZone(t.TempDir()) + if err := os.WriteFile(filepath.Join(other.Root(), storage.RetiredDir), nil, 0o644); err != nil { + t.Fatal(err) + } + if _, err := storage.Retired(other, "s1"); err == nil { + t.Fatal("a marker that cannot be looked at was taken for none") + } +} diff --git a/tests/chain/repack_test.go b/tests/chain/repack_test.go index a271b17..ed80605 100644 --- a/tests/chain/repack_test.go +++ b/tests/chain/repack_test.go @@ -21,6 +21,7 @@ import ( "os" "path/filepath" "reflect" + "runtime" "testing" "time" @@ -90,3 +91,43 @@ func TestRepackRefusesAnOrphanDestinationChain(t *testing.T) { t.Fatalf("chain changed: %q, %v", data, err) } } + +// TestRepackRefusesARetiredSession. repack.Session, called directly as the +// scenario runner calls it, refuses a source that holds a retired session, +// whichever it is asked for, and a destination where the session is retired, +// and a source whose markers cannot be read. Each refusal makes nothing. +func TestRepackRefusesARetiredSession(t *testing.T) { + mark := func(z *storage.Zone, session string) { + if err := os.MkdirAll(filepath.Join(z.Root(), storage.RetiredDir), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(z.RetiredPath(session), nil, 0o444); err != nil { + t.Fatal(err) + } + } + cases := map[string]func(src, dst *storage.Zone, session string){ + "another session retired in the source": func(src, _ *storage.Zone, _ string) { mark(src, "another") }, + "the session retired in the destination": func(_, dst *storage.Zone, session string) { mark(dst, session) }, + } + if runtime.GOOS != "windows" { + cases["markers that cannot be read"] = func(src, _ *storage.Zone, _ string) { + if err := os.WriteFile(filepath.Join(src.Root(), storage.RetiredDir), nil, 0o644); err != nil { + t.Fatal(err) + } + } + } + for name, set := range cases { + t.Run(name, func(t *testing.T) { + s := newStage(t, growing) + s.through("one") + dst := storage.NewZone(t.TempDir()) + set(s.zone, dst, s.session) + if _, err := repack.Session(s.zone, dst, s.session, 1024, time.Now()); err == nil { + t.Fatal("the repack was taken") + } + if _, err := os.Stat(dst.SessionDir(s.session)); !os.IsNotExist(err) { + t.Fatalf("a refused repack made the session's directory: %v", err) + } + }) + } +} diff --git a/tests/chain/retired_test.go b/tests/chain/retired_test.go new file mode 100644 index 0000000..4eb8173 --- /dev/null +++ b/tests/chain/retired_test.go @@ -0,0 +1,273 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package chain_test + +import ( + "errors" + "fmt" + "io/fs" + "os" + "path/filepath" + "sort" + "strings" + "testing" + + collmetricspb "go.opentelemetry.io/proto/otlp/collector/metrics/v1" + + "github.com/apache/skywalking-ai-sessionizer/internal/export/otlp/otlptest" + "github.com/apache/skywalking-ai-sessionizer/internal/metrics" + "github.com/apache/skywalking-ai-sessionizer/internal/parse" + "github.com/apache/skywalking-ai-sessionizer/internal/scenario" + "github.com/apache/skywalking-ai-sessionizer/internal/storage" +) + +// treeOf is every file under the directories given, with its mode, size and +// time, so a write of any kind shows. A directory that is not there says so. +func treeOf(t *testing.T, dirs ...string) []string { + t.Helper() + var out []string + for _, dir := range dirs { + err := filepath.WalkDir(dir, func(path string, d fs.DirEntry, err error) error { + if err != nil { + return err + } + info, err := d.Info() + if err != nil { + return err + } + rel, _ := filepath.Rel(dir, path) + out = append(out, fmt.Sprintf("%s/%s %s %d %s", filepath.Base(dir), filepath.ToSlash(rel), + info.Mode(), info.Size(), info.ModTime())) + return nil + }) + if errors.Is(err, fs.ErrNotExist) { + out = append(out, filepath.Base(dir)+" absent") + continue + } + if err != nil { + t.Fatal(err) + } + } + sort.Strings(out) + return out +} + +// retire marks a session retired, as retirement will: an empty file named by +// its id. +func retire(t *testing.T, z *storage.Zone, session string) { + t.Helper() + if err := os.MkdirAll(filepath.Join(z.Root(), storage.RetiredDir), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(z.RetiredPath(session), nil, 0o444); err != nil { + t.Fatal(err) + } +} + +// grow appends a line to every transcript of the session's sources, as a +// runtime that resumed it would. +func grow(t *testing.T, r *sentRoot, session string) { + t.Helper() + grown := 0 + err := filepath.WalkDir(r.source, func(path string, d fs.DirEntry, err error) error { + if err != nil || d.IsDir() || !strings.HasSuffix(path, ".jsonl") || !strings.Contains(path, session) { + return err + } + f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0) + if err != nil { + return err + } + defer f.Close() + _, err = f.WriteString(`{"type":"user","uuid":"retired-resume","sessionId":"` + session + + `","timestamp":"2026-03-02T00:00:00Z","message":{"role":"user","content":"resumed after it was retired"}}` + "\n") + grown++ + return err + }) + if err != nil { + t.Fatal(err) + } + if grown == 0 { + t.Fatal("no transcript of the session was found to grow") + } +} + +// pointsOf counts the metric points of the requests that name the session, +// and those that name any other. +func pointsOf(reqs []*collmetricspb.ExportMetricsServiceRequest, session string) (points, others int) { + for _, req := range reqs { + for _, rm := range req.GetResourceMetrics() { + for _, sm := range rm.GetScopeMetrics() { + for _, m := range sm.GetMetrics() { + for _, p := range m.GetSum().GetDataPoints() { + if otlptest.Attrs(p.GetAttributes())["session.id"] == session { + points++ + } else { + others++ + } + } + } + } + } + } + return points, others +} + +// TestARetiredSessionIsNeverWrittenAgain. A session collected, derived and +// parsed, and never sent, is retired with what is left of it still there, as +// when its retirement has not finished; with its directory alone or its chain +// alone, as when it stopped between its two directories; and with nothing +// left, as when it has finished. What is left in the first has no index, +// which a writer that did not ask would build again, and the provider's table +// is lost in every case, so it is built again from what is landed. Its +// runtime resumes it, so its sources grow. Every writer passes it over: no +// file of it lands, no index or chain of it is made or changed, nothing of it +// is derived, and none of its files or rounds is sent. Its metrics requests +// are, as any spooled request is. Every kind of source is there: a +// transcript, a child agent, a workflow, the recorder's execution records and +// provider bodies. +func TestARetiredSessionIsNeverWrittenAgain(t *testing.T) { + // What is left: everything, its index aside; its directory alone, or its + // chain alone, as when a retirement stopped between its two directories; + // and nothing. + states := []string{"unfinished", "chain gone", "half finished", "finished"} + for _, name := range []string{"all-kinds", "provider-bodies-long"} { + for _, state := range states { + t.Run(name+", "+state, func(t *testing.T) { + sc, err := scenario.Load(filepath.Join("..", "scenarios", name+".yaml")) + if err != nil { + t.Fatal(err) + } + r := newSentRoot(t) + b := r.build(sc, scenario.Options{}) + r.collectAt(landedAt(0)) + r.derive() + r.parse() + sessionDir := r.zone.SessionDir(b.Session) + chainDir := filepath.Join(r.dir, "_conversations", b.Session) + retire(t, r.zone, b.Session) + gone := map[string][]string{ + "unfinished": {r.zone.IndexDir(b.Session)}, + "chain gone": {chainDir}, + "half finished": {sessionDir}, + "finished": {sessionDir, chainDir}, + }[state] + requests := spoolOf(t, r.dir, b.Session) + if len(requests) == 0 { + t.Fatal("the session spooled no metrics request; the test needs some") + } + // The provider's table is built again from what is landed, + // which a retired session no longer has, or has only in part. + if err := os.Remove(r.zone.ProviderStatePath()); err != nil && !errors.Is(err, fs.ErrNotExist) { + t.Fatal(err) + } + for _, dir := range gone { + if err := storage.DeleteTree(dir, nil); err != nil { + t.Fatal(err) + } + } + before := treeOf(t, sessionDir, chainDir) + grow(t, r, b.Session) + r.rcv.Reset() + + r.collectAt(landedAt(1)) + st, err := (&metrics.Deriver{Zone: r.zone, Options: metrics.Options{Grace: -1, Version: "test"}}).Pass([]string{b.Session}) + if err != nil || len(st.Errors) > 0 { + t.Fatalf("derive: %v %v", err, st) + } + res, err := parse.Session(r.zone, parse.Options{Conversation: b.Session, Session: b.Session, MaxRoundBytes: removalBudget}) + if err != nil || res.Changed() { + t.Fatalf("parse: %v, changed %v", err, res != nil && res.Changed()) + } + r.parse() + r.push() + + if after := treeOf(t, sessionDir, chainDir); strings.Join(after, "\n") != strings.Join(before, "\n") { + t.Fatalf("what is left of the retired session changed:\nbefore:\n%s\nafter:\n%s", + strings.Join(before, "\n"), strings.Join(after, "\n")) + } + for _, rec := range otlptest.Records(r.rcv.Requests()) { + if attrs := otlptest.Attrs(rec.Attributes); attrs["asz.session"] == b.Session { + t.Fatalf("%s of the retired session was sent", attrs["asz.file"]) + } + } + // Its metrics requests still go, as any spooled request does: + // the receiver holds each of them, and the push state records + // each as sent. + received := r.rcv.MetricsRequests() + if len(received) != len(requests) { + t.Errorf("the receiver holds %d metrics requests, want the session's %d", len(received), len(requests)) + } + if points, others := pointsOf(received, b.Session); points == 0 || others != 0 { + t.Errorf("the receiver holds %d points of the retired session and %d of others", points, others) + } + sent, err := otlpSent(r) + if err != nil { + t.Fatal(err) + } + for _, name := range requests { + if _, ok := sent.Digest(storage.SpoolDir + "/" + name); !ok { + t.Errorf("the metrics request %s of the retired session was not sent", name) + } + } + }) + } + } +} + +// TestARoundOfALiveSessionUnderARetiredDirectoryIsSent. A conversation's +// directory need not be its session's: here the rounds of live session B are +// under the directory of A, which is retired while its own directory is +// still there. None of A's files is sent, and B's rounds are, each asked by +// the session its header names. +func TestARoundOfALiveSessionUnderARetiredDirectoryIsSent(t *testing.T) { + r := newSentRoot(t) + sa, err := scenario.Load(filepath.Join("..", "scenarios", "all-kinds.yaml")) + if err != nil { + t.Fatal(err) + } + sb, err := scenario.Load(filepath.Join("..", "scenarios", "growth.yaml")) + if err != nil { + t.Fatal(err) + } + a, b := r.build(sa, scenario.Options{}), r.build(sb, scenario.Options{}) + r.collectAt(landedAt(0)) + for { + res, err := parse.Session(r.zone, parse.Options{Conversation: a.Session, Session: b.Session, MaxRoundBytes: removalBudget}) + if err != nil { + t.Fatal(err) + } + if !res.Changed() || !res.More { + break + } + } + retire(t, r.zone, a.Session) + r.push() + rounds := 0 + for _, rec := range otlptest.Records(r.rcv.Requests()) { + attrs := otlptest.Attrs(rec.Attributes) + if attrs["asz.session"] == a.Session { + t.Fatalf("%s of the retired session was sent", attrs["asz.file"]) + } + if strings.HasPrefix(attrs["asz.file"], "_conversations/"+a.Session+"/") && attrs["asz.session"] == b.Session { + rounds++ + } + } + if rounds == 0 { + t.Fatal("no round of the live session under the retired directory was sent") + } +}