Retire a conversation idle for retire_after

CLAUDE.md

@@ -69,7 +69,7 @@ internal/view/conversation-view/ Horizon's conversation renderer, built from the
 internal/export/otlp/           sends landed files and rounds to an OpenTelemetry logs receiver, one record per file, and the metrics spool
 internal/metrics/               derives the runtime's token metric from landed Session Data into the spool; collector side; testdata holds a redacted capture of the runtime's exporter
 internal/repack/                re-cuts landed files into a new root under another budget
-internal/retention/             keeps only idle conversations' metadata: removes landed files, a skeleton in each one's place (collector side)
+internal/retention/             the two levels for idle conversations: keeps only the metadata, a skeleton in each landed file's place, then retires the session whole (collector side)
 internal/adapters/claudecode/   the claude-code-local adapter
 internal/adapters/claudecodeprovider/ the claude-code-provider adapter: the request and response bodies Claude Code writes, landed into their sessions
 internal/adapters/kiro/         the kiro-local adapter: the sessions Kiro's V3 engine writes, for kiro-cli and Kiro Crew
@@ -180,12 +180,21 @@ Every source file carries the Apache-2.0 header; `make license-fix` inserts miss
   and a reader may have taken the file) and is reported once per directory; failing it made a disk
   that kept failing land the same records on every pass. Retention, when it removes, must make a
   replacement durable, its directory synced and the error checked, before it deletes what it replaces.
-- **asz never removes a real session whole.** Only a session a scenario build marked is removed
-  whole, by the policy in its marker, and only once all of it was sent to the one receiver the
-  pipeline sends to. A removal renames each directory away before it deletes it, so no reader sees
-  half a session. `storage.retention.metadata_only_after`, off by default, removes the landed
-  files of an idle conversation, sent or not, and keeps its structure: a skeleton in each file's
-  place, and the index, cursors, state and chain untouched. It never runs over a scenario root.
+- **Only `retire_after` removes a real session whole, and its id is never collected again.** A
+  scenario build's removal takes only a session the build marked, by the policy in its marker, and
+  only once all of it was sent to the one receiver the pipeline sends to. A removal renames each
+  directory away before it deletes it, so no reader sees half a session. `storage.retention` has
+  two levels on one clock, the header time of the newest landed file, sent or not: off by default,
+  `metadata_only_after` removes the landed files of an idle conversation and keeps its structure,
+  a skeleton in each file's place and the index, cursors, state and chain untouched;
+  `retire_after`, at least as long, removes everything and retires the session. Neither runs over
+  a scenario root.
+- **A retired session is never written again.** Its marker `_retired/<session>` is an empty file,
+  written and synced before anything of the session is deleted, and never removed. Every writer
+  asks `storage.Retired` before the first `MkdirAll` or lock that would make the session's or its
+  chain's directory, since taking a lock makes it; a marker that cannot be read is an error, never
+  taken for none. A retired session's unsent metrics requests are written again from their
+  receipts before its directory goes, and keep its owner in `push.state` until they are sent.
 - **A landed file is cut once.** A round addresses records by `{seq, row}` and binds to file
   digests, so a referenced file is never re-cut. A new budget applies to new files; an existing
   root is brought under it with `asz repack` into a new root, where the chains are built again.

asz.yaml

@@ -33,7 +33,7 @@ storage:
   # How long a conversation may be idle, with nothing new landed in any of
   # its streams, before this root keeps less of it. Whether it was sent, and
   # whether an endpoint is set, does not matter. Each value is a duration
-  # such as 72h, or days such as 30d, and -1 is off. Only off keeps
+  # such as 72h, or days such as 30d, and -1 is off. Only both off keeps
   # everything.
   retention:
     # To save disk, keep only the conversation's metadata: what was said and
@@ -42,6 +42,12 @@ storage:
     # conversation. The removed content cannot be restored here; only a
     # receiver that accepted it still has it.
     metadata_only_after: -1
+    # Remove everything of the conversation, its metadata too, and retire
+    # it. A retired conversation is never collected again, even when its
+    # runtime resumes it, so set this longer than the runtime keeps its own
+    # history. It is at least metadata_only_after, which must be set too;
+    # equal to it goes straight to retirement.
+    retire_after: -1
 
 adapters:
   - name: claude-code-local

cmd/asz/main.go

@@ -639,7 +639,7 @@ func cmdVerify(cfg *config.Config, _ config.Adapter, _ bool) error {
 	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))
+		fmt.Printf("%d session(s) are retired, and none is collected again\n", len(retired))
 	}
 
 	// A conversation's rounds are checked too. They are a digest chain, so a
@@ -861,11 +861,11 @@ func cmdCollect(cfg *config.Config, ads []config.Adapter, once bool) error {
 	}
 	defer closeClient()
 	ref.pusher = pusher
-	keep, err := cfg.Storage.Retention.MetadataOnlyAfterDuration()
+	metadataOnly, retire, err := cfg.Storage.Retention.Durations()
 	if err != nil {
 		return err
 	}
