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") + } +}