Write nothing of a retired session again

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

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

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.

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.

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

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

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))