-	ref.retain(keep)
+	ref.retain(metadataOnly, retire)
 	if once || ref.base.Mode == config.ModeOnce {
 		// Nothing follows a single pass, so whatever it could not do is
 		// reported rather than left for a pass that never comes.

cmd/asz/parse.go

@@ -29,6 +29,7 @@ import (
 	"github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecode"
 	"github.com/apache/skywalking-ai-sessionizer/internal/config"
 	"github.com/apache/skywalking-ai-sessionizer/internal/parse"
+	"github.com/apache/skywalking-ai-sessionizer/internal/retention"
 	"github.com/apache/skywalking-ai-sessionizer/internal/storage"
 	"github.com/apache/skywalking-ai-sessionizer/internal/view"
 	"github.com/apache/skywalking-ai-sessionizer/pkg/model"
@@ -64,6 +65,10 @@ var rootWait = exportWait
 // waits for a pass of another command that holds it. The root has to be
 // there first: taking the lock makes a directory under it, which would make
 // a mistyped path look like an empty root.
+//
+// It then finishes every retirement a pass left unfinished, as a pass does
+// first. A retirement that cannot finish is said and the command goes on:
+// its marker is there, so the command writes nothing of that session.
 func holdRootFor(root, command string) (*storage.SessionLock, error) {
 	if _, err := os.Stat(root); err != nil {
 		return nil, err
@@ -72,7 +77,14 @@ func holdRootFor(root, command string) (*storage.SessionLock, error) {
 	if errors.Is(err, storage.ErrRootBusy) {
 		return nil, fmt.Errorf("%s: another command held this storage root for %s; nothing was done, run it again", command, rootWait)
 	}
-	return lock, err
+	if err != nil {
+		return nil, err
+	}
+	res := (&retention.Remover{Zone: storage.NewZone(root)}).RecoverRetired()
+	for _, e := range res.Errors {
+		fmt.Fprintf(os.Stderr, "  error: %s: %v\n", command, e)
+	}
+	return lock, nil
 }
 
 // parseZone assembles every session under zoneRoot, or the one named by want,

cmd/asz/refresh.go

@@ -438,14 +438,17 @@ func present(dir string, optional bool) (bool, error) {
 }
 
 // retain sets the retention: how long a conversation may be idle before only
-// its metadata is kept. Zero removes nothing, though a removal an earlier
-// pass left in progress is still finished.
-func (r *refresher) retain(keep time.Duration) {
-	r.retention = &retention.Remover{Zone: r.zone, MetadataOnlyAfter: keep, Metrics: r.deriver != nil}
+// its metadata is kept, and before it is retired. Zero does nothing at that
+// level, though a removal or a retirement an earlier pass left in progress is
+// still finished.
+func (r *refresher) retain(metadataOnly, retire time.Duration) {
+	r.retention = &retention.Remover{Zone: r.zone, MetadataOnlyAfter: metadataOnly, RetireAfter: retire, Metrics: r.deriver != nil}
 }
 
 // forget drops what the page cached of sessions whose files were removed,
-// so it reads their skeletons from then on.
+// so it reads their skeletons from then on, or that were retired: the list
+// no longer names a retired conversation, so nothing would read it again to
+// drop it.
 func (r *refresher) forget(sessions []string) {
 	if r.srv == nil {
 		return
@@ -526,8 +529,11 @@ 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.
+	// retired names the retired sessions a collector found evidence of, and
+	// retiredNow counts the sessions retired this pass, by a retirement that
+	// began in it or one it finished.
 	var retired []string
+	retiredNow := 0
 	changed := map[string]bool{}
 
 	// A removal an earlier pass left in progress is finished first, before
@@ -536,6 +542,8 @@ func (r *refresher) pass() error {
 		res := r.retention.Recover()
 		errs = append(errs, res.Errors...)
 		r.forget(res.Removed)
+		r.forget(res.Retired)
+		retiredNow = len(res.Retired)
 	}
 
 	var cs *claudecode.Stats
@@ -843,7 +851,9 @@ func (r *refresher) pass() error {
 		res := r.retention.Pass(retention.Input{Busy: r.removeInput().Busy})
 		errs = append(errs, res.Errors...)
 		retained = res.Files
+		retiredNow += len(res.Retired)
 		r.forget(res.Removed)
+		r.forget(res.Retired)
 	}
 	if r.last && len(r.retry) > 0 {
 		// No pass follows, so these are not deferred, they are unparsed.
@@ -879,6 +889,9 @@ func (r *refresher) pass() error {
 	if retained > 0 {
 		pushed += fmt.Sprintf(" retention=%d", retained)
 	}
+	if retiredNow > 0 {
+		pushed += fmt.Sprintf(" retired=%d", retiredNow)
+	}
 	bodies := waiting != r.waiting || unreadable != r.unreadable
 	r.waiting, r.unreadable = waiting, unreadable
 	if first || landed > 0 || rounds > 0 || busy > 0 || pushed != "" || len(errs) > 0 || bodies || dropped > 0 {