From 3351b0f8dcdf4f95d85e57b6267db10214f02149 Mon Sep 17 00:00:00 2001 From: Wu Sheng Date: Sat, 17 Oct 2026 08:30:46 +0800 Subject: [PATCH] Keep only the metadata of any idle conversation, sent or not Retention removed the landed files only of a conversation that was sent. One collected with no endpoint, or not sent yet, kept every byte, so a root that only collects grew for ever. Whether to remove now follows the conversation's idle time and the setting alone, never an endpoint or what was sent. A conversation never sent loses its content everywhere, and its skeletons say that no receiver confirmed them. storage.retention becomes a section, so the level that removes the session itself can take its place beside it later. Its first setting, metadata_only_after, is what storage.retention was: how long a conversation may be idle before only its metadata is kept, to save disk. The single value it was is refused, with an error that names storage.retention as a section. Metrics are derived only while an endpoint is set, so a conversation that kept only its metadata with none set reaches a deriver enabled later with none of its executions counted. A skeleton keeps no execution record's protocol or server, so no MCP call is derived from it. The first pass of a process that saves its state now says, for each session, how many distinct execution observations its skeletons keep that no pass counted. It reads only execution files, and a file it cannot read keeps its session for another pass. --- CLAUDE.md | 8 +- asz.yaml | 17 +- cmd/asz/main.go | 4 +- cmd/asz/push.go | 1 + cmd/asz/refresh.go | 31 ++- cmd/asz/retention_test.go | 139 +++++++++++ cmd/asz/view.go | 8 +- docs/en/adapters/claude-code-glossary.md | 2 +- docs/en/adapters/kiro-glossary.md | 2 +- docs/en/adapters/langsmith-glossary.md | 2 +- docs/en/changes/changes.md | 22 +- docs/en/formats/session-data.md | 14 +- docs/en/formats/storage-root.md | 35 +-- docs/en/guides/command-line-internals.md | 30 ++- docs/en/guides/scenario.md | 14 +- docs/en/setup/configuration.md | 2 +- internal/adapters/claudecode/glossary.go | 2 +- internal/adapters/kiro/glossary.go | 2 +- internal/adapters/langsmith/glossary.go | 2 +- internal/config/config.go | 50 ++-- internal/config/config_test.go | 61 ++++- internal/metrics/mcp.go | 71 ++++++ internal/metrics/mcp_test.go | 270 +++++++++++++++++++++ internal/metrics/metrics.go | 35 +++ internal/retention/retention.go | 38 ++- tests/chain/retention_test.go | 289 ++++++++++++++++++++++- 26 files changed, 1022 insertions(+), 129 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 7f6fdcb..dfbf17b 100644 --- a/CLAUDE.md +++ b/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/ removes idle sent conversations' landed files, a skeleton in each one's place (collector side) +internal/retention/ keeps only idle conversations' metadata: removes landed files, a skeleton in each one's place (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 @@ -183,9 +183,9 @@ Every source file carries the Apache-2.0 header; `make license-fix` inserts miss - **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`, off by default, removes the landed files of an idle sent - conversation 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. + 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. - **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. diff --git a/asz.yaml b/asz.yaml index f9e3244..b2842d9 100644 --- a/asz.yaml +++ b/asz.yaml @@ -31,12 +31,17 @@ storage: root: ./data # How long a conversation may be idle, with nothing new landed in any of - # its streams, before its landed files are removed from this root. Each - # file is replaced by a skeleton that keeps the conversation's structure, so - # new activity continues the same conversation; the text is kept only by - # the receiver it was sent to. Only a conversation that was sent is removed. - # A duration such as 72h, or days such as 30d. -1 keeps everything. - retention: -1 + # 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 + # everything. + retention: + # To save disk, keep only the conversation's metadata: what was said and + # run is removed from every landed file, and who did what, when, in what + # order and at what cost stays. New activity continues the same + # conversation. The removed content cannot be restored here; only a + # receiver that accepted it still has it. + metadata_only_after: -1 adapters: - name: claude-code-local diff --git a/cmd/asz/main.go b/cmd/asz/main.go index f600101..f6174c9 100644 --- a/cmd/asz/main.go +++ b/cmd/asz/main.go @@ -588,7 +588,7 @@ func cmdVerify(cfg *config.Config, _ config.Adapter, _ bool) error { } if n := removed[sessiondata.ReceiverAccepted] + removed[sessiondata.ReceiverRejected] + removed[sessiondata.ReceiverUnconfirmed]; n > 0 { - // Removed on purpose, by storage.retention. A skeleton keeps every + // Removed on purpose, by retention. A skeleton keeps every // record's place and the file's digest, so the checks go on through // it, but the content is gone and is not verified here. fmt.Printf("%d file(s) were removed by retention and are not verified: the receiver accepted %d, accepted %d with rejected records, and confirmed nothing of %d\n", @@ -820,7 +820,7 @@ func cmdCollect(cfg *config.Config, ads []config.Adapter, once bool) error { } defer closeClient() ref.pusher = pusher - keep, err := cfg.Storage.RetentionDuration() + keep, err := cfg.Storage.Retention.MetadataOnlyAfterDuration() if err != nil { return err } diff --git a/cmd/asz/push.go b/cmd/asz/push.go index 262649c..b22505b 100644 --- a/cmd/asz/push.go +++ b/cmd/asz/push.go @@ -106,6 +106,7 @@ func cmdPush(cfg *config.Config, _ config.Adapter, _ bool) error { for _, e := range ms.Errors { fmt.Fprintf(os.Stderr, " error: %v\n", e) } + sayUnread(ms.Unread) derived = ms.Errors return nil } diff --git a/cmd/asz/refresh.go b/cmd/asz/refresh.go index 78fd5d3..18d600c 100644 --- a/cmd/asz/refresh.go +++ b/cmd/asz/refresh.go @@ -164,9 +164,9 @@ type refresher struct { // only while this pipeline holds the scenario lock. A root no build wrote // has no marker, so nothing there is ever removed. remover *remove.Remover - // retention removes the landed files of conversations idle longer than - // storage.retention, and finishes a removal an earlier pass left in - // progress. It never runs over a scenario root, which has its own. + // retention keeps only the metadata of conversations idle longer than + // storage.retention.metadata_only_after, and finishes a removal an + // earlier pass left in progress. It never runs over a scenario root, which has its own. retention *retention.Remover // scenarioLock is held for the life of the process once the root is a // scenario root. scenarioWait is how long a single pass waits for it. @@ -434,11 +434,11 @@ func present(dir string, optional bool) (bool, error) { return true, nil } -// retain sets the retention: how long a conversation may be idle before its -// landed files are removed. Zero removes nothing, though a removal an -// earlier pass left in progress is still finished. +// 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, Retention: keep, Metrics: r.deriver != nil} + r.retention = &retention.Remover{Zone: r.zone, MetadataOnlyAfter: keep, Metrics: r.deriver != nil} } // forget drops what the page cached of sessions whose files were removed, @@ -718,6 +718,7 @@ func (r *refresher) pass() error { for _, id := range ms.PendingSessions { r.deferred[id] = true } + sayUnread(ms.Unread) } } rounds := 0 @@ -942,3 +943,19 @@ func (r *refresher) loop() { _ = r.pass() } } + +// sayUnread says, for each session, how many distinct execution observations +// its skeletons keep that no pass counted. A skeleton says neither whether one +// was an MCP call nor why it was not counted, so the line says what is there, +// not what was lost. None is an error. +func sayUnread(unread map[string]int) { + sessions := make([]string, 0, len(unread)) + for id := range unread { + sessions = append(sessions, id) + } + sort.Strings(sessions) + for _, id := range sessions { + fmt.Fprintf(os.Stderr, " metrics: %s: %d execution observations lack retained details; their MCP metrics cannot be derived\n", + id, unread[id]) + } +} diff --git a/cmd/asz/retention_test.go b/cmd/asz/retention_test.go index 08ce49e..5fa1f39 100644 --- a/cmd/asz/retention_test.go +++ b/cmd/asz/retention_test.go @@ -25,7 +25,9 @@ import ( "time" "github.com/apache/skywalking-ai-sessionizer/internal/config" + "github.com/apache/skywalking-ai-sessionizer/internal/export/otlp" "github.com/apache/skywalking-ai-sessionizer/internal/storage" + "github.com/apache/skywalking-ai-sessionizer/internal/view" "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata" ) @@ -203,3 +205,140 @@ func TestShowAndVerifySayWhatWasRemoved(t *testing.T) { t.Fatalf("verify does not report the removed files:\n%s", out) } } + +// TestUnreadObservationsAreSaidBySession. What a deriver could not derive +// is said once a session, in session order, and nothing when it is empty. +func TestUnreadObservationsAreSaidBySession(t *testing.T) { + out := stderrOf(t, func() { sayUnread(map[string]int{"s-b": 2, "s-a": 1}) }) + want := " metrics: s-a: 1 execution observations lack retained details; their MCP metrics cannot be derived\n" + + " metrics: s-b: 2 execution observations lack retained details; their MCP metrics cannot be derived\n" + if out != want { + t.Errorf("said %q, want %q", out, want) + } + if out := stderrOf(t, func() { sayUnread(nil) }); out != "" { + t.Errorf("said %q for nothing unread", out) + } +} + +// unsentRemovedRoot is a root whose conversation a pipeline with no endpoint +// collected and then kept only the metadata of. Nothing was sent, so no +// metrics were derived either. +func unsentRemovedRoot(t *testing.T, scenarioName string) (string, string) { + t.Helper() + root, b := scenarioRoot(t, scenarioName) + if err := os.RemoveAll(filepath.Join(root, storage.ScenarioDir)); err != nil { + t.Fatal(err) + } + zone := storage.NewZone(root) + ref, err := newRefresher(view.New(zone, nil), zone, bothAdapters(root, ""), pipelineMetrics(&config.Config{Metrics: allMetrics}), 2<<20, true) + if err != nil { + t.Fatal(err) + } + t.Cleanup(ref.close) + ref.retain(time.Nanosecond) + stdoutOf(t, func() { + if err := ref.pass(); err != nil { + t.Fatal(err) + } + }) + if sk, landed := skeletonsOf(t, root, b.Session); sk == 0 || landed != 0 { + t.Fatalf("a pipeline with no endpoint kept %d landed files and wrote %d skeletons", landed, sk) + } + return root, b.Session +} + +// A pipeline with no endpoint keeps only the metadata of an idle +// conversation it never sent. asz show says no receiver confirmed the file, +// and asz verify counts it among the removed. +func TestAPipelineWithNoEndpointKeepsOnlyMetadata(t *testing.T) { + root, session := unsentRemovedRoot(t, "growth.yaml") + saved := positional + t.Cleanup(func() { positional = saved }) + cfg := config.Default() + cfg.Storage.Root = root + files, err := storage.LandedFiles(storage.NewZone(root), session) + if err != nil { + t.Fatal(err) + } + f, err := sessiondata.Open(files[0].Path) + if err != nil { + t.Fatal(err) + } + _, recs, err := sessiondata.All(f) + f.Close() + if err != nil { + t.Fatal(err) + } + id := "" + for _, r := range recs { + if r.ID != "" { + id = r.ID + break + } + } + positional = []string{session, id} + out := stdoutOf(t, func() { + if err := cmdShow(cfg, config.Adapter{}, false); err != nil { + t.Fatal(err) + } + }) + if !strings.Contains(out, "no receiver confirmed the file") { + t.Errorf("show output does not say no receiver confirmed the file:\n%s", out) + } + positional = nil + out = stdoutOf(t, func() { + if err := cmdVerify(cfg, config.Adapter{}, false); err != nil { + t.Fatalf("verify over a root with nothing sent: %v", err) + } + }) + if !strings.Contains(out, "accepted 0, accepted 0 with rejected records, and confirmed nothing of") { + t.Fatalf("verify does not report the removed files as confirmed by no receiver:\n%s", out) + } +} + +// Metrics turned on after a conversation kept only its metadata: the first +// pass of a pipeline says how many of its execution observations their +// metrics cannot be derived for, and the pass after it does not. asz push, +// a process of its own, says it again. +func TestLateMetricsAreSaidByThePipelineAndByPush(t *testing.T) { + root, session := unsentRemovedRoot(t, "mcp-calls.yaml") + line := " metrics: " + session + ": " + ref := pipelineOver(t, root, bothAdapters(root, "")) + var out string + stdoutOf(t, func() { + out = stderrOf(t, func() { + if err := ref.pass(); err != nil { + t.Fatal(err) + } + }) + }) + if !strings.Contains(out, line) || !strings.Contains(out, " execution observations lack retained details; their MCP metrics cannot be derived") { + t.Fatalf("the first pass with metrics on said:\n%s", 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) + } + + rcv := receiver(t) + cfg := &config.Config{Metrics: allMetrics} + cfg.Storage.Root = root + o := rcv.Options(otlp.ProtocolGRPC) + cfg.Export.OTLP.Protocol, cfg.Export.OTLP.Endpoint, cfg.Export.OTLP.ServiceName = o.Protocol, o.Endpoint, "Push Test" + stdoutOf(t, func() { + out = stderrOf(t, func() { + if err := cmdPush(cfg, config.Adapter{}, false); err != nil { + t.Fatal(err) + } + }) + }) + if !strings.Contains(out, line) { + t.Fatalf("asz push did not say it:\n%s", out) + } +} diff --git a/cmd/asz/view.go b/cmd/asz/view.go index 65e4795..bd7ecf3 100644 --- a/cmd/asz/view.go +++ b/cmd/asz/view.go @@ -166,7 +166,7 @@ func cmdServer(cfg *config.Config, ads []config.Adapter, once bool) error { } defer closeClient() ref.pusher = pusher - keep, err := cfg.Storage.RetentionDuration() + keep, err := cfg.Storage.Retention.MetadataOnlyAfterDuration() if err != nil { return err } @@ -262,11 +262,11 @@ func printPipeline(ref *refresher, cfg *config.Config) { } fmt.Printf("export : %s, at the end of every pass\n", endpoint) } - if ref.retention != nil && ref.retention.Retention > 0 { - fmt.Printf("retention: a sent conversation idle for %s loses its landed files; a skeleton keeps its structure, and the receiver its text\n", ref.retention.Retention) + if ref.retention != nil && ref.retention.MetadataOnlyAfter > 0 { + fmt.Printf("retention: a conversation idle for %s keeps only its metadata; its content is removed, and only a receiver that accepted it still has it\n", ref.retention.MetadataOnlyAfter) } // A root a claude-code scenario build wrote. Only there does a pipeline - // remove anything, and only what a build marked. + // remove a session whole, and only one a build marked. if fi, err := os.Stat(filepath.Join(ref.zone.Root(), storage.ScenarioDir)); err == nil && fi.IsDir() { fmt.Printf("scenario : %s\n", ref.removalSays()) } diff --git a/docs/en/adapters/claude-code-glossary.md b/docs/en/adapters/claude-code-glossary.md index 4ab4f06..77d8b3f 100644 --- a/docs/en/adapters/claude-code-glossary.md +++ b/docs/en/adapters/claude-code-glossary.md @@ -92,7 +92,7 @@ than a plausible word that would send a reader looking for a field that does not | `in_segment` | the activity window a Talk falls in | | `input_of` | unavailable; the serialized provider request is not written locally | | `omitted` | deliberately not carried | -| `removed` | no equivalent: asz removes the payloads of a conversation idle longer than storage.retention from its own root; Claude Code keeps its transcript as before | +| `removed` | no equivalent: asz removes the payloads of a conversation idle longer than storage.retention.metadata_only_after from its own root; Claude Code keeps its transcript as before | | `retries` | unavailable; retry identity is not recorded | | `segment` | an activity window | | `size_only` | only the length was kept | diff --git a/docs/en/adapters/kiro-glossary.md b/docs/en/adapters/kiro-glossary.md index 06cbbf1..2d341af 100644 --- a/docs/en/adapters/kiro-glossary.md +++ b/docs/en/adapters/kiro-glossary.md @@ -90,7 +90,7 @@ not there. | `notification` | — | derived here | no equivalent | | `omitted` | — | derived here | no equivalent | | `parent` | — | derived here | no equivalent: a record names no record before it | -| `removed` | — | derived here | no equivalent: asz removes the payloads of a conversation idle longer than storage.retention from its own root | +| `removed` | — | derived here | no equivalent: asz removes the payloads of a conversation idle longer than storage.retention.metadata_only_after from its own root | | `reports` | — | derived here | no equivalent | | `retries` | — | derived here | no equivalent | | `runtime.notification` | — | derived here | no equivalent: a subagent's end is the result of its call | diff --git a/docs/en/adapters/langsmith-glossary.md b/docs/en/adapters/langsmith-glossary.md index 86a9415..73d85e2 100644 --- a/docs/en/adapters/langsmith-glossary.md +++ b/docs/en/adapters/langsmith-glossary.md @@ -88,7 +88,7 @@ not there. | `notification` | — | derived here | no equivalent | | `omitted` | — | derived here | a graph's own run: its repeated message list is not landed | | `redacted` | — | derived here | no equivalent: nothing on the wire says it redacted anything | -| `removed` | — | derived here | no equivalent: asz removes the payloads of a conversation idle longer than storage.retention from its own root | +| `removed` | — | derived here | no equivalent: asz removes the payloads of a conversation idle longer than storage.retention.metadata_only_after from its own root | | `reports` | — | derived here | derived | | `retries` | — | derived here | no equivalent yet: the client does not say a run retried another | | `runtime.notification` | — | derived here | no equivalent | diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md index 87c431d..9db0674 100644 --- a/docs/en/changes/changes.md +++ b/docs/en/changes/changes.md @@ -283,13 +283,21 @@ ## Retention -- A storage root no longer only grows. `storage.retention`, `-1` and off by default, removes the - landed files of a conversation that was sent and has had nothing new land for that long, such as - `72h` or `30d`. On one root, 7,235 landed files of 847.5 MB had skeletons of 29.6 MB, 3.5% of - them, measured on 2026-10-13. `asz collect` and `asz server` remove at - the end of a pass, after the send, the whole conversation's files at once, and only once its head - round covers them and, with metrics derived, every file of it is derived. A conversation never - sent is kept. See [Retention](../formats/storage-root.md#retention). +- A storage root no longer only grows. `storage.retention.metadata_only_after`, `-1` and off by + default, keeps only the metadata of a conversation that has had nothing new land for that long, + such as `72h` or `30d`, to save disk: it removes the landed files, and a skeleton of each keeps + who did what, when, in what order and at what cost. It applies whether the conversation was sent + or not, and whether an endpoint is set or not; the content of one never sent is then gone + everywhere. On one root, 7,235 landed files of 847.5 MB had skeletons of 29.6 MB, 3.5% of them, + measured on 2026-10-13. `asz collect` and `asz server` remove at the end of a pass, after the + send, the whole conversation's files at once, and only once its head round covers them and, with + metrics derived, every file of it is derived. See [Retention](../formats/storage-root.md#retention). +- Metrics turned on after a conversation kept only its metadata count its tokens from the + skeletons, which keep no execution record's protocol or server, so no MCP call is derived from + them. The first pass of a process that saves its state says, for each session, how many distinct + execution observations its skeletons keep that no pass counted, as `metrics: : + execution observations lack retained details`. The count does not say how many were MCP calls, or + why one was not counted. Only execution files are read for it. - A provider body landed after a removal is cut exactly as with nothing removed. The cutter holds each removed body by its skeleton, and the session's store, under `provider/`, keeps the bytes of every chain tip at a removal and of each removed piece a later body cites, so the same pieces are diff --git a/docs/en/formats/session-data.md b/docs/en/formats/session-data.md index 2685391..4e2dc47 100644 --- a/docs/en/formats/session-data.md +++ b/docs/en/formats/session-data.md @@ -514,11 +514,12 @@ the whitespace and older-file limits in [What data holds](#what-data-holds). ## Skeletons -When `storage.retention` removes a landed file, a skeleton takes its place: the same name with the -extension `.sk`, in the same directory, written once and read-only. A skeleton is Session Data -compressed with gzip, and it is local: it is never sent. It keeps what a conversation's structure is -built from and none of the content, so the index rebuilt from it is the index rebuilt from the -landed file. [Retention](storage-root.md#retention) says when a file is removed. +When retention keeps only a conversation's metadata, a skeleton takes each landed file's place: +the same name with the extension `.sk`, in the same directory, written once and read-only. A +skeleton is Session Data compressed with gzip, and it is local: it is never sent. It keeps what a +conversation's structure is built from and none of the content, so the index rebuilt from it is +the index rebuilt from the landed file. [Retention](storage-root.md#retention) says when a file is +removed. - **The header**, whole, with one more key, `removed`: @@ -532,7 +533,8 @@ landed file. [Retention](storage-root.md#retention) says when a file is removed. removed, and `generation` which removal of the session removed it, counting from 1. `receiver` says what the receiver confirmed of it: `accepted`, `rejected` when it took the file and rejected some of its records, or `unconfirmed` when no answer confirmed it. `owner` is the owner its - session is sent under, as `push.state` records it. + session is sent under, as `push.state` records it, and is absent when its session was never sent + or tried to be. - **Every record, in order**, with `ord`, `off`, `bytes`, `id`, `parent`, `call`, `run`, `continues`, `tool`, `child`, `batch`, `label`, `started_by`, `from`, `time`, `trigger`, `flags`, `usage`, `model` and `dropped`. A provider body keeps `sha` too; any other record diff --git a/docs/en/formats/storage-root.md b/docs/en/formats/storage-root.md index 9dca564..1f774f9 100644 --- a/docs/en/formats/storage-root.md +++ b/docs/en/formats/storage-root.md @@ -418,27 +418,32 @@ The store had also outlived its source: it held 48 sessions where the source dir ### storage.retention -`storage.retention` removes the landed files of a conversation that has been idle for longer than -it, by asz's clock: nothing new has landed in any of its streams since. It is `-1` by default, which -keeps everything. A pipeline, `asz collect` or `asz server`, removes at the end of a pass, after the -send. A session's files go together, every landed file of it through its newest, never some of -them. A session keeps them while any of these holds: +`storage.retention` says how long an idle conversation keeps what in the root. A conversation is +idle for as long as nothing new has landed in any of its streams, by asz's clock: the time is read +from the header of its newest landed file. Whether it was sent, and whether an endpoint is set, does +not change that. Each setting is `-1`, off, by default, and only off keeps everything. + +`storage.retention.metadata_only_after` keeps only the metadata of a conversation idle for longer +than it, to save disk: it removes the landed files and keeps a skeleton of each. A pipeline, `asz +collect` or `asz server`, removes at the end of a pass, after the send. A session's files go +together, every landed file of it through its newest, never some of them. A session keeps them +while any of these holds: -- Nothing of it was ever sent, or tried to be: `push.state` names no owner for it. - Something landed for it within the retention. - Its head round does not reach its newest landed file, or its parse failed this pass. - With metrics derived, a landed file of it is not derived yet. -Sending is not waited for. A receiver that is down for longer than the retention misses the files -removed meanwhile, and asz does not send them again. A scenario root keeps its own removal, below, -and retention never runs over it. +Sending is neither waited for nor needed. A receiver that is down for longer than the retention +misses the files removed meanwhile, and asz does not send them again. The content of a conversation +never sent is gone everywhere once its files are removed, and its skeletons say that no receiver +confirmed them. A scenario root keeps its own removal, below, and retention never runs over it. Each removed file is replaced by its skeleton, `.sk` in the same directory, written once and read-only. A [skeleton](session-data.md#skeletons) keeps every record and every part in its place, with every field the index is built from, and none of the content: text, data and media go. The index rebuilt from a skeleton is the index rebuilt from its file, so a conversation that becomes active again continues the same chain, with the same rounds, as if nothing was removed. Its text -is kept only by the receiver it was sent to. A skeleton is never sent. Measured on 2026-10-13 on +is kept only by a receiver that accepted it. A skeleton is never sent. Measured on 2026-10-13 on one machine's root, 7,235 landed files of 847.5 MB had skeletons of 29.6 MB, 3.5% of them: 3.5% for transcripts, under 5% for workflow files, and 77% for child agents' sidecars, which are mostly identifiers. Its 16 provider body files of 133 KB had skeletons of 13.6 KB, 10% of them, 0.7 KB of @@ -561,10 +566,12 @@ with a `no rounds` error, and so does the request for the document behind the pa ### A scenario root -asz removes a collected session on its own in one case only: a session a `claude-code` scenario -build wrote. [Scenarios](../guides/scenario.md#removal-a-session-goes-once-it-is-sent) describes -the policy. A real Claude Code session is never removed. Only the build writes the marker a removal -needs, and the product configuration has no removal setting. +asz removes a collected session whole on its own in one case only: a session a `claude-code` +scenario build wrote. [Scenarios](../guides/scenario.md#removal-a-session-goes-once-it-is-sent) +describes the policy. A real session is never removed whole. Only the build writes the marker a +whole removal needs. [Retention](#retention), when it is set, removes the landed content of an idle +conversation, and keeps its structure, in any root but one with `_scenario/`: an sd build's +included. The build writes one marker for each session, `DIR/_source/.asz-scenario/.json`, and a pipeline reads it under the `source_root` of `claude-code-local`. The build writes it after every diff --git a/docs/en/guides/command-line-internals.md b/docs/en/guides/command-line-internals.md index eb378aa..b764452 100644 --- a/docs/en/guides/command-line-internals.md +++ b/docs/en/guides/command-line-internals.md @@ -35,14 +35,24 @@ The pipeline. One pass does three things, in this order: The order matters. The push runs last so a pass sends the rounds it has just written, rather than leaving them for the next one. -With `storage.retention` set, a pass then removes the landed files of every conversation that was -sent and has been idle for longer, leaving a skeleton in each one's place. Before it lands -anything, every pass finishes a removal an earlier one left in progress, whatever the setting says -now. A scenario root keeps its own removal, and retention never runs there. See -[Retention](../formats/storage-root.md#retention). The startup lines say the retention: +With `storage.retention.metadata_only_after` set, a pass then keeps only the metadata of every +conversation idle for longer, sent or not: it removes the landed files and leaves a skeleton in +each one's place. Before it lands anything, every pass finishes a removal an earlier one left in +progress, whatever the setting says now. A scenario root keeps its own removal, and retention never +runs there. See [Retention](../formats/storage-root.md#retention). The startup lines say the +retention: ```text -retention: a sent conversation idle for 72h0m0s loses its landed files; a skeleton keeps its structure, and the receiver its text +retention: a conversation idle for 72h0m0s keeps only its metadata; its content is removed, and only a receiver that accepted it still has it +``` + +When metrics are turned on after a conversation kept only its metadata, the skeletons give its +tokens, and not whether an execution was a call to an MCP server. The first pass of a process says, +for each session, how many distinct execution observations its skeletons keep that no pass counted. +The count does not say how many of them were MCP calls: + +```text + metrics: : 4 execution observations lack retained details; their MCP metrics cannot be derived ``` In watch mode, the default, it repeats every `interval`; with `-once` or `mode: once` it makes one @@ -67,7 +77,7 @@ One line per pass, and a quiet pass prints nothing: | `metrics` | spooled metrics requests sent | | `rejected` | records and points a receiver took and rejected, absent when 0. It also gives an `error:` line, because they are never sent again. | | `removed` | scenario sessions removed, absent when 0. See [A scenario root](#a-scenario-root). | -| `retention` | landed files `storage.retention` removed, absent when 0. See [Retention](../formats/storage-root.md#retention). | +| `retention` | landed files `storage.retention.metadata_only_after` removed, absent when 0. See [Retention](../formats/storage-root.md#retention). | | `errors` | listed on standard error after the line | `busy` is not an error. Running two pipelines on one root is supported, and the process that holds @@ -95,7 +105,9 @@ status. A root a `claude-code` scenario build wrote holds `_scenario/`. There a pass does one more thing, after the send. It removes each session the build marked that is now sent in full, by the policy in the session's marker. [Removal](scenario.md#removal-a-session-goes-once-it-is-sent) says -what that takes. A root no build wrote has no `_scenario/`, and nothing in it is ever removed. +what that takes. A root no build wrote has no `_scenario/`, and no session in it is ever removed +whole. Retention, when it is set, keeps only the metadata of an idle conversation in such a root, +and in one an sd build wrote. It never runs over a root with `_scenario/`. When the root holds `_scenario/` at the start, a `scenario :` line says what the removal will do. It asks the check each removal pass makes before it looks at any session, so it never promises a @@ -349,7 +361,7 @@ remove : 24h after its last record, once sent ``` `--remove` with `--format sd` is refused, because an sd build writes no source and its sessions are -never removed. `check` refuses `--remove`, and `build` refuses zero, a negative duration and +never removed whole. Retention, when it is set, still applies to its root. `check` refuses `--remove`, and `build` refuses zero, a negative duration and anything it cannot read. Each refusal exits with status 2 and writes nothing. ## push diff --git a/docs/en/guides/scenario.md b/docs/en/guides/scenario.md index c557044..2774c2d 100644 --- a/docs/en/guides/scenario.md +++ b/docs/en/guides/scenario.md @@ -143,8 +143,8 @@ step. The pipeline removes only over the storage root the build wrote, `DIR` itself, where the build creates `_scenario/`. The `DIR/asz.yaml` the build writes sets `storage.root` to `DIR`. A pipeline -with another `storage.root` lands and sends the sessions but never removes them, and prints no line -about it. +with another `storage.root` lands and sends the sessions but never removes them whole, and prints +no line about it. A duration counts from the session's last record, never from when it landed, so the same landed data always gives the same answer. A feed stamps each session so its last record falls when it was @@ -152,10 +152,12 @@ written, so a feed's session stays about that long after it was written. A one-s old `--at` is already past its duration, and goes as soon as it is sent. `d` counts whole days, as `metrics.lookback` does. One run applies one policy to every session it writes. -The policy belongs to the scenario, never to the product. The configuration has no removal -setting, and a session no build marked, which is every real Claude Code session, is never removed. -An sd build writes no source and no marker, so its sessions are never removed, and `--remove` with -`--format sd` is refused. +The policy belongs to the scenario, never to the product. The configuration has no setting that +removes a session whole, and a session no build marked, which is every real Claude Code session, is +never removed whole. An sd build writes no source and no marker, so its sessions are never removed +whole, and `--remove` with `--format sd` is refused. `storage.retention`, a separate setting, keeps +only the metadata of an idle conversation in an unmarked root, an sd build's included. It never +runs over a root with `_scenario/`, which keeps its own removal. Nothing is removed that the pipeline cannot prove sent. Each of these keeps every marked session, and the pass says why once, on standard error, as `kept: every marked session: `: diff --git a/docs/en/setup/configuration.md b/docs/en/setup/configuration.md index e178578..5ed4864 100644 --- a/docs/en/setup/configuration.md +++ b/docs/en/setup/configuration.md @@ -17,7 +17,7 @@ curl -fsSL -o asz.yaml "https://raw.githubusercontent.com/apache/skywalking-ai-s | Key | Default | Meaning | | --- | --- | --- | | `root` | `./data` | Where collected data lands and conversations are assembled. A relative path is resolved from the working directory. | -| `retention` | `-1` | How long a conversation may be idle before its landed files are removed from the root, such as `72h` or `30d`. Each file is replaced by a skeleton that keeps the conversation's structure, so new activity continues it; its text is kept only by the receiver it was sent to. Only a conversation that was sent is removed. `-1` keeps everything. See [Retention](../formats/storage-root.md#retention). | +| `retention.metadata_only_after` | `-1` | How long a conversation may be idle, with nothing new landed for it, before the root keeps only its metadata, to save disk: what was said and run is removed, and who did what, when, in what order and at what cost stays. New activity continues the same conversation. The removed content cannot be restored from the root; only a receiver that accepted it still has it. A duration such as `72h`, or days such as `30d`. It applies whether the conversation was sent or not, and whether an endpoint is set or not. `-1` is off, and only off keeps everything. See [Retention](../formats/storage-root.md#retention). | ## adapters diff --git a/internal/adapters/claudecode/glossary.go b/internal/adapters/claudecode/glossary.go index 9b923bc..f37fc73 100644 --- a/internal/adapters/claudecode/glossary.go +++ b/internal/adapters/claudecode/glossary.go @@ -132,7 +132,7 @@ var glossary = model.NewGlossary(Dialect, model.Term{Unified: model.ContentSizeOnly, Note: "only the length was kept"}, model.Term{Unified: model.ContentTruncated, Note: "part of the bytes, with the full length reported"}, model.Term{Unified: model.ContentTooLarge, Note: "no equivalent: asz omits a payload whose record alone is larger than a landed file's byte budget, such as a pasted screenshot"}, - model.Term{Unified: model.ContentRemoved, Note: "no equivalent: asz removes the payloads of a conversation idle longer than storage.retention from its own root; Claude Code keeps its transcript as before"}, + model.Term{Unified: model.ContentRemoved, Note: "no equivalent: asz removes the payloads of a conversation idle longer than storage.retention.metadata_only_after from its own root; Claude Code keeps its transcript as before"}, model.Term{Unified: model.ContentUnavailable, Note: "the runtime never wrote it, or the source was pruned"}, model.Term{Unified: model.TriggerExternal, Native: "origin.kind == \"human\"", Where: "record", diff --git a/internal/adapters/kiro/glossary.go b/internal/adapters/kiro/glossary.go index 7fad068..d5b7167 100644 --- a/internal/adapters/kiro/glossary.go +++ b/internal/adapters/kiro/glossary.go @@ -119,7 +119,7 @@ var glossary = model.NewGlossary(Dialect, model.Term{Unified: model.ContentSizeOnly, Note: "no equivalent"}, model.Term{Unified: model.ContentTruncated, Note: "not read: the model is sent a preview of a large tool output, and that preview is the result here; the whole output is in tool-outputs, which is not collected"}, model.Term{Unified: model.ContentTooLarge, Note: "no equivalent: asz omits a payload whose record alone is larger than a landed file's byte budget"}, - model.Term{Unified: model.ContentRemoved, Note: "no equivalent: asz removes the payloads of a conversation idle longer than storage.retention from its own root"}, + model.Term{Unified: model.ContentRemoved, Note: "no equivalent: asz removes the payloads of a conversation idle longer than storage.retention.metadata_only_after from its own root"}, model.Term{Unified: model.ContentUnavailable, Native: "...", Where: "the content of a Reasoning record"}, // ---- what began a turn ---- diff --git a/internal/adapters/langsmith/glossary.go b/internal/adapters/langsmith/glossary.go index 8af4cf8..b3d4bdb 100644 --- a/internal/adapters/langsmith/glossary.go +++ b/internal/adapters/langsmith/glossary.go @@ -121,7 +121,7 @@ var glossary = model.NewGlossary(Dialect, model.Term{Unified: model.ContentSizeOnly, Note: "no equivalent"}, model.Term{Unified: model.ContentTruncated, Note: "no equivalent: nothing measured arrived truncated"}, model.Term{Unified: model.ContentTooLarge, Note: "no equivalent: asz omits a payload whose record alone is larger than a landed file's byte budget"}, - model.Term{Unified: model.ContentRemoved, Note: "no equivalent: asz removes the payloads of a conversation idle longer than storage.retention from its own root"}, + model.Term{Unified: model.ContentRemoved, Note: "no equivalent: asz removes the payloads of a conversation idle longer than storage.retention.metadata_only_after from its own root"}, model.Term{Unified: model.ContentUnavailable, Note: "a run that arrived unfinished: its content is not there yet"}, // ---- what began a turn ---- diff --git a/internal/config/config.go b/internal/config/config.go index 61ffa48..5a3689a 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -195,17 +195,31 @@ type Storage struct { // Root is where landed data is written. Default "./data", i.e. beside the // binary; tests point it beside their fixtures so a case is self-contained. Root string `yaml:"root"` - // Retention is how long a conversation may be idle before its landed - // files are removed from the root, each replaced by a skeleton. -1 or - // empty keeps everything, the default. See RetentionDuration. - Retention string `yaml:"retention"` + // Retention is how long an idle conversation keeps what in the root. + Retention Retention `yaml:"retention"` } -// RetentionDuration is the retention as a duration: zero for -1 or empty, -// which keep everything, and a d suffix counts days, as a retention is -// spoken of in days. -func (s Storage) RetentionDuration() (time.Duration, error) { - v := strings.TrimSpace(s.Retention) +// Retention is how long a conversation may be idle before the root keeps +// less of it. Idle is measured from the newest file landed for it, whether +// it was sent or not, and whether an endpoint is set or not. Every value is +// -1, off, by default, and only off keeps everything. +type Retention struct { + // MetadataOnlyAfter is how long a conversation may be idle before only + // its metadata is kept, to save disk: each landed file is replaced by + // its skeleton, which keeps who did what, when and in what order, and + // none of what was said or run. See MetadataOnlyAfterDuration. + MetadataOnlyAfter string `yaml:"metadata_only_after"` +} + +// MetadataOnlyAfterDuration is metadata_only_after as a duration: zero for -1 +// or empty, which keep everything, and a d suffix counts days, as a retention +// is spoken of in days. +func (r Retention) MetadataOnlyAfterDuration() (time.Duration, error) { + return retentionDuration("storage.retention.metadata_only_after", r.MetadataOnlyAfter) +} + +func retentionDuration(name, value string) (time.Duration, error) { + v := strings.TrimSpace(value) switch v { case "", "-1": return 0, nil @@ -215,13 +229,13 @@ func (s Storage) RetentionDuration() (time.Duration, error) { // A number of days past what a duration holds would wrap around to // a short one, and remove what was meant to be kept for years. if err != nil || n <= 0 || int64(n) > int64(math.MaxInt64/(24*time.Hour)) { - return 0, fmt.Errorf("config: storage.retention %q is not a positive number of days a duration can hold", v) + return 0, fmt.Errorf("config: %s %q is not a positive number of days a duration can hold", name, v) } return time.Duration(n) * 24 * time.Hour, nil } d, err := time.ParseDuration(v) if err != nil || d <= 0 { - return 0, fmt.Errorf("config: storage.retention %q is not -1 or a duration such as 72h or 7d", v) + return 0, fmt.Errorf("config: %s %q is not -1 or a duration such as 72h or 7d", name, v) } return d, nil } @@ -375,7 +389,7 @@ const ( // Default returns the configuration used when none is supplied. func Default() *Config { return &Config{ - Storage: Storage{Root: "./data", Retention: "-1"}, + Storage: Storage{Root: "./data", Retention: Retention{MetadataOnlyAfter: "-1"}}, Adapters: []Adapter{{ Name: AdapterClaudeCodeLocal, Enabled: true, @@ -479,7 +493,11 @@ func Load(path string) (*Config, error) { // their gaps below; pointers preserve explicit false export switches. var loaded Config if err := yaml.Unmarshal(data, &loaded); err != nil { - return nil, fmt.Errorf("config: parse %s: %w", path, err) + // The library names a Go type a person editing the file has never + // seen. storage.retention is a section, and a single value there is + // the one place a reader is likely to meet it. + return nil, fmt.Errorf("config: parse %s: %s", path, strings.ReplaceAll(err.Error(), + "into config.Retention", "into storage.retention, which is a section of settings such as metadata_only_after")) } if err := checkKeys(data); err != nil { return nil, fmt.Errorf("config: %s: %w", path, err) @@ -487,8 +505,8 @@ func Load(path string) (*Config, error) { if loaded.Storage.Root != "" { cfg.Storage.Root = loaded.Storage.Root } - if loaded.Storage.Retention != "" { - cfg.Storage.Retention = loaded.Storage.Retention + if loaded.Storage.Retention.MetadataOnlyAfter != "" { + cfg.Storage.Retention.MetadataOnlyAfter = loaded.Storage.Retention.MetadataOnlyAfter } if len(loaded.Adapters) > 0 { cfg.Adapters = loaded.Adapters @@ -692,7 +710,7 @@ func (c *Config) Validate() error { if _, err := c.Metrics.LookbackDuration(); err != nil { return err } - if _, err := c.Storage.RetentionDuration(); err != nil { + if _, err := c.Storage.Retention.MetadataOnlyAfterDuration(); err != nil { return err } if err := sessiondata.CheckWithholdable(c.View.Hide); err != nil { diff --git a/internal/config/config_test.go b/internal/config/config_test.go index f17b627..3eee641 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -454,24 +454,63 @@ func TestARefusedViewKeySaysWhere(t *testing.T) { } } -// storage.retention keeps everything by default and for -1, takes a duration -// or a number of days, and refuses anything else, including zero, which -// would remove a conversation the moment it is sent. -func TestRetentionDuration(t *testing.T) { +// storage.retention is a section, read from the file as written: the +// duration given, off when the section or the key is left out or is -1, a +// value that is not a duration refused by the setting's whole name, and the +// single value it once was refused. +func TestRetentionIsReadFromTheFile(t *testing.T) { + path := filepath.Join(t.TempDir(), "asz.yaml") + for body, want := range map[string]time.Duration{ + "storage:\n retention:\n metadata_only_after: 30d\n": 30 * 24 * time.Hour, + "storage:\n retention:\n metadata_only_after: 72h\n": 72 * time.Hour, + "storage:\n retention:\n metadata_only_after: -1\n": 0, + "storage:\n retention: {}\n": 0, + "storage:\n root: ./elsewhere\n": 0, + } { + if err := os.WriteFile(path, []byte(body), 0o644); err != nil { + t.Fatal(err) + } + cfg, err := Load(path) + if err != nil { + t.Fatalf("%q: %v", body, err) + } + if d, err := cfg.Storage.Retention.MetadataOnlyAfterDuration(); err != nil || d != want { + t.Errorf("%q read as %v, %v; want %v", body, d, err, want) + } + } + for body, says := range map[string]string{ + "storage:\n retention:\n metadata_only_after: soon\n": "storage.retention.metadata_only_after", + "storage:\n retention: -1\n": "storage.retention, which is a section", + "storage:\n retention: 30d\n": "storage.retention, which is a section", + } { + if err := os.WriteFile(path, []byte(body), 0o644); err != nil { + t.Fatal(err) + } + if _, err := Load(path); err == nil || !strings.Contains(err.Error(), says) { + t.Errorf("%q was taken, or refused without naming %s: %v", body, says, err) + } + } +} + +// storage.retention.metadata_only_after keeps everything by default and for +// -1, takes a duration or a number of days, and refuses anything else, +// including zero, which would remove a conversation's content the moment it +// landed. +func TestMetadataOnlyAfterDuration(t *testing.T) { for in, want := range map[string]time.Duration{"": 0, "-1": 0, " -1 ": 0, "72h": 72 * time.Hour, "30d": 30 * 24 * time.Hour, "90m": 90 * time.Minute, "106751d": 106751 * 24 * time.Hour} { - if d, err := (Storage{Retention: in}).RetentionDuration(); err != nil || d != want { - t.Errorf("storage.retention %q is %v, %v; want %v", in, d, err, want) + if d, err := (Retention{MetadataOnlyAfter: in}).MetadataOnlyAfterDuration(); err != nil || d != want { + t.Errorf("metadata_only_after %q is %v, %v; want %v", in, d, err, want) } } for _, in := range []string{"0", "0d", "-2", "-1h", "soon", "1x", "213504d", "99999999999d"} { - if d, err := (Storage{Retention: in}).RetentionDuration(); err == nil { - t.Errorf("storage.retention %q was taken as %v", in, d) + if d, err := (Retention{MetadataOnlyAfter: in}).MetadataOnlyAfterDuration(); err == nil { + t.Errorf("metadata_only_after %q was taken as %v", in, d) } } cfg := Default() - cfg.Storage.Retention = "soon" - if err := cfg.Validate(); err == nil || !strings.Contains(err.Error(), "storage.retention") { - t.Fatalf("a configuration with storage.retention soon validates: %v", err) + cfg.Storage.Retention.MetadataOnlyAfter = "soon" + if err := cfg.Validate(); err == nil || !strings.Contains(err.Error(), "storage.retention.metadata_only_after") { + t.Fatalf("a configuration with metadata_only_after soon validates: %v", err) } } diff --git a/internal/metrics/mcp.go b/internal/metrics/mcp.go index 9924526..c3b2a1c 100644 --- a/internal/metrics/mcp.go +++ b/internal/metrics/mcp.go @@ -19,11 +19,15 @@ package metrics import ( "errors" + "fmt" "hash" "io" + "path/filepath" + "strings" "time" "github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecode" + "github.com/apache/skywalking-ai-sessionizer/internal/storage" "github.com/apache/skywalking-ai-sessionizer/pkg/execution" "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata" ) @@ -113,3 +117,70 @@ func deriveExecutions(rd *sessiondata.Reader, f io.Reader, h hash.Hash, hdr *ses res.points = windows(points, res, ss) return res, nil } + +// unread counts the execution observations of a session's skeletons that no +// pass counted. +// +// A skeleton keeps an execution record's id, tool, time and outcome, and not +// its protocol or its server. So whether it was a call to an MCP server is not +// known, and its MCP metrics cannot be derived. Metrics are derived only while +// an endpoint is set, so a session removed with none set reaches a deriver +// enabled later with nothing of its executions counted. Nothing records why an +// observation was not counted before its file was removed, so this is a count +// of what is there now: distinct observations a skeleton keeps that are not +// among the counted ones. Some of them may not have been MCP calls. +// +// Only execution files are opened, by their name: a root's skeletons are +// mostly transcripts, and a reader holds a buffer of its own. +func unread(files []storage.LandedFile, ss *sessionState) (int, error) { + ids := map[string]bool{} + for _, lf := range files { + if !lf.Skeleton || !strings.HasPrefix(filepath.Base(lf.Path), string(sessiondata.KindExecution)+"-") { + continue + } + if err := executionIDs(lf.Path, ids); err != nil { + return 0, fmt.Errorf("%s: %w", lf.Path, err) + } + } + n := 0 + for id := range ids { + if !ss.Calls[executionKey(id)] { + n++ + } + } + return n, nil +} + +// executionIDs adds the observation id of every execution record a file +// holds. +func executionIDs(path string, ids map[string]bool) error { + f, err := sessiondata.Open(path) + if err != nil { + return err + } + defer f.Close() + rd, err := sessiondata.NewReader(f) + if err != nil { + return err + } + if rd.Header().Kind != sessiondata.KindExecution { + return nil + } + for { + rec, err := rd.Next() + if errors.Is(err, io.EOF) { + return nil + } + if err != nil { + return err + } + for _, p := range rec.Parts { + if p.Kind != sessiondata.PartData { + continue + } + if r, ok := execution.Decode(p.Data); ok && r.ID != "" { + ids[r.ID] = true + } + } + } +} diff --git a/internal/metrics/mcp_test.go b/internal/metrics/mcp_test.go index 620b848..ae0ef13 100644 --- a/internal/metrics/mcp_test.go +++ b/internal/metrics/mcp_test.go @@ -18,9 +18,13 @@ package metrics_test import ( + "compress/gzip" "io" "os" "path/filepath" + "runtime" + "slices" + "strconv" "testing" "time" @@ -33,6 +37,7 @@ import ( "github.com/apache/skywalking-ai-sessionizer/pkg/execution" "github.com/apache/skywalking-ai-sessionizer/pkg/model" "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata" + "github.com/apache/skywalking-ai-sessionizer/pkg/skeleton" ) // observed is one record the plugin writes after a call to an MCP server. @@ -139,3 +144,268 @@ func TestAnObservationIsCountedOnceAcrossFiles(t *testing.T) { t.Fatalf("the spool holds %v, want %v", got, want) } } + +// removeAll replaces every landed file of s1 with its skeleton, as retention +// does, so only what a skeleton keeps is left to derive from. +func removeAll(t *testing.T, z *storage.Zone) { + t.Helper() + files, err := storage.LandedFiles(z, "s1") + if err != nil { + t.Fatal(err) + } + for _, lf := range files { + if lf.Skeleton { + continue + } + data, err := os.ReadFile(lf.Path) + if err != nil { + t.Fatal(err) + } + removed := sessiondata.Removed{At: "2026-09-08T00:00:00Z", Generation: 1, Receiver: sessiondata.ReceiverUnconfirmed} + if err := storage.WriteAtomic(storage.SkeletonPath(lf.Path), storage.PermLanded, func(w io.Writer) error { + return skeleton.Write(w, data, removed) + }); err != nil { + t.Fatal(err) + } + if err := os.Chmod(lf.Path, 0o644); err != nil { + t.Fatal(err) + } + if err := os.Remove(lf.Path); err != nil { + t.Fatal(err) + } + } +} + +// TestAnObservationASkeletonKeepsIsSaidOnce. Retention removed the files of a +// session before its executions were all derived. A skeleton keeps an +// observation's id and outcome, not whether it was an MCP call, so its MCP +// metrics cannot be derived, and the deriver says how many observations +// that is. An observation counted before the removal is not among them. One +// that was not an MCP call is, since nothing recorded why it was not +// counted. Two observations of one call are two. A process says it once; +// a restart says it again. +func TestAnObservationASkeletonKeepsIsSaidOnce(t *testing.T) { + z := storage.NewZone(t.TempDir()) + other := observed("toolu_other", "WebFetch", "", 50, execution.OutcomeReturned) + other.Protocol, other.Server = "http", nil + landExecutions(t, z, 1, observed("toolu_a", "mcp__status__lookup", "status", 100, execution.OutcomeReturned), other) + derived := &metrics.Deriver{Zone: z, Options: metrics.Options{Grace: -1, Version: "test"}} + if st, err := derived.Pass(nil); err != nil || len(st.Errors) > 0 || len(st.Unread) != 0 { + t.Fatalf("%v, %+v: a session with no skeleton has nothing unread", err, st) + } + hook := observed("toolu_b", "mcp__status__lookup", "status", 100, execution.OutcomeReturned) + // A second observer of the same call, which writes its own record. + second := observed("toolu_b", "mcp__status__lookup", "status", 100, execution.OutcomeReturned) + second.ID = "toolu_b/second" + landExecutions(t, z, 2, hook, second) + removeAll(t, z) + + d := &metrics.Deriver{Zone: z, Options: metrics.Options{Grace: -1, Version: "test"}} + st, err := d.Pass(nil) + if err != nil || len(st.Errors) > 0 { + t.Fatalf("%v %v", err, st) + } + if st.Unread["s1"] != 3 { + t.Fatalf("unread %v, want 3: the observation that was not an MCP call and both of the second call", st.Unread) + } + if again, err := d.Pass(nil); err != nil || len(again.Unread) != 0 { + t.Fatalf("%v, unread %v: a process says it once", err, again.Unread) + } + restarted := &metrics.Deriver{Zone: z, Options: metrics.Options{Grace: -1, Version: "test"}} + if st, err := restarted.Pass(nil); err != nil || st.Unread["s1"] != 3 { + t.Fatalf("%v, unread %v: a restart says it again", err, st.Unread) + } +} + +// unreadDeriver is a deriver as a process starts it. +func unreadDeriver(z *storage.Zone) *metrics.Deriver { + return &metrics.Deriver{Zone: z, Options: metrics.Options{Grace: -1, Version: "test"}} +} + +// landedPathOf is where landExecutions puts file seq of s1. +func landedPathOf(z *storage.Zone, prefix string, seq uint64) string { + return filepath.Join(z.StreamDir("s1", "main"), storage.LandedName(prefix, storage.Stamp(time.Unix(int64(seq), 0)), seq)) +} + +// handSkeleton writes a skeleton by hand, for a shape no landed file has: +// a file named as one kind and saying another, or parts the skeleton writer +// would not keep. +func handSkeleton(t *testing.T, z *storage.Zone, prefix string, kind sessiondata.Kind, seq uint64, parts ...sessiondata.Part) { + t.Helper() + path := storage.SkeletonPath(landedPathOf(z, prefix, seq)) + err := storage.WriteAtomic(path, storage.PermLanded, func(w io.Writer) error { + zw := gzip.NewWriter(w) + sw, err := sessiondata.NewWriter(zw, &sessiondata.Header{Seq: seq, At: "2026-09-07T00:00:00Z", Kind: kind, + Adapter: "test/0", Dialect: "execution/1", Src: "s1/main.execution", Session: "s1", Stream: "main", + Removed: &sessiondata.Removed{Digest: "0", At: "2026-09-08T00:00:00Z", Generation: 1, Receiver: sessiondata.ReceiverUnconfirmed}}) + if err != nil { + return err + } + for i, p := range parts { + if err := sw.Write(&sessiondata.Record{Ord: uint64(i + 1), Sha: "x", Bytes: 1, ID: "r" + strconv.Itoa(i), + Parts: []sessiondata.Part{p}}); err != nil { + return err + } + } + if err := sw.Close(); err != nil { + return err + } + return zw.Close() + }) + if err != nil { + t.Fatal(err) + } +} + +func executionPart(t *testing.T, r *execution.Record, kind sessiondata.PartKind) sessiondata.Part { + t.Helper() + line, err := r.Marshal() + if err != nil { + t.Fatal(err) + } + return sessiondata.Part{Kind: kind, Data: line, State: model.ContentRemoved, Bytes: len(line)} +} + +// TestAnUnsavedPassSaysItsCountAgain. A pass that counted the observations a +// skeleton keeps and then could not save its state returns nothing, so its +// count was never said. The next pass of the same process says it, and the +// one after it does not. +func TestAnUnsavedPassSaysItsCountAgain(t *testing.T) { + if runtime.GOOS == "windows" || os.Geteuid() == 0 { + t.Skip("a read-only directory stops a write only for a user that is not root, and not on Windows") + } + z := storage.NewZone(t.TempDir()) + landExecutions(t, z, 1, observed("toolu_a", "mcp__status__lookup", "status", 100, execution.OutcomeReturned)) + removeAll(t, z) + if _, err := unreadDeriver(z).Pass(nil); err != nil { + t.Fatal(err) + } + dir := filepath.Join(z.Root(), storage.SpoolDir) + if err := os.Chmod(dir, 0o555); err != nil { + t.Fatal(err) + } + d := unreadDeriver(z) + _, err := d.Pass(nil) + if chmodErr := os.Chmod(dir, 0o755); chmodErr != nil { + t.Fatal(chmodErr) + } + if err == nil { + t.Fatal("a pass that could not save its state succeeded") + } + if st, err := d.Pass(nil); err != nil || st.Unread["s1"] != 1 { + t.Fatalf("%v, unread %v: the pass after an unsaved one must say the count", err, st.Unread) + } + if st, err := d.Pass(nil); err != nil || len(st.Unread) != 0 { + t.Fatalf("%v, unread %v: then it is said", err, st.Unread) + } +} + +// TestASkeletonThatCannotBeReadIsTriedAgain. A session's files were all +// derived, then removed, and one of its execution skeletons cannot be read. +// The session stays pending, so a later pass reads it again once it can, and +// says the count then. +func TestASkeletonThatCannotBeReadIsTriedAgain(t *testing.T) { + z := storage.NewZone(t.TempDir()) + other := observed("toolu_other", "WebFetch", "", 50, execution.OutcomeReturned) + other.Protocol, other.Server = "http", nil + landExecutions(t, z, 1, other) + if _, err := unreadDeriver(z).Pass(nil); err != nil { + t.Fatal(err) + } + removeAll(t, z) + sk := storage.SkeletonPath(landedPathOf(z, "execution", 1)) + good, err := os.ReadFile(sk) + if err != nil { + t.Fatal(err) + } + if err := os.Chmod(sk, 0o644); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(sk, []byte("not a skeleton"), 0o644); err != nil { + t.Fatal(err) + } + d := unreadDeriver(z) + st, err := d.Pass(nil) + if err != nil || len(st.Errors) == 0 || !slices.Contains(st.PendingSessions, "s1") { + t.Fatalf("%v, %+v: a skeleton that cannot be read must keep its session pending", err, st) + } + if err := os.WriteFile(sk, good, 0o444); err != nil { + t.Fatal(err) + } + if st, err := d.Pass(st.PendingSessions); err != nil || len(st.Errors) > 0 || st.Unread["s1"] != 1 { + t.Fatalf("%v, %+v: once it reads, the count is said", err, st) + } +} + +// TestOnlyExecutionSkeletonsAreRead. A transcript's skeleton holds no +// execution record, and a root's skeletons are mostly transcripts, so the +// count does not open them: a transcript skeleton that cannot be read is no +// error, once its file was derived. +func TestOnlyExecutionSkeletonsAreRead(t *testing.T) { + z := storage.NewZone(t.TempDir()) + path := landedPathOf(z, "transcript", 1) + err := storage.WriteAtomic(path, storage.PermLanded, func(w io.Writer) error { + sw, err := sessiondata.NewWriter(w, &sessiondata.Header{Seq: 1, At: "2026-09-07T00:00:00Z", Kind: sessiondata.KindTranscript, + Adapter: "test/0", Dialect: "test/1", Src: "s1/main.jsonl", Session: "s1", Stream: "main"}) + if err != nil { + return err + } + if err := sw.Write(&sessiondata.Record{Ord: 1, Sha: "x", Bytes: 2, ID: "u1", + Parts: []sessiondata.Part{{Kind: sessiondata.PartText, Text: "hi", State: model.ContentAvailable, Bytes: 2}}}); err != nil { + return err + } + return sw.Close() + }) + if err != nil { + t.Fatal(err) + } + if _, err := unreadDeriver(z).Pass(nil); err != nil { + t.Fatal(err) + } + removeAll(t, z) + sk := storage.SkeletonPath(path) + if err := os.Chmod(sk, 0o644); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(sk, []byte("not a skeleton"), 0o644); err != nil { + t.Fatal(err) + } + if st, err := unreadDeriver(z).Pass(nil); err != nil || len(st.Errors) > 0 { + t.Fatalf("%v, %+v: a transcript skeleton was opened", err, st) + } +} + +// TestTheCountReadsWhatItShould. An observation landed twice, in two +// skeletons, is one. An observation counted before the removal is not among +// them, also when the state that said so was lost and its file's receipt says +// it again. A part that is not data, a record with no id, and a file named +// for executions that says it holds something else add nothing. +func TestTheCountReadsWhatItShould(t *testing.T) { + z := storage.NewZone(t.TempDir()) + counted := observed("toolu_a", "mcp__status__lookup", "status", 100, execution.OutcomeReturned) + landExecutions(t, z, 1, counted) + if _, err := unreadDeriver(z).Pass(nil); err != nil { + t.Fatal(err) + } + twice := observed("toolu_b", "mcp__status__lookup", "status", 100, execution.OutcomeReturned) + landExecutions(t, z, 2, twice) + landExecutions(t, z, 3, twice) + removeAll(t, z) + if err := os.Remove(filepath.Join(z.Root(), storage.SpoolDir, metrics.StateFile)); err != nil { + t.Fatal(err) + } + unnamed := observed("toolu_c", "mcp__status__lookup", "status", 100, execution.OutcomeReturned) + unnamed.ID = "" + inText := observed("toolu_d", "mcp__status__lookup", "status", 100, execution.OutcomeReturned) + handSkeleton(t, z, "execution", sessiondata.KindExecution, 4, + executionPart(t, unnamed, sessiondata.PartData), executionPart(t, inText, sessiondata.PartText)) + misnamed := observed("toolu_e", "mcp__status__lookup", "status", 100, execution.OutcomeReturned) + handSkeleton(t, z, "execution", sessiondata.KindTranscript, 5, executionPart(t, misnamed, sessiondata.PartData)) + st, err := unreadDeriver(z).Pass(nil) + if err != nil { + t.Fatal(err) + } + if st.Unread["s1"] != 1 { + t.Fatalf("unread %v, want 1: the observation landed twice, and nothing else", st.Unread) + } +} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index c9b7cc0..5435ad5 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -174,6 +174,10 @@ type Stats struct { // continuation or could not be derived. A watching pipeline must carry // them even when only other sessions land new records. PendingSessions []string + // Unread counts, by session, the execution observations kept in its + // skeletons that no pass counted. A process says a session once, in the + // first pass whose state it saved. + Unread map[string]int } // Deriver derives from the landed files of a root that no earlier pass @@ -188,6 +192,9 @@ type Stats struct { type Deriver struct { Zone *storage.Zone Options + // reported names the sessions this process has said Unread for, in a + // pass whose state was saved. + reported map[string]bool } // Pass derives what is new in the sessions named, or in every session when @@ -226,6 +233,7 @@ func (d *Deriver) Pass(sessions []string) (*Stats, error) { return nil, err } } + var scanned []string for _, session := range sessions { // 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 @@ -308,6 +316,25 @@ func (d *Deriver) Pass(sessions []string) (*Stats, error) { state.Derived[rel] = receipt.Digest st.Files++ } + if !d.reported[session] { + // A file that cannot be read now may be readable later, so the + // session is kept for another pass, as a file that could not + // be derived is. + n, err := unread(files, ss) + switch { + case err != nil: + st.Errors = append(st.Errors, err) + pending = true + case n > 0: + if st.Unread == nil { + st.Unread = map[string]int{} + } + st.Unread[session] = n + scanned = append(scanned, session) + default: + scanned = append(scanned, session) + } + } if pending { st.PendingSessions = append(st.PendingSessions, session) } else { @@ -317,6 +344,14 @@ func (d *Deriver) Pass(sessions []string) (*Stats, error) { if err := state.save(filepath.Join(spool.Dir(), StateFile), d.Now()); err != nil { return nil, err } + // Only now: a pass whose state was not saved returns nothing, so what it + // counted was never said, and the next pass counts it again. + if d.reported == nil { + d.reported = map[string]bool{} + } + for _, session := range scanned { + d.reported[session] = true + } return st, nil } diff --git a/internal/retention/retention.go b/internal/retention/retention.go index b7fe458..d7b30cf 100644 --- a/internal/retention/retention.go +++ b/internal/retention/retention.go @@ -15,9 +15,9 @@ // specific language governing permissions and limitations // under the License. -// Package retention removes the landed files of conversations that have been -// idle for longer than storage.retention, and leaves a skeleton in the place -// of each. A skeleton keeps every field the index is built from, so the +// Package retention keeps only the metadata of conversations that have been +// idle for longer than storage.retention.metadata_only_after: it removes their +// landed files, and leaves a skeleton in the place of each. A skeleton keeps every field the index is built from, so the // conversation's index, and so its chain, go on as if nothing was removed. // // It is collector side: it reads the round chain by its format alone, as the @@ -52,10 +52,11 @@ import ( // Remover removes idle conversations' landed files from one storage root. type Remover struct { Zone *storage.Zone - // Retention is how long a conversation must be idle before its landed - // files are removed. Zero removes nothing, though Recover still finishes - // a removal an earlier pass left in progress. - Retention time.Duration + // MetadataOnlyAfter is how long a conversation must be idle before its + // landed files are removed, whether it was sent or not. Zero removes + // nothing, though Recover still finishes a removal an earlier pass left in + // progress. + MetadataOnlyAfter time.Duration // Metrics says the pipeline derives metrics, so every file's derivation // must be committed before the file is removed: a removed file keeps no // execution record to derive from. @@ -170,7 +171,7 @@ func (r *Remover) Recover() Result { // what metrics derived from a removed file once the receiver has it. func (r *Remover) Pass(in Input) Result { var res Result - if r.Retention <= 0 { + if r.MetadataOnlyAfter <= 0 { return res } sessions, err := sessionDirs(r.Zone.Root()) @@ -185,16 +186,14 @@ func (r *Remover) Pass(in Input) Result { return } for _, id := range sessions { - // A session never sent is passed over before its locks are - // taken; removable asks again under them. - if in.Busy[id] || sent.Owner(id) == "" { + if in.Busy[id] { continue } n, err := r.locked(id, func() (int, error) { if err := r.tidy(id, sent, requests); err != nil { return 0, err } - why, err := r.removable(id, sent) + why, err := r.removable(id) if err != nil || why != "" { return 0, err } @@ -268,14 +267,11 @@ func (r *Remover) locked(id string, fn func() (int, error)) (int, error) { // // A session goes once every file of it is parsed, its metrics are derived // when the pipeline derives them, and nothing new has landed for the whole -// retention. Sending is not waited for: a receiver that is down longer than -// the retention misses the files removed meanwhile, and asz does not repair -// that. A session never sent is kept until a later change covers it: its -// removal would start it again from its sources. -func (r *Remover) removable(id string, sent *otlp.Sent) (string, error) { - if sent.Owner(id) == "" { - return "never sent", nil - } +// retention. Sending is neither waited for nor needed: a receiver that is down +// longer than the retention misses the files removed meanwhile, a session +// never sent loses its content everywhere, and asz repairs neither. Its +// skeletons keep its structure, so new activity continues it either way. +func (r *Remover) removable(id string) (string, error) { files, err := storage.LandedFiles(r.Zone, id) if err != nil { return "", err @@ -297,7 +293,7 @@ func (r *Remover) removable(id string, sent *otlp.Sent) (string, error) { if err != nil { return "", err } - if r.now().Sub(at) < r.Retention { + if r.now().Sub(at) < r.MetadataOnlyAfter { return "active", nil } through, ok, err := chainThrough(r.Zone.Root(), id) diff --git a/tests/chain/retention_test.go b/tests/chain/retention_test.go index 0baa5de..a188c12 100644 --- a/tests/chain/retention_test.go +++ b/tests/chain/retention_test.go @@ -33,6 +33,7 @@ import ( "testing" "time" + collmetricspb "go.opentelemetry.io/proto/otlp/collector/metrics/v1" "google.golang.org/protobuf/proto" "github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecode" @@ -41,6 +42,7 @@ import ( "github.com/apache/skywalking-ai-sessionizer/internal/export/otlp" "github.com/apache/skywalking-ai-sessionizer/internal/export/otlp/otlptest" "github.com/apache/skywalking-ai-sessionizer/internal/index" + "github.com/apache/skywalking-ai-sessionizer/internal/metrics" "github.com/apache/skywalking-ai-sessionizer/internal/retention" "github.com/apache/skywalking-ai-sessionizer/internal/scenario" "github.com/apache/skywalking-ai-sessionizer/internal/scenario/expect" @@ -67,7 +69,7 @@ func landedAt(k int) time.Time { return time.Date(2026, 3, 1, k, 0, 0, 0, time.U func later() time.Time { return time.Date(2027, 1, 1, 0, 0, 0, 0, time.UTC) } func retentionOf(r *sentRoot) *retention.Remover { - return &retention.Remover{Zone: r.zone, Retention: time.Hour, Metrics: true, Now: later} + return &retention.Remover{Zone: r.zone, MetadataOnlyAfter: time.Hour, Metrics: true, Now: later} } // pipelineAt is one pass of asz collect, collecting on the clock given. @@ -433,9 +435,10 @@ func skeletons(t *testing.T, z *storage.Zone, session string) []sessiondata.Head return out } -// A session keeps its files while any reason holds: it was never sent, it is -// still active, a landed file is not parsed, its metrics are not derived, or -// its parse failed this pass. Once none holds, the files go. +// A session keeps its files while any reason holds: it is still active, a +// landed file is not parsed, its metrics are not derived, or its parse failed +// this pass. Once none holds, the files go. Whether it was sent is not one of +// them: see TestASessionKeepsOnlyItsMetadataSentOrNot. func TestRetentionKeepsWhatItMust(t *testing.T) { type setup func(r *sentRoot, k int) full := func(r *sentRoot, k int) { r.pipelineAt(landedAt(k)) } @@ -444,11 +447,6 @@ func TestRetentionKeepsWhatItMust(t *testing.T) { land setup remove func(r *sentRoot, rem *retention.Remover, session string) retention.Result }{ - {"never sent", func(r *sentRoot, k int) { - r.collectAt(landedAt(k)) - r.derive() - r.parse() - }, nil}, {"active", full, func(_ *sentRoot, rem *retention.Remover, _ string) retention.Result { rem.Now = func() time.Time { return landedAt(1).Add(30 * time.Minute) } return rem.Pass(retention.Input{}) @@ -535,6 +533,279 @@ func TestRetentionKeepsWhatItMust(t *testing.T) { }) } +// A session keeps only its metadata once it is idle whether it was sent or +// not: with no endpoint set, with one set and nothing sent yet, with a send +// tried that no answer confirmed, and with part of it sent, the last two +// also with the endpoint taken away afterwards. Each skeleton says what the +// receiver confirmed of its file, and names the owner only where a send was +// tried. The root verifies, the page draws the conversation and serves its +// provider files as their skeletons, and a later send carries its rounds and +// none of its removed files. +func TestASessionKeepsOnlyItsMetadataSentOrNot(t *testing.T) { + sc, err := scenario.Load(filepath.Join("..", "scenarios", "provider-bodies-long.yaml")) + if err != nil { + t.Fatal(err) + } + tried := func(r *sentRoot) { + r.rcv.Fail(true) + defer r.rcv.Fail(false) + client, err := r.rcv.Client(otlp.ProtocolGRPC) + if err != nil { + t.Fatal(err) + } + defer client.Close() + p := &otlp.Pusher{Zone: r.zone, Client: client, Endpoint: r.endpoint, Version: "test", + ServiceName: "Removal Test", InstanceID: "removal-test", Layer: "AI_AGENT"} + if _, err := p.Pass(); err != nil { + t.Fatal(err) + } + } + whole := func(r *sentRoot, metrics bool) *scenario.Built { + b := r.build(sc, scenario.Options{}) + r.collectAt(landedAt(0)) + if metrics { + r.derive() + } + r.parse() + return b + } + for _, c := range []struct { + name string + // metrics is whether the pipeline that removes derives, as it does + // only with an endpoint set, so it is false once the endpoint is + // gone, whatever was derived before; owned is whether a send was + // tried; sendAfter is whether the endpoint is still there after the + // removal. + metrics, owned, sendAfter bool + land func(r *sentRoot) *scenario.Built + receivers []string + }{ + {"no endpoint", false, false, false, func(r *sentRoot) *scenario.Built { return whole(r, false) }, + []string{sessiondata.ReceiverUnconfirmed}}, + {"an endpoint, nothing sent", true, false, true, func(r *sentRoot) *scenario.Built { return whole(r, true) }, + []string{sessiondata.ReceiverUnconfirmed}}, + {"a send no answer confirmed", true, true, true, func(r *sentRoot) *scenario.Built { + b := whole(r, true) + tried(r) + return b + }, []string{sessiondata.ReceiverUnconfirmed}}, + {"a send no answer confirmed, the endpoint then gone", false, true, false, func(r *sentRoot) *scenario.Built { + b := whole(r, true) + tried(r) + return b + }, []string{sessiondata.ReceiverUnconfirmed}}, + {"part of it sent, the endpoint then gone", false, true, false, func(r *sentRoot) *scenario.Built { + r.build(sc, scenario.Options{Through: sc.Checkpoints()[0]}) + r.pipelineAt(landedAt(0)) + b := r.build(sc, scenario.Options{}) + r.collectAt(landedAt(1)) + r.derive() + r.parse() + return b + }, []string{sessiondata.ReceiverAccepted, sessiondata.ReceiverUnconfirmed}}, + } { + t.Run(c.name, func(t *testing.T) { + r := newSentRoot(t) + b := c.land(r) + rem := retentionOf(r) + rem.Metrics = c.metrics + if res := rem.Pass(retention.Input{}); len(res.Errors) > 0 || res.Files == 0 { + t.Fatalf("an idle session kept its files: %+v", res) + } + sent, err := otlpSent(r) + if err != nil { + t.Fatal(err) + } + owner := sent.Owner(b.Session) + if (owner != "") != c.owned { + t.Fatalf("owner %q: only a tried send names one", owner) + } + heads := skeletons(t, r.zone, b.Session) + if len(heads) == 0 { + t.Fatal("no skeleton took a removed file's place") + } + said := map[string]bool{} + for _, h := range heads { + if h.Removed == nil || h.Removed.Owner != owner { + t.Fatalf("a skeleton says %+v, want the owner %q", h.Removed, owner) + } + said[h.Removed.Receiver] = true + } + if len(said) != len(c.receivers) { + t.Fatalf("the skeletons say %v, want %v", said, c.receivers) + } + for _, want := range c.receivers { + if !said[want] { + t.Fatalf("the skeletons say %v, want %v", said, c.receivers) + } + } + landed, _ := storage.LandedFiles(r.zone, b.Session) + for _, lf := range landed { + if !lf.Skeleton { + t.Fatalf("%s was not removed", lf.Path) + } + } + verified(t, r.zone, b.Session) + + h := view.New(r.zone, nil).Handler() + get := func(path string) (int, []byte) { + w := httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequest(http.MethodGet, path, nil)) + return w.Code, w.Body.Bytes() + } + code, body := get("/api/c/" + b.Session + "/view") + if code != http.StatusOK || !bytes.Contains(body, []byte(`"state":"removed"`)) { + t.Fatalf("the document is %d and says nothing was removed", code) + } + var doc sessionview.Conversation + if err := json.Unmarshal(body, &doc); err != nil { + t.Fatal(err) + } + q := url.Values{"session": {b.Session}} + bodies := 0 + for _, f := range doc.Files { + if f.Kind == string(sessiondata.KindProviderBody) && f.Seq != nil { + q.Add("seq", fmt.Sprint(*f.Seq)) + bodies++ + } + } + if bodies == 0 { + t.Fatal("the scenario lands no provider body; the test needs some") + } + code, body = get("/api/c/" + b.Session + "/files?" + q.Encode()) + var served []struct { + Seq uint64 `json:"seq"` + Bytes []byte `json:"bytes"` + } + if code != http.StatusOK || json.Unmarshal(body, &served) != nil || len(served) != bodies { + t.Fatalf("the files route answered %d with %d of %d provider files", code, len(served), bodies) + } + for _, f := range served { + if len(f.Bytes) == 0 { + t.Fatalf("provider file %d was served empty", f.Seq) + } + } + + if !c.sendAfter { + return + } + r.rcv.Reset() + r.push() + names := map[string]bool{} + for _, rec := range otlptest.Records(r.rcv.Requests()) { + names[otlptest.Attrs(rec.Attributes)["asz.file"]] = true + } + rounds := 0 + for name := range names { + if strings.HasPrefix(name, b.Session+"/") { + t.Fatalf("%s was sent, although retention removed it", name) + } + if strings.HasPrefix(name, "_conversations/") { + rounds++ + } + } + if rounds == 0 { + t.Fatal("the send carried no round of the conversation") + } + }) + } +} + +// Metrics turned on after a session that was never sent kept only its +// metadata count the same tokens from its skeletons as from its landed files, +// derive no MCP call from them, and say how many of its execution +// observations they cannot derive MCP metrics for: every distinct one, since +// nothing of them was counted before the removal. +func TestLateMetricsSayWhatTheyCannotDerive(t *testing.T) { + sc, err := scenario.Load(filepath.Join("..", "scenarios", "mcp-calls.yaml")) + if err != nil { + t.Fatal(err) + } + plain, kept := newSentRoot(t), newSentRoot(t) + var b *scenario.Built + for _, r := range []*sentRoot{plain, kept} { + b = r.build(sc, scenario.Options{}) + r.collectAt(landedAt(0)) + r.parse() + } + observations := map[string]bool{} + landed, _ := storage.LandedFiles(kept.zone, b.Session) + for _, lf := range landed { + f, err := sessiondata.Open(lf.Path) + if err != nil { + t.Fatal(err) + } + hdr, recs, err := sessiondata.All(f) + f.Close() + if err != nil { + t.Fatal(err) + } + if hdr.Kind != sessiondata.KindExecution { + continue + } + for _, rec := range recs { + observations[rec.ID] = true + } + } + if len(observations) == 0 { + t.Fatal("the scenario lands no execution record") + } + rem := retentionOf(kept) + rem.Metrics = false + if res := rem.Pass(retention.Input{}); res.Files == 0 || len(res.Errors) > 0 { + t.Fatalf("the removal: %+v", res) + } + if plain.derive() == 0 { + t.Fatal("the landed files derive no request") + } + st, err := (&metrics.Deriver{Zone: kept.zone, Options: metrics.Options{Grace: -1, Version: "test"}}).Pass(nil) + if err != nil || len(st.Errors) > 0 { + t.Fatalf("%v %v", err, st) + } + if st.Unread[b.Session] != len(observations) { + t.Errorf("unread %v, want the %d observations the session landed", st.Unread, len(observations)) + } + whole, removed := metricTotals(t, plain.dir), metricTotals(t, kept.dir) + if whole[metrics.TokenUsage] == 0 || whole[metrics.TokenUsage] != removed[metrics.TokenUsage] { + t.Errorf("tokens: %v from the landed files, %v from the skeletons", whole[metrics.TokenUsage], removed[metrics.TokenUsage]) + } + if whole[metrics.MCPCalls] == 0 || removed[metrics.MCPCalls] != 0 || removed[metrics.MCPDuration] != 0 { + t.Errorf("MCP calls: %v from the landed files, %v and %v ms from the skeletons", whole[metrics.MCPCalls], + removed[metrics.MCPCalls], removed[metrics.MCPDuration]) + } +} + +// metricTotals adds up every point of every spooled request of a root, by +// metric. +func metricTotals(t *testing.T, dir string) map[string]float64 { + t.Helper() + files, err := storage.NewSpool(storage.NewZone(dir)).List() + if err != nil { + t.Fatal(err) + } + out := map[string]float64{} + for _, f := range files { + data, err := os.ReadFile(f.Path) + if err != nil { + t.Fatal(err) + } + var req collmetricspb.ExportMetricsServiceRequest + if err := proto.Unmarshal(data, &req); err != nil { + t.Fatal(err) + } + for _, rm := range req.ResourceMetrics { + for _, sm := range rm.ScopeMetrics { + for _, m := range sm.Metrics { + for _, p := range m.GetSum().GetDataPoints() { + out[m.Name] += p.GetAsDouble() + } + } + } + } + } + return out +} + // Each skeleton says what the receiver confirmed of its file: accepted, // accepted with rejected records, or nothing, for a file landed after the // last send. Each removal is a generation of its own.