You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2026/10/12 12:32:57 UTC
(skywalking-ai-sessionizer) branch main updated: Fit every landed file to its budget, and take out what one record cannot fit (#62)
This is an automated email from the ASF dual-hosted git repository.
wu-sheng pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-ai-sessionizer.git
The following commit(s) were added to refs/heads/main by this push:
new 9dd07db Fit every landed file to its budget, and take out what one record cannot fit (#62)
9dd07db is described below
commit 9dd07db6c947b46bf6d97a5ffccd6c3051f45d33
Author: 吴晟 Wu Sheng <wu...@foxmail.com>
AuthorDate: Mon Oct 12 20:32:51 2026 +0800
Fit every landed file to its budget, and take out what one record cannot fit (#62)
A landed file travels whole as one log record, so max_delta_bytes is the
most a receiver has to store at once. Two things broke that.
A collector cut a file by the bytes it read, and a record is larger than its
source line by its envelope: 4 KiB of short lines landed as 19 KB. Files are
now cut by what their records take once written (sessiondata.Cut), in the
Claude Code, Kiro, changes and LangSmith collectors and the scenario writer.
One record cannot be cut: a pasted screenshot measured 4.6 MB in one
transcript line. Such a record lands alone, and its largest payloads are
taken out, one at a time, until the file fits. A part loses its data before
its text, so a result keeps its readable text and says truncated; a part
with nothing left says too_large. Either keeps its place, kind and
identifiers, the original size, and the SHA-256 of what went first, in a
new part field, sha256. A change record or an execution record is never
taken out whole, since a step joins to it: a change record shrinks in its
own format, its hunks and then its file list going, and its document entry
says so with a new key, state. The provider cutter fits its own records and
never takes out a manifest: a later body names a piece taken out as any
piece, and a body of short strings lands whole and is taken out whole. Such
a body cannot be rebuilt from the root; Session.BodyFrom rebuilds it from
another copy. A step that stands on a whole record says too_large when a part
of it was taken out. asz verify counts both apart from damage, and holds
every body's record to its manifest.
The tail reader, after one line longer than the budget, read on to the end
of the source and landed every line up to the last in one file: one 4.6 MB
line made an 8 MB file. It now takes that one line, and stops growing its
window when the source was cut short while it was read.
---
CLAUDE.md | 8 +-
asz.yaml | 14 +-
cmd/asz/main.go | 13 +-
docs/en/adapters/claude-code-glossary.md | 1 +
docs/en/adapters/kiro-glossary.md | 1 +
docs/en/adapters/langsmith-glossary.md | 1 +
docs/en/changes/changes.md | 36 ++
docs/en/formats/asz-view.md | 3 +-
docs/en/formats/otlp.md | 7 +-
docs/en/formats/session-data.md | 60 +++-
docs/en/guides/command-line-internals.md | 8 +
docs/en/setup/command-line.md | 5 +-
docs/en/setup/configuration.md | 2 +-
internal/adapters/claudecode/collect.go | 33 +-
internal/adapters/claudecode/glossary.go | 1 +
internal/adapters/claudecode/regress_test.go | 144 ++++++++
internal/adapters/claudecode/tail.go | 54 ++-
internal/adapters/claudecode/tail_internal_test.go | 109 ++++++
internal/adapters/claudecodechanges/collect.go | 25 +-
.../adapters/claudecodechanges/collect_test.go | 66 ++++
internal/adapters/claudecodeprovider/collect.go | 5 +-
internal/adapters/kiro/collect.go | 41 ++-
internal/adapters/kiro/glossary.go | 1 +
internal/adapters/kiro/kiro_test.go | 110 ++++++
internal/adapters/langsmith/bodies.go | 1 +
internal/adapters/langsmith/budget_test.go | 43 +++
internal/adapters/langsmith/collect.go | 50 ++-
internal/adapters/langsmith/glossary.go | 1 +
internal/config/config.go | 5 +-
internal/scenario/expect/budget_test.go | 84 +++++
internal/scenario/expect/expect.go | 104 +++++-
internal/scenario/run/run.go | 3 +
internal/scenario/run/sourcebytes.go | 89 ++++-
internal/scenario/run/sourcebytes_test.go | 84 +++++
internal/scenario/scenario.go | 62 ++++
internal/scenario/write_sd.go | 62 +++-
internal/verify/provider.go | 20 +-
internal/verify/provider_test.go | 138 ++++++++
internal/verify/session.go | 7 +
internal/verify/verify.go | 21 ++
internal/view/changes.go | 11 +-
internal/view/fill_test.go | 50 +++
internal/view/step.go | 8 +
pkg/changes/changes.go | 39 +++
pkg/model/model.go | 4 +
pkg/model/term.go | 2 +-
pkg/providerbody/session.go | 147 ++++++--
pkg/providerbody/toolarge_test.go | 253 ++++++++++++++
pkg/sessiondata/fit.go | 247 ++++++++++++++
pkg/sessiondata/fit_test.go | 369 +++++++++++++++++++++
pkg/sessiondata/sessiondata.go | 6 +-
pkg/sessiondata/writer.go | 29 +-
pkg/sessionview/sessionview.go | 6 +
tests/langchain_reset_test.go | 18 +-
tests/scenarios/too-large.expect.yaml | 30 ++
tests/scenarios/too-large.yaml | 62 ++++
56 files changed, 2690 insertions(+), 113 deletions(-)
diff --git a/CLAUDE.md b/CLAUDE.md
index 3dea773..659e885 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -139,7 +139,7 @@ Every source file carries the Apache-2.0 header; `make license-fix` inserts miss
- **Entity ids come from stable evidence, never from position.** An id that shifts when an earlier
source is backfilled cannot supersede its own earlier revision.
- **A round is cut at a byte budget** by narrowing its input window. A round covering one landed
- file may exceed the budget, as a landed file holding one oversize record may. Revisions that need
+ file may exceed the budget. Revisions that need
no new evidence, as after an upgrade, go first in rounds over no new file, cut by their frames.
`Round.More` says the chain has not reached the index yet, or has revisions left.
- **A round carries no wall-clock time**, so the same evidence reproduces the same bytes and digest.
@@ -176,6 +176,12 @@ Every source file carries the Apache-2.0 header; `make license-fix` inserts miss
- **Payload bytes are the source bytes** where a part carries the source's own JSON: byte for
byte, apart from whitespace between tokens. `json.RawMessage` both directions, and the writer
never re-encodes a part's data. Older files hold `\u003c`, `\u003e` and `\u0026` for the same values.
+- **A landed file fits its budget.** A collector cuts a file by what its records take once
+ written (`sessiondata.Cut`), never by source bytes: an envelope made 4 KiB of short lines 19 KB.
+ A single record over `max_delta_bytes` lands alone with its largest payloads taken out as
+ `too_large`, size and SHA-256 kept, through the writer's `Budget`. A change or execution record
+ is never taken out whole, and the provider cutter's own `Budget` never takes out a manifest. A
+ collector without a cut and a budget is a bug. `asz repack` carries records whole.
- **Land before committing the cursor.** At-least-once is intentional; the reverse loses data.
- **`next_seq` is monotonic across all streams in a session**, so a session is processed
single-threaded and under a lock. The filesystem is the authority on the counter.
diff --git a/asz.yaml b/asz.yaml
index 08e40c4..3ee031c 100644
--- a/asz.yaml
+++ b/asz.yaml
@@ -71,11 +71,12 @@ adapters:
interval: 10m
# The largest landed file the collector writes, in bytes. A large
- # catch-up is split into several files rather than one enormous one,
- # and a single record larger than this is landed whole. A file travels
- # whole as one log record, so this is also the largest record a
- # receiver has to accept. A change applies to new files only: asz
- # repack re-cuts an existing root into a new one.
+ # catch-up is split into several files rather than one enormous one.
+ # A single record larger than this lands alone, with its largest
+ # payloads omitted until it fits: each keeps its size and SHA-256 and
+ # says too_large. A file travels whole as one log record, so this is
+ # also the largest record a receiver has to accept. A change applies
+ # to new files only: asz repack re-cuts an existing root into a new one.
max_delta_bytes: 2097152
# The workspace change records the asz Claude Code plugin writes: which
@@ -145,7 +146,8 @@ adapters:
# The largest landed file a conversion writes. One request can carry
# more than this: a single turn with a 20 KB command and a 32 KB result
# measured 3 MB, so a request is cut across files rather than landing as
- # one. A record larger than the budget lands alone in a file of its own.
+ # one. A record larger than the budget lands alone in a file of its own,
+ # with its largest payloads omitted until it fits.
max_delta_bytes: 2097152
- name: changes
diff --git a/cmd/asz/main.go b/cmd/asz/main.go
index 52b433f..810abb3 100644
--- a/cmd/asz/main.go
+++ b/cmd/asz/main.go
@@ -539,7 +539,7 @@ func cmdVerify(cfg *config.Config, _ config.Adapter, _ bool) error {
return err
}
- var sessions, streams, records, problems, relanded int
+ var sessions, streams, records, problems, relanded, tooLarge, unbuilt int
for _, d := range entries {
if !d.IsDir() || (want != "" && d.Name() != want) {
continue
@@ -556,6 +556,8 @@ func cmdVerify(cfg *config.Config, _ config.Adapter, _ bool) error {
records += rep.Records
problems += rep.Problems
relanded += rep.Relanded
+ tooLarge += rep.TooLarge
+ unbuilt += rep.Unbuilt
if !rep.OK() {
fmt.Printf("%s: %d problem(s)\n", rep.Session, rep.Problems)
for _, line := range rep.Details() {
@@ -570,6 +572,15 @@ func cmdVerify(cfg *config.Config, _ config.Adapter, _ bool) error {
// the cursor is committed, so a crash repeats rather than loses.
fmt.Printf("%d record(s) landed more than once by an interrupted pass; assembly removes them\n", relanded)
}
+ if tooLarge > 0 {
+ // Omitted on purpose, so a landed file fits its byte budget. The
+ // part keeps its size and digest.
+ fmt.Printf("%d part(s) were too large to land and hold their size and digest only", tooLarge)
+ if unbuilt > 0 {
+ fmt.Printf("; %d provider body(ies) cannot be rebuilt for it", unbuilt)
+ }
+ fmt.Println()
+ }
// A conversation's rounds are checked too. They are a digest chain, so a
// missing or edited round is detectable in a way landed files are not, and
diff --git a/docs/en/adapters/claude-code-glossary.md b/docs/en/adapters/claude-code-glossary.md
index 3202c37..ad85602 100644
--- a/docs/en/adapters/claude-code-glossary.md
+++ b/docs/en/adapters/claude-code-glossary.md
@@ -97,6 +97,7 @@ than a plausible word that would send a reader looking for a field that does not
| `size_only` | only the length was kept |
| `strong_inference` | no identifier, but a well-founded structural argument |
| `talk` | one readable interaction; derived from where a run was triggered |
+| `too_large` | no equivalent: asz omits a payload whose record alone is larger than a landed file's byte budget, such as a pasted screenshot |
| `truncated` | part of the bytes, with the full length reported |
| `unavailable` | the runtime never wrote it, or the source was pruned |
| `unknown` | no record of the run states a trigger |
diff --git a/docs/en/adapters/kiro-glossary.md b/docs/en/adapters/kiro-glossary.md
index 0cc1688..ecbd945 100644
--- a/docs/en/adapters/kiro-glossary.md
+++ b/docs/en/adapters/kiro-glossary.md
@@ -98,6 +98,7 @@ not there.
| `strong_inference` | — | derived here | derived |
| `summarizes` | — | derived here | not read: nothing joins a summary to its tombstone |
| `talk` | — | derived here | one readable interaction: a prompt and the turn it opened |
+| `too_large` | — | derived here | no equivalent: asz omits a payload whose record alone is larger than a landed file's byte budget |
| `trigger` | — | derived here | derived: a user record that is neither a steering note nor written by the engine |
| `truncated` | — | derived here | 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 |
| `unknown` | — | derived here | derived |
diff --git a/docs/en/adapters/langsmith-glossary.md b/docs/en/adapters/langsmith-glossary.md
index c43b2b1..8ecb26d 100644
--- a/docs/en/adapters/langsmith-glossary.md
+++ b/docs/en/adapters/langsmith-glossary.md
@@ -99,6 +99,7 @@ not there.
| `summarizes` | — | derived here | derived: the summary and its boundary come from one marked message |
| `talk` | — | derived here | one readable interaction; one trace is one turn |
| `thinking` | — | derived here | no equivalent yet: a provider's reasoning arrives inside the message content |
+| `too_large` | — | derived here | no equivalent: asz omits a payload whose record alone is larger than a landed file's byte budget |
| `trigger` | — | derived here | no equivalent: derived from a trace's root run |
| `truncated` | — | derived here | no equivalent: nothing measured arrived truncated |
| `turn.duration` | — | derived here | no equivalent as a record; a run's own start and end give it |
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index 7c44037..27a5b17 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -191,6 +191,29 @@
stopped the session's pass at the failed file, before anything was saved, and they land in the same
order now. Records enter the index only once their file has landed, and a file that did not land
gives its number back.
+- Every landed file fits `max_delta_bytes`. A collector cut a file by the bytes it read, and a
+ record is larger than its source line by its envelope: 4 KiB of short lines landed as 19 KB. Files
+ are now cut by what their records take once written. A single record larger than the budget lands
+ alone, and its largest payloads are taken out, one at a time, until the file fits. A part taken
+ out keeps its place, its kind and its identifiers, and says `too_large`, with the size of the
+ original and the SHA-256 of what it carried in the new part field `sha256`. A part loses its data
+ before its text, so a result whose runtime record of a written file is large keeps its readable
+ text, and says `truncated`. A change record or an
+ execution record is never taken out whole, since a step joins to it: a change record is made
+ smaller in its own format, its largest files losing their hunks and then its file list going, so
+ it keeps its identity and says `truncated`, as its entry in the document now says with `state`. A pasted screenshot measured 4.6 MB in
+ one transcript line, and a file that size is more than a receiver may have to store at once. A
+ provider body is fitted by its cutter, which never takes out the manifest: a later body names a
+ piece taken out as it names any piece, a body of short strings with nothing to take out lands
+ whole and is taken out whole, and such a body cannot be rebuilt from the root. In the document, a
+ step that stands on a whole record says `too_large` when a part of it was taken out. `asz verify`
+ counts both apart from damage, and still holds a body's record to its manifest. See
+ [Too large to land](../formats/session-data.md#too-large-to-land).
+- A source line larger than the budget lands in a file of its own. The Claude Code, Kiro and
+ workspace change collectors read on to the end of the source to find where the long line ended,
+ and landed every line up to the last one in the same file: one 4.6 MB line made an 8 MB file. The
+ window now grows until it holds the line's end, and stops growing when the source was cut short
+ while it was read.
## Configuration
@@ -239,6 +262,19 @@
- golangci-lint is v2.14.0. Go 1.27.2 writes export data version 5, which v2.13.2 cannot read, so
under it lint stopped at the first package with a type-check error. CI takes the newest Go 1.27
release, so CI failed as soon as 1.27.2 came out.
+- A scenario's tool result and queued message can be padded, `result: {text, pad}` and
+ `queued: {text, pad}`, to carry more than a landed file's budget without spelling it out. An
+ expectation counts the document's steps by content state, `view: {states}`, and its change records
+ by theirs, `view: {change_states}`. A new property,
+ `files_fit_the_budget`, holds every landed file of every scenario to the budget.
+ `parts_keep_source_bytes` checks that a part taken out names the digest of a value of its source
+ record, and rebuilds a provider body that took a piece out from the file the runtime wrote, and
+ `records_match` compares the size and digest of a part taken out or shrunk across the two formats.
+ The `too-large` scenario returns a 40,000-byte result, queues a 30,000-byte message, and writes two
+ 30,000-byte files, one through a shell command the plugin observes and one through the runtime's
+ own Write tool, under a 16 KiB budget. The scenario's Session Data writer now gives an editing
+ tool's result the runtime's own record of it as data, as the adapter lands it; it gave such a
+ result the shell's shape before.
## Conversation model
diff --git a/docs/en/formats/asz-view.md b/docs/en/formats/asz-view.md
index c1b33f5..5687180 100644
--- a/docs/en/formats/asz-view.md
+++ b/docs/en/formats/asz-view.md
@@ -127,7 +127,7 @@ say.
| `id`, `kind`, `parent`, `stream`, `attrs` | the node as the fold holds it; `kind` is one of the node kinds of Session Flow; `attrs` is as the round wrote it, numbers included, without `provider_bodies`, which a call gives under a key of its own |
| `at` | when its record happened, from the record; `0` when nothing observed it |
| `ref`, `refs` | the record it stands on and every record it covers, as `{seq, row, block}`, kept so a viewer can show the evidence |
-| `text`, `state`, `bytes` | the part the node stands on: its readable text, clipped to the longest prefix of whole characters within 2,000 bytes, whether the content is `available`, and its full size. For a `data` part the text is the record's readable text when it has one: its `text` parts, or the prompt inside a `queued_command` envelope, the form a message typed while the agent works arrives in. Otherwise it is the data exactly as the record holds it, which is compact JSON. Earlier w [...]
+| `text`, `state`, `bytes` | the part the node stands on: its readable text, clipped to the longest prefix of whole characters within 2,000 bytes, the part's state, `available` or why not, such as `too_large` for a payload [too large to land](session-data.md#too-large-to-land), and its full size. For a `data` part the text is the record's readable text when it has one: its `text` parts, or the prompt inside a `queued_command` envelope, the form a message typed while the agent works arriv [...]
| `usage`, `flags`, `dropped` | what else the referenced record says, copied once: on an `llm.call`, the token counts `in`, `out`, `cache_read`, `cache_write` from the one record `usage_at` names, never a sum over fragments, a count of zero left out; the record's `flags`, which on a model call are only those a reader may [withhold](#withholding); and its `dropped` list, each entry its `what` and `bytes`, and `why` when there is one, so a viewer can say what was left out and why. A field [...]
| a talk adds | `label` and `reply`, clipped the same way and described below, then `runs`, `steps`, `tools`, `from`, `to`, `child`, `segment` |
| a tool or agent call adds | `name`, `failed`, `result`, `result_state`, `result_bytes`, `request_to_result_ms` and `request_to_result_join`, the time from the request record to the result record where the assembler joined them exactly; `changes` and `executions`, the records joined to it |
@@ -183,6 +183,7 @@ record's own fields as `changes/1` lists them:
| `ref` | the source: the landed record the entry was built from, in the shape every node's `ref` has. `seq` names a file under `files`, `row` the record in it counting from 1, `block` the part. A patch the runtime recorded sits on the `Edit` or `Write` result record, beside the raw result; a record the plugin wrote sits in a `changes` file, one record per line |
| `schema`, `id`, `captured_by`, `session`, `stream`, `tool`, `tool_name`, `time`, `basis` | the record's identity. `id` is the tool-use id the record belongs to, unique per call, or `gap/<root>/<step>` for a change no tool window covers. `captured_by` is `claude-code` for a patch the runtime recorded on its own editing tool, `asz-plugin` for one the asz Claude Code plugin observed. `basis` is `runtime_reported`, `tool_window`, `unattributed` or `skipped_read_only` |
| `root`, `policy`, `window`, `outcome`, `coverage`, `gaps`, `overlaps` | where it was observed, under which rules, between which scans, what the call reported, whether the whole scope was seen, and which other windows were open on the root at the same time |
+| `state` | `truncated` when the record was made smaller to fit its landed file, its hunks or its file list taken out ([Too large to land](session-data.md#too-large-to-land)), and absent when it landed whole. A record whose file list was taken out keeps `changed_files`, so an empty `changes` with a positive count means files changed, not none |
| `changed_files`, `changes` | the count, null when unknown, and one entry per file: `path`, `operation`, `before` and `after` as `present`, `bytes`, `sha256`, `no_newline_at_end`; `diff`, one of `available`, `binary`, `too_large`, `unavailable`; `attribution` and `windows`, which windows could have made the change; `additions`, `deletions` and `hunks`, each hunk `old_start`, `old_lines`, `new_start`, `new_lines` and `lines` prefixed with `-`, `+` or a space |
A tool step lists the ids of its records under `changes`, in the order `workspace_changes` lists
diff --git a/docs/en/formats/otlp.md b/docs/en/formats/otlp.md
index 642eeef..c816706 100644
--- a/docs/en/formats/otlp.md
+++ b/docs/en/formats/otlp.md
@@ -260,9 +260,10 @@ A request carries at most `export.otlp.batch_bytes` of file bytes, 8 MiB by defa
request under the 10 MiB the OAP's HTTP server accepts. A file larger than that is sent alone, in
a request of its own. A landed file is cut at
`max_delta_bytes`, 2 MiB by default, and a round is cut at `parse.max_round_bytes`, also
-2 MiB, so a request normally carries several files. The exception on both sides is a single
-unit larger than the budget: a source record is landed whole, and a round covering one landed
-file is published whole. The largest source record in the measured corpus is 4.5 MB.
+2 MiB, so a request normally carries several files. A source record larger than the budget lands
+alone, with its largest payloads omitted until the file fits: the largest source record in the
+measured corpus is a 4.6 MB pasted screenshot. The exception is a round covering one landed file,
+which is published whole.
The receiver's limit must cover the largest single request. The OAP accepts 50 MB over gRPC and
10 MiB over HTTP by default. An OpenTelemetry Collector accepts 4 MiB over gRPC and 20 MiB over
diff --git a/docs/en/formats/session-data.md b/docs/en/formats/session-data.md
index d90ad3e..f95a9f9 100644
--- a/docs/en/formats/session-data.md
+++ b/docs/en/formats/session-data.md
@@ -270,6 +270,22 @@ reads it back and rebuilds the body. A body that does not come back whole, or is
in valid UTF-8, lands whole in one `unknown` part, whose text says why it was not cut. `asz verify` rebuilds every landed
body and compares it with `sha256`, and reports a reference to something the session does not hold.
+A record whose new pieces are larger than the collector's `max_delta_bytes` has its largest pieces
+taken out until it fits, as [Too large to land](#too-large-to-land) describes for any record, and
+never its manifest. This is the first request after a person pasted a screenshot. A piece taken out
+keeps its place, so every segment still names it, and its `sha256` is the piece's digest, so a later
+body that holds the same piece names it as any earlier piece. A body made of short strings keeps
+nearly all of itself in the manifest's literals, so when the record is still too large with every
+piece taken out, the body lands whole in one `unknown` part, and that part is taken out.
+
+A body with a piece taken out cannot be rebuilt from the root: `Session.Body` returns `ErrTooLarge`,
+and `Session.BodyFrom` rebuilds it from another copy of the body, such as the file the runtime wrote.
+A reader registers a piece taken out by its `sha256`, as any piece, so a later body that names it is
+read, and one that names only other pieces still rebuilds. The adapter's check before landing
+rebuilds the record with the pieces it took out. No body copies the front of a body that cannot be
+rebuilt. `asz verify` counts these bodies apart from damage, and still holds the record's `bytes`
+and `sha` to its manifest.
+
On two captures of Claude Code 2.1.260, the landed `provider_body` files were 25.5% and 21.6% of the
bytes Claude Code wrote, and all 42 bodies rebuilt exactly.
@@ -299,8 +315,8 @@ reads every landed file this way.
| `data` | structure that is not prose: a record the runtime keeps for itself, a manifest | `data` |
| `unknown` | content the dialect could not describe | the bytes in `data` as one JSON string, the bytes themselves or their base64 when `encoding` is `base64`, and the reason in `text` |
-In a part, `k`, `text`, `id`, `name`, `of`, `server`, `server_tool`, `media`, `encoding` and
-`state` are strings, `bytes` is an integer, `failed` is a boolean, and `data` is any JSON value.
+In a part, `k`, `text`, `id`, `name`, `of`, `server`, `server_tool`, `media`, `encoding`, `state`
+and `sha256` are strings, `bytes` is an integer, `failed` is a boolean, and `data` is any JSON value.
`server` and `server_tool` say which MCP server a call is addressed to and which of its tools. An
adapter sets them only where its runtime's name for the call splits exactly. Claude Code names such a
@@ -311,8 +327,41 @@ form of the configured name, so "claude.ai Claude Docs" is `claude_ai_Claude_Doc
server was configured with is in the call's execution record, when there is one.
Every part carries `state` and `bytes`. `state` is one of `available`, `truncated`, `redacted`,
-`omitted` or `unavailable`, and `bytes` is the size of the original even when the part holds less.
-A reader is always told how much of the original it has.
+`omitted`, `unavailable` or `too_large`, and `bytes` is the size of the original even when the part
+holds less. A reader is always told how much of the original it has.
+
+### Too large to land
+
+A landed file travels whole, so the collector's `max_delta_bytes` is the most a receiver has to store
+at once. A collector cuts a file by what its records take once written, not by the bytes they were
+read from: a record carries an envelope its source line does not, and 4 KiB of short source lines
+measured 19 KB of records. One record cannot be cut: a pasted screenshot measured 4.6 MB in one
+transcript line. Such a record lands alone in a file of its own, and its largest payloads are taken
+out, one at a time, until the file fits.
+
+A part taken out keeps its place, its kind and every identifier: `id`, `name`, `of`, `server`,
+`server_tool`, `failed` and `media`. Its `data` and `encoding` go first, and then its `text`, except
+an `unknown` part's, which is the reason its bytes were not described rather than content. A part
+with text and data, such as a result with the runtime's structured copy of it, so keeps its readable
+text while it can: its `state` is `truncated`. A part with nothing left says `too_large`. Its
+`bytes` is the size of the original, and its `sha256` is the SHA-256 of what was taken out first:
+an `unknown` part's bytes, other data compacted as the writer lands it, or the text of a part with
+no data. A reader still sees the step, and can match the payload against another copy of it.
+
+A `changes/1` record and an `execution/1` record are evidence a step joins to by its tool call, so
+neither is ever taken out whole: a reader that found nothing would read that the call changed no file
+or ran nowhere. A change record is made smaller in its own format instead. Its file with the largest
+hunks loses them and says its `diff` is `too_large`, as a diff too large to compute says, and its
+additions and deletions stay. This goes on, file by file, until the record fits. When no file has
+hunks left, its files, overlaps and gaps go, and it keeps its identity, its tool, its window and
+`changed_files`. Its part says `truncated`, with the original's `bytes` and `sha256`, and the record
+still joins its step. An execution record keeps sizes and digests only, never what was sent or
+returned, so it is a few hundred bytes and is left as it is.
+
+Of two payloads of the same size, the one in the earlier part goes first, so the same record always
+loses the same parts. No other part carries `sha256`. A record that is still too large with every
+payload taken out lands as it is: what is left is its identifiers, its label and the list of its
+parts. `asz repack` carries records as they landed, and takes nothing out.
### What data holds
@@ -360,7 +409,8 @@ The measured corpus is 62 Claude Code sessions: 3,032 files and 1,100.1 MB of so
tool result in it was 1.1 MB. Over its 28 sessions larger than 1 MB, the share of a session that is
tool output ran from 17.3% to 83.5%, with a median of 35.1%. A rule tuned to the median is wrong by
a factor of two at both ends. So every part states its own size and state, and a reader decides.
-No built-in adapter writes `truncated` or `omitted`, so no part they write is clipped. Clipping is
+No built-in adapter writes `omitted`, and a part one writes as `truncated` is one made smaller to fit
+the budget, so no part they write is clipped otherwise. Clipping is
left to a reader, such as the [asz.view](asz-view.md#a-node-in-talks) document, which clips a
node's text and gives the full size beside it. A part is still not always all the runtime gave.
The Claude Code dialect leaves some content out with no `dropped` entry. Its
diff --git a/docs/en/guides/command-line-internals.md b/docs/en/guides/command-line-internals.md
index fcba489..16bc66d 100644
--- a/docs/en/guides/command-line-internals.md
+++ b/docs/en/guides/command-line-internals.md
@@ -398,3 +398,11 @@ A record that repeats a range already landed is counted, not reported as a probl
`N record(s) landed more than once by an interrupted pass; assembly removes them`. A file lands
before its cursor is committed, so an interrupted pass repeats data rather than losing it.
[Storage Root](../formats/storage-root.md#landed-files) explains the order.
+
+A part whose payload was [too large to land](../formats/session-data.md#too-large-to-land) is
+counted the same way, with the provider bodies that cannot be rebuilt because of it: verify prints
+`N part(s) were too large to land and hold their size and digest only; M provider body(ies) cannot
+be rebuilt for it`. The collector left the payload out on purpose, so the file fits its budget. A
+body that holds or refers to such a piece is not damage: its references are still checked, and its
+record's `bytes` and `sha` are still held to its manifest. Only the comparison of its rebuilt bytes
+with its digest cannot be made.
diff --git a/docs/en/setup/command-line.md b/docs/en/setup/command-line.md
index 2b9bbda..7e9f7b5 100644
--- a/docs/en/setup/command-line.md
+++ b/docs/en/setup/command-line.md
@@ -95,11 +95,14 @@ checked 44 conversation chain(s), 44 round(s)
all landed data is contiguous and matches its digests
```
+It also counts the parts that were too large to land, which are not a problem.
+
## repack
Copies the storage root, or one session, into the new or empty directory `DEST`, with no file
larger than `max_delta_bytes`, and assembles the conversations there. Use it after changing
-`max_delta_bytes`.
+`max_delta_bytes`. A record larger than the new budget is carried whole, in a file of its own:
+repack never leaves anything out.
## index
diff --git a/docs/en/setup/configuration.md b/docs/en/setup/configuration.md
index 828a15a..586ac7f 100644
--- a/docs/en/setup/configuration.md
+++ b/docs/en/setup/configuration.md
@@ -75,7 +75,7 @@ Set on each adapter.
| --- | --- | --- |
| `mode` | `watch` | `watch` collects again every `interval`. `once` collects once; `asz collect` then exits. `-once` on the command line does the same. |
| `interval` | `10m` | The time between collections. `asz collect` and `asz server` also collect when they start. |
-| `max_delta_bytes` | `2097152` | The largest file asz writes from one source, 2 MiB. A single longer record gets a file of its own. |
+| `max_delta_bytes` | `2097152` | The largest file asz writes from one source, 2 MiB. A single longer record gets a file of its own, and its largest payloads are left out until it fits: see [Too large to land](../formats/session-data.md#too-large-to-land). |
When several adapters are enabled, asz collects from all of them together, at the shortest
`interval` among them, and watches if any of them has `mode: watch`.
diff --git a/internal/adapters/claudecode/collect.go b/internal/adapters/claudecode/collect.go
index 60910f0..e286f00 100644
--- a/internal/adapters/claudecode/collect.go
+++ b/internal/adapters/claudecode/collect.go
@@ -399,6 +399,30 @@ func (c *Collector) landAppend(src Source, cur *storage.Cursor, cursorPath, dir,
Src: src.Rel, Session: src.Session, Stream: src.Stream, Batch: src.RunID,
}
+ // Conversion happens HERE, once, while the source is being read. Nothing
+ // above this line ever sees a Claude Code shape. A record is larger than
+ // its line, so the file ends where the converted records reach the
+ // budget, and the cursor stops before the first line left out.
+ cut, err := sessiondata.NewCut(hdr, c.MaxDelta)
+ if err != nil {
+ state.GiveBack(seq)
+ return false, false, err
+ }
+ var recs []*sessiondata.Record
+ for _, ln := range chunk.Lines {
+ rec := Convert(src, ln.Ord, ln.Off, ln.Bytes)
+ ok, err := cut.Take(rec)
+ if err != nil {
+ state.GiveBack(seq)
+ return false, false, err
+ }
+ if !ok {
+ break
+ }
+ recs = append(recs, rec)
+ }
+ chunk.Keep(len(recs), cur.LastUUID)
+
var written int64
var staged index.Staged
err = storage.WriteAtomic(filepath.Join(dir, name), storage.PermLanded, func(w io.Writer) error {
@@ -406,14 +430,12 @@ func (c *Collector) landAppend(src Source, cur *storage.Cursor, cursorPath, dir,
if err != nil {
return err
}
- for row, ln := range chunk.Lines {
- // Conversion happens HERE, once, while the source is being read.
- // Nothing above this line ever sees a Claude Code shape.
- rec := Convert(src, ln.Ord, ln.Off, ln.Bytes)
+ rw.Budget(c.MaxDelta)
+ for row, rec := range recs {
if err := rw.Write(rec); err != nil {
return err
}
- written += int64(len(ln.Bytes))
+ written += int64(len(chunk.Lines[row].Bytes))
// The index is built from the CONVERTED record, so the source is
// parsed once and the index has no dialect in it.
staged.Add(index.FromRecord(p.ix, hdr, rec, uint32(seq), uint32(row+1)))
@@ -485,6 +507,7 @@ func (c *Collector) landSnapshot(src Source, cur *storage.Cursor, cursorPath, di
if err != nil {
return err
}
+ rw.Budget(c.MaxDelta)
rec := Convert(src, 1, 0, body)
if err := rw.Write(rec); err != nil {
return err
diff --git a/internal/adapters/claudecode/glossary.go b/internal/adapters/claudecode/glossary.go
index 8e1000c..0adb194 100644
--- a/internal/adapters/claudecode/glossary.go
+++ b/internal/adapters/claudecode/glossary.go
@@ -131,6 +131,7 @@ var glossary = model.NewGlossary(Dialect,
model.Term{Unified: model.ContentHashOnly, Note: "only a digest of the bytes was kept"},
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.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/claudecode/regress_test.go b/internal/adapters/claudecode/regress_test.go
index 122f5d7..095bdd8 100644
--- a/internal/adapters/claudecode/regress_test.go
+++ b/internal/adapters/claudecode/regress_test.go
@@ -18,6 +18,9 @@
package claudecode_test
import (
+ "bytes"
+ "crypto/sha256"
+ "encoding/hex"
"encoding/json"
"fmt"
"os"
@@ -31,6 +34,7 @@ import (
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecode"
"github.com/apache/skywalking-ai-sessionizer/internal/index"
"github.com/apache/skywalking-ai-sessionizer/internal/storage"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/model"
"github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
)
@@ -177,6 +181,146 @@ func TestOversizedLineDoesNotStall(t *testing.T) {
}
}
+// TestOversizedLineLandsAlone covers what happens after the window grows past
+// the budget: the long line lands in a file of its own, the lines after it in
+// later files, and the record fits the budget with its payload omitted. The
+// window used to grow to the end of the source and land every line up to the
+// last newline in one file: a 4.6 MB line made an 8 MB file.
+func TestOversizedLineLandsAlone(t *testing.T) {
+ src, zone := t.TempDir(), t.TempDir()
+ small := func(id string) string { return fmt.Sprintf("{\"uuid\":%q,\"pad\":%q}\n", id, strings.Repeat("s", 300)) }
+ big := fmt.Sprintf("{\"uuid\":\"big\",\"type\":\"user\",\"message\":{\"role\":\"user\",\"content\":%q}}\n", strings.Repeat("y", 50_000))
+ mk(t, filepath.Join(src, "-proj-a", tSess+".jsonl"), small("before")+big+small("a")+small("b")+small("c"))
+
+ const budget = 4096
+ z := storage.NewZone(zone)
+ st, err := claudecode.New(src, z, budget).CollectAll(nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if st.Records != 5 {
+ t.Fatalf("landed %d records, want 5", st.Records)
+ }
+ files, err := storage.LandedFiles(z, tSess)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, lf := range files {
+ data, err := os.ReadFile(lf.Path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ _, recs, err := sessiondata.All(bytes.NewReader(data))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(data) > budget {
+ t.Errorf("%s is %d bytes, over the budget of %d", filepath.Base(lf.Path), len(data), budget)
+ }
+ for _, rec := range recs {
+ if rec.ID != "big" {
+ continue
+ }
+ if len(recs) != 1 {
+ t.Errorf("the long line shares %s with %d other records", filepath.Base(lf.Path), len(recs)-1)
+ }
+ p := rec.Parts[0]
+ if p.State != model.ContentTooLarge || p.Text != "" || p.Bytes != 50_000 || p.SHA256 != sha256Hex(strings.Repeat("y", 50_000)) {
+ t.Errorf("the long line's part is %s with %d bytes of text, size %d and digest %q; want too_large with no text, the size and the text's digest",
+ p.State, len(p.Text), p.Bytes, p.SHA256)
+ }
+ }
+ }
+}
+
+// TestFilesAreCutByWhatTheRecordsTake covers lines that are small at the
+// source and larger once converted: each record adds its envelope, so a
+// budget's worth of short lines makes a file over the budget. The collector
+// cuts by the converted records, lands every line, and its cursor ends where
+// the source does.
+func TestFilesAreCutByWhatTheRecordsTake(t *testing.T) {
+ src, zone := t.TempDir(), t.TempDir()
+ var b strings.Builder
+ for i := 0; i < 400; i++ {
+ fmt.Fprintf(&b, "{\"uuid\":\"u%03d\",\"type\":\"system\",\"content\":\"x\"}\n", i)
+ }
+ mk(t, filepath.Join(src, "-proj-a", tSess+".jsonl"), b.String())
+ const budget = 4096
+ z := storage.NewZone(zone)
+ st, err := claudecode.New(src, z, budget).CollectAll(nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if st.Records != 400 {
+ t.Fatalf("landed %d records, want 400", st.Records)
+ }
+ files, err := storage.LandedFiles(z, tSess)
+ if err != nil {
+ t.Fatal(err)
+ }
+ ords := 0
+ for _, lf := range files {
+ data, err := os.ReadFile(lf.Path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(data) > budget {
+ t.Errorf("%s is %d bytes, over the budget of %d", filepath.Base(lf.Path), len(data), budget)
+ }
+ _, recs, err := sessiondata.All(bytes.NewReader(data))
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, rec := range recs {
+ ords++
+ if rec.Ord != uint64(ords) {
+ t.Fatalf("record %d has ord %d: a line was skipped or repeated at a cut", ords, rec.Ord)
+ }
+ }
+ }
+ if len(files) < 3 {
+ t.Fatalf("400 records went into %d file(s)", len(files))
+ }
+}
+
+func sha256Hex(s string) string {
+ sum := sha256.Sum256([]byte(s))
+ return hex.EncodeToString(sum[:])
+}
+
+// TestTailTakesOneLongLine is the tail half of the same rule: past the
+// budget, one line is taken, the rest is left for the next read, and a long
+// line still being written is not taken at all.
+func TestTailTakesOneLongLine(t *testing.T) {
+ p := filepath.Join(t.TempDir(), "s.jsonl")
+ long := strings.Repeat("x", 5000)
+ writeFile(t, p, long+"\nafter one\nafter two\n"+long)
+ cur := storage.NewCursor(storage.CursorAppend, "s.jsonl")
+ ch, err := claudecode.TailAppend(p, cur, 64)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(ch.Lines) != 1 || len(ch.Lines[0].Bytes) != 5000 || !ch.More || ch.NewOffset != 5001 {
+ t.Fatalf("took %d line(s) to offset %d, more %v; want the one long line to 5001 with more to come", len(ch.Lines), ch.NewOffset, ch.More)
+ }
+ cur.Offset, cur.Ord = ch.NewOffset, ch.NewOrd
+ ch, err = claudecode.TailAppend(p, cur, 64)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(ch.Lines) != 2 || string(ch.Lines[1].Bytes) != "after two" {
+ t.Fatalf("took %d line(s), want the two short ones", len(ch.Lines))
+ }
+ cur.Offset, cur.Ord = ch.NewOffset, ch.NewOrd
+ ch, err = claudecode.TailAppend(p, cur, 64)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(ch.Lines) != 0 || ch.NewOffset != cur.Offset {
+ t.Fatalf("took %d line(s) of a long line still being written", len(ch.Lines))
+ }
+}
+
// TestStubDirectoryDoesNotVetoSession is the regression for the discovery half
// of the exclusion bug: a session directory containing nothing collectable was
// still recorded as a source directory, letting it disqualify the whole session.
diff --git a/internal/adapters/claudecode/tail.go b/internal/adapters/claudecode/tail.go
index d5d532d..bb4d5a4 100644
--- a/internal/adapters/claudecode/tail.go
+++ b/internal/adapters/claudecode/tail.go
@@ -79,6 +79,24 @@ type Chunk struct {
Moved bool
}
+// Keep trims the chunk to its first n lines, as when a landed file is full
+// before the chunk is: reading stops before the first line left out, and
+// the caller drains again. prev is the uuid the cursor held before the chunk.
+func (ch *Chunk) Keep(n int, prev string) {
+ if n >= len(ch.Lines) {
+ return
+ }
+ left := ch.Lines[n]
+ ch.Lines = ch.Lines[:n]
+ ch.NewOffset, ch.NewOrd, ch.More = left.Off, left.Ord-1, true
+ ch.LastUUID = prev
+ for _, ln := range ch.Lines {
+ if u := extractUUID(ln.Bytes); u != "" {
+ ch.LastUUID = u
+ }
+ }
+}
+
// statInfo carries the identity and size checks a cursor is validated against.
// statSource fills it and lives in one file per platform, because a file's
// identity is an inode on Unix and a volume serial plus file index on Windows.
@@ -164,13 +182,13 @@ func TailAppend(path string, cur *storage.Cursor, maxBytes int64) (*Chunk, error
// The window is full and there is more file beyond it, so this single
// line is larger than the byte budget. Reading only the budget would
// stall this source forever, silently, with no error - so grow past the
- // budget to find the line's end. The budget bounds throughput, not the
- // maximum record we are willing to store.
- buf, err = readWindow(f, int64(cur.Offset), info.size, 0)
- if err != nil {
+ // budget to find the line's end, and take that one line alone. Taking
+ // every line up to the last newline instead landed all the rest of a
+ // source in the same file: one 4.6 MB line made an 8 MB file. The
+ // writer takes out what does not fit; see sessiondata.Fit.
+ if buf, end, err = longLine(f, int64(cur.Offset), info.size, maxBytes); err != nil {
return nil, err
}
- end = bytes.LastIndexByte(buf, '\n')
}
if end < 0 {
// Still no complete line: an in-flight write. Nothing safe to land yet.
@@ -201,6 +219,32 @@ func TailAppend(path string, cur *storage.Cursor, maxBytes int64) (*Chunk, error
return ch, nil
}
+// longLine reads the line starting at from that is longer than maxBytes, and
+// returns the window that holds it and where it ends, or -1 when the window
+// reached the end of the file first: a line still being written. The window
+// doubles each time. A read that brings nothing new means the file is shorter
+// than size, cut while it was read, and the line is left for the next pass,
+// which checks the file again.
+func longLine(f *os.File, from, size, maxBytes int64) ([]byte, int, error) {
+ var buf []byte
+ for n := 2 * maxBytes; ; n *= 2 {
+ next, err := readWindow(f, from, size, n)
+ if err != nil {
+ return nil, -1, err
+ }
+ if len(next) <= len(buf) {
+ return buf, -1, nil
+ }
+ buf = next
+ if end := bytes.IndexByte(buf, '\n'); end >= 0 {
+ return buf, end, nil
+ }
+ if from+int64(len(buf)) >= size || n <= 0 {
+ return buf, -1, nil
+ }
+ }
+}
+
// readWindow reads at most maxBytes starting at from.
func readWindow(f *os.File, from, size, maxBytes int64) ([]byte, error) {
want := size - from
diff --git a/internal/adapters/claudecode/tail_internal_test.go b/internal/adapters/claudecode/tail_internal_test.go
new file mode 100644
index 0000000..82d517f
--- /dev/null
+++ b/internal/adapters/claudecode/tail_internal_test.go
@@ -0,0 +1,109 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package claudecode
+
+import (
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+)
+
+// TestLongLineEndsWhenTheFileIsShorterThanItWas covers a source cut short
+// while its long line is read: the size taken before reading is larger than
+// the file, so the window can never reach it. The window must stop growing
+// when a read brings nothing new. Before, it doubled on, each read sized to
+// the stale size: a source cut from gigabytes to a few bytes made it allocate
+// gigabytes again and again. The size here is 64 GiB, so a window that keeps
+// growing fails to allocate.
+func TestLongLineEndsWhenTheFileIsShorterThanItWas(t *testing.T) {
+ p := filepath.Join(t.TempDir(), "s.jsonl")
+ if err := os.WriteFile(p, []byte(strings.Repeat("x", 300)), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ f, err := os.Open(p)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer f.Close()
+ buf, end, err := longLine(f, 0, 64<<30, 16)
+ if err != nil || end != -1 || len(buf) != 300 {
+ t.Fatalf("got %d bytes, end %d, %v; want the 300 bytes there are and no line end", len(buf), end, err)
+ }
+}
+
+func TestLongLineFindsTheEndOfTheFirstLine(t *testing.T) {
+ p := filepath.Join(t.TempDir(), "s.jsonl")
+ long := strings.Repeat("x", 1000)
+ body := long + "\n" + long + "\n" + "short\n"
+ if err := os.WriteFile(p, []byte(body), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ f, err := os.Open(p)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer f.Close()
+ buf, end, err := longLine(f, 0, int64(len(body)), 16)
+ if err != nil || end != 1000 || len(buf) < 1001 {
+ t.Fatalf("got %d bytes, end %d, %v; want the first line's end at 1000", len(buf), end, err)
+ }
+ // The same from the second long line, which runs to the end of a file
+ // still being written.
+ unfinished := long + "\n" + long
+ if err := os.WriteFile(p, []byte(unfinished), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ g, err := os.Open(p)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer g.Close()
+ if _, end, err := longLine(g, 1001, int64(len(unfinished)), 16); err != nil || end != -1 {
+ t.Fatalf("end %d, %v; want no line end for a line still being written", end, err)
+ }
+}
+
+// Keep trims a chunk to the lines that fit a file: reading stops before the
+// first line left out, the uuid is the last landed line's, and more is to
+// come. Keeping none leaves the uuid the cursor had.
+func TestKeepStopsBeforeTheFirstLineLeftOut(t *testing.T) {
+ mk := func() *Chunk {
+ ch := &Chunk{NewOffset: 300, NewOrd: 13, LastUUID: "u3"}
+ for i, u := range []string{"u1", "u2", "u3"} {
+ ch.Lines = append(ch.Lines, Line{Ord: uint64(11 + i), Off: uint64(100 * i), Bytes: []byte(`{"uuid":"` + u + `"}`)})
+ }
+ return ch
+ }
+ ch := mk()
+ ch.Keep(2, "u0")
+ if len(ch.Lines) != 2 || ch.NewOffset != 200 || ch.NewOrd != 12 || !ch.More || ch.LastUUID != "u2" {
+ t.Fatalf("kept %d lines to offset %d and line %d, more %v, uuid %q; want 2 to 200 and 12, more, u2",
+ len(ch.Lines), ch.NewOffset, ch.NewOrd, ch.More, ch.LastUUID)
+ }
+ ch = mk()
+ ch.Keep(0, "u0")
+ if len(ch.Lines) != 0 || ch.NewOffset != 0 || ch.NewOrd != 10 || ch.LastUUID != "u0" {
+ t.Fatalf("kept %d lines to offset %d and line %d, uuid %q; want none, the cursor's place and uuid", len(ch.Lines), ch.NewOffset, ch.NewOrd, ch.LastUUID)
+ }
+ ch = mk()
+ ch.Keep(3, "u0")
+ if ch.NewOffset != 300 || ch.More || ch.LastUUID != "u3" {
+ t.Fatal("keeping every line changed the chunk")
+ }
+}
diff --git a/internal/adapters/claudecodechanges/collect.go b/internal/adapters/claudecodechanges/collect.go
index 050147a..7de37d5 100644
--- a/internal/adapters/claudecodechanges/collect.go
+++ b/internal/adapters/claudecodechanges/collect.go
@@ -335,6 +335,28 @@ func (c *Collector) collectSource(src Source, ix *index.Index, state *storage.Se
Kind: src.Kind, Adapter: Name + "/" + Version, Dialect: Dialect,
Src: src.Rel, Session: src.Session, Stream: src.Stream,
}
+ // A record is larger than its line, so the file ends where the converted
+ // records reach the budget, and the cursor stops before the first line
+ // left out.
+ cut, err := sessiondata.NewCut(hdr, c.MaxDelta)
+ if err != nil {
+ state.GiveBack(seq)
+ return false, false, err
+ }
+ var recs []*sessiondata.Record
+ for _, ln := range chunk.Lines {
+ rec := ConvertKind(src.Kind, ln.Ord, ln.Off, ln.Bytes)
+ ok, err := cut.Take(rec)
+ if err != nil {
+ state.GiveBack(seq)
+ return false, false, err
+ }
+ if !ok {
+ break
+ }
+ recs = append(recs, rec)
+ }
+ chunk.Keep(len(recs), cur.LastUUID)
var written int64
var staged index.Staged
err = storage.WriteAtomic(filepath.Join(dir, name), storage.PermLanded, func(w io.Writer) error {
@@ -342,8 +364,9 @@ func (c *Collector) collectSource(src Source, ix *index.Index, state *storage.Se
if err != nil {
return err
}
+ rw.Budget(c.MaxDelta)
for row, ln := range chunk.Lines {
- rec := ConvertKind(src.Kind, ln.Ord, ln.Off, ln.Bytes)
+ rec := recs[row]
if err := rw.Write(rec); err != nil {
return err
}
diff --git a/internal/adapters/claudecodechanges/collect_test.go b/internal/adapters/claudecodechanges/collect_test.go
index f9267d3..33486ca 100644
--- a/internal/adapters/claudecodechanges/collect_test.go
+++ b/internal/adapters/claudecodechanges/collect_test.go
@@ -452,3 +452,69 @@ func TestALegacyCursorIsNotClaimedOnASharedTailAlone(t *testing.T) {
t.Fatalf("b landed %d records, want all %d of its own: a cursor read from a's file was claimed on the shared tail", records, want)
}
}
+
+// A change record larger than the budget lands as a change record: its
+// largest files lose their hunks and say their diff is too_large, and its
+// identity, files and counts stay, so it still joins its step and adds up.
+func TestALargeChangeRecordKeepsItsIdentity(t *testing.T) {
+ src, root := t.TempDir(), t.TempDir()
+ r := &changes.Record{
+ Schema: changes.Schema, ID: "p1/big", CapturedBy: changes.CapturedByASZPlugin, Session: session, Stream: "main",
+ Tool: "toolu_9", ToolName: "Bash", Time: "2026-09-08T02:00:04Z", Basis: changes.BasisToolWindow,
+ Root: &changes.Root{Path: "/workspace/project"}, ChangedFiles: changes.Int(2),
+ }
+ for _, path := range []string{"big.json", "small.go"} {
+ n := 2_000
+ if path == "small.go" {
+ n = 2
+ }
+ lines := make([]string, n)
+ for i := range lines {
+ lines[i] = "+" + strings.Repeat("y", 40)
+ }
+ r.Changes = append(r.Changes, changes.FileChange{Path: path, Operation: changes.OpModify,
+ Diff: changes.DiffAvailable, Additions: changes.Int(n), Deletions: changes.Int(0),
+ Hunks: []changes.Hunk{{NewStart: 1, NewLines: n, Lines: lines}}})
+ }
+ b, _ := r.Marshal()
+ out := filepath.Join(src, "asz-changes-inline", "output", session)
+ appendTo(t, filepath.Join(out, "main.jsonl"), string(b)+"\n")
+
+ const budget = 16 << 10
+ zone := storage.NewZone(root)
+ if _, err := claudecodechanges.New(src, zone, budget).CollectAll(nil); err != nil {
+ t.Fatal(err)
+ }
+ files, err := storage.LandedFiles(zone, session)
+ if err != nil || len(files) != 1 {
+ t.Fatalf("landed %d files, %v", len(files), err)
+ }
+ if fi, _ := os.Stat(files[0].Path); fi.Size() > budget {
+ t.Fatalf("the file is %d bytes, over the budget of %d", fi.Size(), budget)
+ }
+ _, recs := readAll(t, files[0].Path)
+ p := recs[0].Parts[0]
+ got, ok := changes.Decode(p.Data)
+ if !ok || p.State != "truncated" || p.Bytes != len(b) {
+ t.Fatalf("the part is %s and decodes %v, want a truncated change record of the original size", p.State, ok)
+ }
+ if got.ID != "p1/big" || got.Tool != "toolu_9" || len(got.Changes) != 2 ||
+ got.Changes[0].Diff != changes.DiffTooLarge || got.Changes[0].Hunks != nil || *got.Changes[0].Additions != 2_000 ||
+ got.Changes[1].Diff != changes.DiffAvailable || len(got.Changes[1].Hunks) != 1 {
+ t.Fatalf("the change record kept %+v", got.Changes)
+ }
+ // The index sees a change record, with its lines.
+ ix, ok, err := index.Load(zone.IndexDir(session), session)
+ if err != nil || !ok {
+ t.Fatalf("no index: %v", err)
+ }
+ adds := uint32(0)
+ for _, blk := range ix.Blocks {
+ if blk.Kind == index.BlockChanges {
+ adds += blk.Adds
+ }
+ }
+ if adds != 2_002 {
+ t.Fatalf("the index counts %d added lines, want 2002", adds)
+ }
+}
diff --git a/internal/adapters/claudecodeprovider/collect.go b/internal/adapters/claudecodeprovider/collect.go
index 62e721d..2606270 100644
--- a/internal/adapters/claudecodeprovider/collect.go
+++ b/internal/adapters/claudecodeprovider/collect.go
@@ -550,8 +550,8 @@ func (c *Collector) collectSession(session string, files []*file, sn *seen, st *
st.Errors = append(st.Errors, fmt.Errorf("claudecodeprovider: %s: %w", f.name, err))
continue
}
- // A file ends before a body that would take it past the budget, so
- // only a body larger than the budget on its own lands in a larger file.
+ // A file ends before a body that would take it past the budget, and
+ // the cutter fits a body larger than the budget on its own.
n := providerbody.RecordBytes(rec)
if len(batch) > 0 && providerbody.FileOverhead+size+n > c.MaxDelta {
if err := flush(); err != nil {
@@ -610,6 +610,7 @@ func (c *Collector) markLanded(sn *seen, f *file, session string) {
// takes new bodies. asz verify reports the damage itself.
func (c *Collector) loadSession(session string) (*providerbody.Session, error) {
held := providerbody.NewSession()
+ held.Budget(c.MaxDelta)
files, err := storage.LandedFiles(c.Zone, session)
if err != nil {
return nil, err
diff --git a/internal/adapters/kiro/collect.go b/internal/adapters/kiro/collect.go
index f3117cd..cf8a429 100644
--- a/internal/adapters/kiro/collect.go
+++ b/internal/adapters/kiro/collect.go
@@ -309,13 +309,16 @@ func (c *Collector) landAppend(src Source, cur *storage.Cursor, cursorPath, dir,
return false, false, err
}
lines, newOffset, newOrd, more := chunk.Lines, chunk.NewOffset, chunk.NewOrd, chunk.More
- if keep < len(lines) {
+ held := keep < len(lines)
+ if held {
// The rest waits for its turn. The cursor stops before it, so the
// next pass reads it again with whatever followed.
- st.Held++
lines, newOffset, newOrd, more = lines[:keep], lines[keep].Off, lines[keep].Ord-1, false
}
if len(lines) == 0 {
+ if held {
+ st.Held++
+ }
return false, false, nil
}
@@ -326,6 +329,36 @@ func (c *Collector) landAppend(src Source, cur *storage.Cursor, cursorPath, dir,
Kind: recordKind(src.Kind), Adapter: Name + "/" + Version, Dialect: Dialect,
Src: src.Rel, Session: src.Session, Stream: src.Stream,
}
+ // A record is larger than its line, so the file ends where the converted
+ // records reach the budget, and the cursor stops before the first line
+ // left out.
+ cut, err := sessiondata.NewCut(hdr, c.MaxDelta)
+ if err != nil {
+ p.state.GiveBack(seq)
+ return false, false, err
+ }
+ var recs []*sessiondata.Record
+ for row, ln := range lines {
+ rec := Convert(src, ln.Ord, ln.Off, ln.Bytes, runs[row])
+ ok, err := cut.Take(rec)
+ if err != nil {
+ p.state.GiveBack(seq)
+ return false, false, err
+ }
+ if !ok {
+ break
+ }
+ recs = append(recs, rec)
+ }
+ switch {
+ case len(recs) < len(lines):
+ newOffset, newOrd, more = lines[len(recs)].Off, lines[len(recs)].Ord-1, true
+ lines = lines[:len(recs)]
+ case held:
+ // Counted once, by the call that lands the last line before what is
+ // held, not by every call a cut makes on the way.
+ st.Held++
+ }
var written int64
var staged index.Staged
err = storage.WriteAtomic(filepath.Join(dir, name), storage.PermLanded, func(w io.Writer) error {
@@ -333,8 +366,9 @@ func (c *Collector) landAppend(src Source, cur *storage.Cursor, cursorPath, dir,
if err != nil {
return err
}
+ rw.Budget(c.MaxDelta)
for row, ln := range lines {
- rec := Convert(src, ln.Ord, ln.Off, ln.Bytes, runs[row])
+ rec := recs[row]
if err := rw.Write(rec); err != nil {
return err
}
@@ -402,6 +436,7 @@ func (c *Collector) landSnapshot(src Source, cur *storage.Cursor, cursorPath, di
if err != nil {
return err
}
+ rw.Budget(c.MaxDelta)
rec := Convert(src, 1, 0, body, "")
if err := rw.Write(rec); err != nil {
return err
diff --git a/internal/adapters/kiro/glossary.go b/internal/adapters/kiro/glossary.go
index 066017a..a4997bd 100644
--- a/internal/adapters/kiro/glossary.go
+++ b/internal/adapters/kiro/glossary.go
@@ -118,6 +118,7 @@ var glossary = model.NewGlossary(Dialect,
model.Term{Unified: model.ContentHashOnly, Note: "no equivalent"},
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.ContentUnavailable, Native: "...", Where: "the content of a Reasoning record"},
// ---- what began a turn ----
diff --git a/internal/adapters/kiro/kiro_test.go b/internal/adapters/kiro/kiro_test.go
index e59c528..6e5cdc1 100644
--- a/internal/adapters/kiro/kiro_test.go
+++ b/internal/adapters/kiro/kiro_test.go
@@ -455,6 +455,47 @@ func TestCancelledTurn(t *testing.T) {
}
}
+// A prompt held for its turn is counted once, by the call that lands the
+// last line before it, though a small budget cuts the lines before it into
+// several files and each cut drains the source again.
+func TestAHeldPromptIsCountedOnceAcrossCuts(t *testing.T) {
+ lines := fixtureLines(t, combined)
+ src := t.TempDir()
+ dir := filepath.Join(src, "sessions", "01fceef077f333f9", combined)
+ header, err := os.ReadFile(filepath.Join(fixtures, "sessions", "01fceef077f333f9", combined, "session.json"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ mkfile(t, filepath.Join(dir, "session.json"), string(header))
+ // The first turn whole, then the next prompt, whose turn is not written yet.
+ mkfile(t, filepath.Join(dir, "messages.jsonl"), strings.Join(lines[:27], "\n")+"\n")
+
+ z := storage.NewZone(t.TempDir())
+ col := kiro.New(src, z, 2048)
+ held, records := 0, 0
+ for i := 0; ; i++ {
+ st, err := col.CollectAll(nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ held, records = held+st.Held, records+st.Records
+ if st.Complete() {
+ break
+ }
+ if i > 10 {
+ t.Fatal("collection does not finish")
+ }
+ }
+ files, err := storage.LandedFiles(z, combined)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if held != 1 || records != 27 || len(files) < 3 {
+ t.Fatalf("held %d, landed %d records in %d files; want the prompt held once and the rest landed in several files",
+ held, records, len(files))
+ }
+}
+
// A prompt waits for the turn it opens. Landed before its turn's first record
// is written, it would sit outside that turn for good, since a landed record
// is never written again.
@@ -845,3 +886,72 @@ func mkfile(t *testing.T, path, body string) {
t.Fatal(err)
}
}
+
+// TestImagesTooLargeToLandAreLeftOut lands Kiro Crew's prompt with two
+// images under a budget its record does not fit. The two images are left
+// out, the sentence beside them lands, every file fits, and the prompt's step
+// says too_large. The budget is the one measured to leave out the images
+// alone: at 1,050 bytes the sentence goes too, at 1,100 one image stays.
+func TestImagesTooLargeToLandAreLeftOut(t *testing.T) {
+ const budget = 1075
+ z := storage.NewZone(t.TempDir())
+ r := &root{t: t, zone: z, col: kiro.New(fixtures, z, budget)}
+ // A small budget takes more files than one pass lands.
+ for i := 0; ; i++ {
+ st, err := r.col.CollectAll(nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if st.Complete() {
+ break
+ }
+ if i > 10 {
+ t.Fatal("collection does not finish")
+ }
+ }
+ r.parse(images)
+ files, err := storage.LandedFiles(z, images)
+ if err != nil {
+ t.Fatal(err)
+ }
+ media := 0
+ for _, lf := range files {
+ data, err := os.ReadFile(lf.Path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(data) > budget {
+ t.Errorf("%s is %d bytes, over the budget of %d", filepath.Base(lf.Path), len(data), budget)
+ }
+ _, recs, err := sessiondata.All(bytes.NewReader(data))
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, rec := range recs {
+ for _, p := range rec.Parts {
+ switch {
+ case p.State != model.ContentTooLarge:
+ case p.Kind == sessiondata.PartMedia && p.Media == "image/png" && len(p.SHA256) == 64 && p.Bytes > 0 && len(p.Data) == 0:
+ media++
+ default:
+ t.Errorf("%s left out a %s part", rec.ID, p.Kind)
+ }
+ }
+ }
+ }
+ if media != 2 {
+ t.Fatalf("left out %d image(s), want the two", media)
+ }
+ found := false
+ for _, n := range nodes(r.doc(images)) {
+ if n.State == model.ContentTooLarge {
+ if n.Kind != model.KindMessageExternal || n.Text == "" {
+ t.Fatalf("step %s %s says too_large with text %q; want the prompt, with its sentence", n.ID, n.Kind, n.Text)
+ }
+ found = true
+ }
+ }
+ if !found {
+ t.Fatal("no step says too_large")
+ }
+}
diff --git a/internal/adapters/langsmith/bodies.go b/internal/adapters/langsmith/bodies.go
index e3f0341..9333db5 100644
--- a/internal/adapters/langsmith/bodies.go
+++ b/internal/adapters/langsmith/bodies.go
@@ -112,6 +112,7 @@ func bodiesOf(items []placed, records [][]sessiondata.Record) []body {
// verify reports the damage itself.
func (c *Collector) loadHeld(session string) (*providerbody.Session, error) {
held := providerbody.NewSession()
+ held.Budget(c.MaxDeltaBytes)
files, err := storage.LandedFiles(c.Zone, session)
if err != nil {
return nil, err
diff --git a/internal/adapters/langsmith/budget_test.go b/internal/adapters/langsmith/budget_test.go
new file mode 100644
index 0000000..309fbba
--- /dev/null
+++ b/internal/adapters/langsmith/budget_test.go
@@ -0,0 +1,43 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package langsmith
+
+import (
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/apache/skywalking-ai-sessionizer/pkg/model"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+)
+
+// The check that every record can be written writes them to nowhere, under
+// the budget, and must not change them: a question it fitted would be gone
+// when the conversation is placed and named afterwards.
+func TestTheEncodableCheckChangesNoRecord(t *testing.T) {
+ c := &Collector{MaxDeltaBytes: 4 << 10, Now: func() time.Time { return time.Unix(0, 0) }}
+ question := strings.Repeat("why is the sky blue? ", 1_000)
+ records := []sessiondata.Record{{ID: "r1", Bytes: len(question),
+ Parts: []sessiondata.Part{{Kind: sessiondata.PartText, Text: question, State: model.ContentAvailable}}}}
+ if err := c.encodable("s1", "main", records, Waiting{Path: "request-1.json"}); err != nil {
+ t.Fatal(err)
+ }
+ if p := records[0].Parts[0]; p.Text != question || p.State != model.ContentAvailable {
+ t.Fatalf("the check left the question %s with %d bytes of text", p.State, len(p.Text))
+ }
+}
diff --git a/internal/adapters/langsmith/collect.go b/internal/adapters/langsmith/collect.go
index 769ebcc..db63966 100644
--- a/internal/adapters/langsmith/collect.go
+++ b/internal/adapters/langsmith/collect.go
@@ -52,9 +52,9 @@ type Collector struct {
Ownership Ownership
// MaxDeltaBytes is the largest landed file a conversion writes. One
// request can carry more: a single turn with a 46 KB command and a 32 KB
- // result measured 1.29 MB. A record larger than the budget lands alone,
- // because a file is cut by narrowing what goes in it and never by
- // splitting a record.
+ // result measured 1.29 MB. A file is cut by narrowing what goes in it and
+ // never by splitting a record, so a record larger than the budget lands
+ // alone, with its largest payloads omitted until it fits.
MaxDeltaBytes int64
// ProviderBodies lands what each model call was sent and what came
// back, beside the conversation, cut against what the session already
@@ -814,9 +814,14 @@ func (c *Collector) encodable(session, stream string, records []sessiondata.Reco
if err != nil {
return err
}
+ writer.Budget(c.MaxDeltaBytes)
for i := range records {
+ // A copy, parts included: fitting under the budget changes a record
+ // in place, and this check must not change what is placed and
+ // landed afterwards.
record := records[i]
record.Ord, record.Off = uint64(i+1), 0
+ record.Parts = append([]sessiondata.Part(nil), record.Parts...)
if err := writer.Write(&record); err != nil {
if errors.Is(err, sessiondata.ErrNotEncodable) {
return &unreadable{err}
@@ -847,19 +852,12 @@ func (c *Collector) landStream(g *grouped, state *storage.SessionState, stream s
off += uint64(records[i].Bytes) + 1
}
// One request can be larger than a landed file should be, so it is cut
- // across files here. The cut narrows what goes in a file and never splits
- // a record: a record over the budget lands alone in a file of its own,
- // which is the same rule every other landed file follows.
+ // across files here, by what each record takes once encoded. The cut
+ // narrows what goes in a file and never splits a record: a record over
+ // the budget lands alone in a file of its own, with its largest payloads
+ // omitted until it fits, which is the same rule every other landed file
+ // follows.
for start := 0; start < len(records); {
- end, size := start, int64(0)
- for end < len(records) {
- next := size + int64(records[end].Bytes)
- if end > start && next > c.MaxDeltaBytes {
- break
- }
- size = next
- end++
- }
seq := state.Take()
header := &sessiondata.Header{
Seq: seq, At: now.Format(time.RFC3339Nano),
@@ -867,6 +865,27 @@ func (c *Collector) landStream(g *grouped, state *storage.SessionState, stream s
Dialect: Dialect, Src: filepath.Base(request.Path),
Session: g.session, Stream: stream,
}
+ cut, err := sessiondata.NewCut(header, c.MaxDeltaBytes)
+ if err != nil {
+ state.GiveBack(seq)
+ return err
+ }
+ end := start
+ for end < len(records) {
+ ok, err := cut.Take(&records[end])
+ if errors.Is(err, sessiondata.ErrNotEncodable) {
+ state.GiveBack(seq)
+ return &unreadable{err}
+ }
+ if err != nil {
+ state.GiveBack(seq)
+ return err
+ }
+ if !ok {
+ break
+ }
+ end++
+ }
path := filepath.Join(streamDir, storage.LandedName("transcript", storage.Stamp(now), seq))
batch := records[start:end]
// A failure here is almost always the disk, and a disk is retried:
@@ -880,6 +899,7 @@ func (c *Collector) landStream(g *grouped, state *storage.SessionState, stream s
if err != nil {
return err
}
+ writer.Budget(c.MaxDeltaBytes)
for i := range batch {
if err := writer.Write(&batch[i]); err != nil {
return err
diff --git a/internal/adapters/langsmith/glossary.go b/internal/adapters/langsmith/glossary.go
index ab51f5d..372ae48 100644
--- a/internal/adapters/langsmith/glossary.go
+++ b/internal/adapters/langsmith/glossary.go
@@ -120,6 +120,7 @@ var glossary = model.NewGlossary(Dialect,
model.Term{Unified: model.ContentHashOnly, Note: "no equivalent"},
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.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 349798a..d94b4a5 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -289,8 +289,9 @@ type Collector struct {
// MaxDeltaBytes caps how much of a source is landed in a single file, so a
// large catch-up is split rather than producing one enormous delta. A
// file travels whole as one log record, so this is also the largest
- // record a receiver has to accept, apart from a single source record
- // larger than the budget, which is landed whole. The default is 2 MiB.
+ // record a receiver has to accept. A single source record larger than the
+ // budget lands alone, with its largest payloads omitted until it fits.
+ // The default is 2 MiB.
MaxDeltaBytes int64 `yaml:"max_delta_bytes"`
}
diff --git a/internal/scenario/expect/budget_test.go b/internal/scenario/expect/budget_test.go
new file mode 100644
index 0000000..42ed861
--- /dev/null
+++ b/internal/scenario/expect/budget_test.go
@@ -0,0 +1,84 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package expect
+
+import (
+ "bytes"
+ "encoding/json"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/storage"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+)
+
+// landOne writes one landed file of one record, unfitted, as a file a
+// collector could not make smaller would be.
+func landOne(t *testing.T, root string, kind sessiondata.Kind, part sessiondata.Part) {
+ t.Helper()
+ hdr := &sessiondata.Header{Seq: 1, At: "2026-01-01T00:00:00Z", Kind: kind, Adapter: "a/1", Dialect: "d", Src: "s.jsonl", Session: "s1", Stream: "main"}
+ var buf bytes.Buffer
+ w, err := sessiondata.NewWriter(&buf, hdr)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := w.Write(&sessiondata.Record{Ord: 1, Sha: "x", Bytes: 1, Parts: []sessiondata.Part{part}}); err != nil {
+ t.Fatal(err)
+ }
+ if err := w.Close(); err != nil {
+ t.Fatal(err)
+ }
+ dir := storage.NewZone(root).StreamDir("s1", "main")
+ if err := os.MkdirAll(dir, 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(filepath.Join(dir, storage.LandedName(string(kind), "20260101T000000.000000000Z", 1)), buf.Bytes(), 0o444); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// A file over the budget is allowed only when it holds one record Fit could
+// make no smaller, by Fit's own rule: an execution record is never taken
+// out, so it may stand over a tiny budget, and a record that still holds
+// text may not.
+func TestFilesFitTheBudgetFollowsFitsRule(t *testing.T) {
+ exec := json.RawMessage(`{"schema":"execution/1","id":"e1","observed_by":"asz-plugin","boundary":"mcp","session":"s1","stream":"main","tool":"t1","protocol":"mcp","time":"2026-01-01T00:00:00Z","outcome":"returned"}`)
+ root := t.TempDir()
+ landOne(t, root, sessiondata.KindExecution, sessiondata.Part{Kind: sessiondata.PartData, Data: exec, State: "available"})
+ if out, err := FilesFitTheBudget(root, "s1", 50); err != nil || len(out) != 0 {
+ t.Fatalf("an execution record over a 50-byte budget: %v, %v", out, err)
+ }
+ root = t.TempDir()
+ landOne(t, root, sessiondata.KindTranscript, sessiondata.Part{Kind: sessiondata.PartText, Text: strings.Repeat("a", 500), State: "available"})
+ if out, err := FilesFitTheBudget(root, "s1", 200); err != nil || len(out) != 1 {
+ t.Fatalf("a record that still holds text over the budget: %v, %v", out, err)
+ }
+}
+
+// Only a provider body's own manifest is excused, not a transcript record
+// whose data happens to read like one.
+func TestOnlyAProviderBodysManifestIsExcused(t *testing.T) {
+ root := t.TempDir()
+ lookalike := json.RawMessage(`{"schema":"provider_body/1","payload":"` + strings.Repeat("p", 500) + `"}`)
+ landOne(t, root, sessiondata.KindTranscript, sessiondata.Part{Kind: sessiondata.PartData, Data: lookalike, State: "available"})
+ if out, err := FilesFitTheBudget(root, "s1", 200); err != nil || len(out) != 1 {
+ t.Fatalf("a transcript record that reads like a manifest, over the budget: %v, %v", out, err)
+ }
+}
diff --git a/internal/scenario/expect/expect.go b/internal/scenario/expect/expect.go
index ea5befd..fb76363 100644
--- a/internal/scenario/expect/expect.go
+++ b/internal/scenario/expect/expect.go
@@ -22,6 +22,7 @@
package expect
import (
+ "bytes"
"encoding/json"
"errors"
"fmt"
@@ -88,6 +89,7 @@ type Properties struct {
HeaderMatchesFold *bool `yaml:"header_matches_fold"`
RecordsMatch *bool `yaml:"records_match"`
RecordsWellFormed *bool `yaml:"records_well_formed"`
+ FilesFitTheBudget *bool `yaml:"files_fit_the_budget"`
EveryLineARecord *bool `yaml:"every_line_a_record"`
PartsKeepSourceBytes *bool `yaml:"parts_keep_source_bytes"`
DiscoveryIgnoresNoise *bool `yaml:"discovery_ignores_noise"`
@@ -299,6 +301,13 @@ type View struct {
// Flags counts the document's steps by each flag they carry, so a
// scenario can say which records landed named.
Flags map[string]int `yaml:"flags"`
+ // States counts the document's steps by the state of their content and
+ // of their result, so a scenario can say which parts landed whole and
+ // which did not. A step whose content is available is not counted.
+ States map[string]int `yaml:"states"`
+ // ChangeStates counts the document's change records by the state of
+ // the part they landed in; one that landed whole is not counted.
+ ChangeStates map[string]int `yaml:"change_states"`
// Hide builds the document with these flags withheld, as a view
// configured or asked to hide them does, and Withheld is what its
// summary must then count, every key of it, zeros included. Every step
@@ -697,6 +706,34 @@ func checkView(root, session string, want *View) ([]string, error) {
}
}
}
+ if want.States != nil {
+ got := map[string]int{}
+ eachNode(doc, func(n *sessionview.Node) {
+ for _, state := range []string{n.State, n.ResultState} {
+ if state != "" && state != model.ContentAvailable {
+ got[state]++
+ }
+ }
+ })
+ for state, n := range want.States {
+ if got[state] != n {
+ bad("view.states.%s is %d, want %d", state, got[state], n)
+ }
+ }
+ }
+ if want.ChangeStates != nil {
+ got := map[string]int{}
+ for _, wc := range doc.WorkspaceChanges {
+ if wc.State != "" {
+ got[wc.State]++
+ }
+ }
+ for state, n := range want.ChangeStates {
+ if got[state] != n {
+ bad("view.change_states.%s is %d, want %d", state, got[state], n)
+ }
+ }
+ }
if want.Executions != nil && len(doc.ToolExecutions) != *want.Executions {
bad("view.executions is %d, want %d", len(doc.ToolExecutions), *want.Executions)
}
@@ -1111,7 +1148,13 @@ func viewOf(hdr *sessiondata.Header, r *sessiondata.Record) recordView {
if p.Failed != nil {
failed = fmt.Sprint(*p.Failed)
}
- parts = append(parts, fmt.Sprintf("%s|%s|%s|%s|%s|%s|%s", p.Kind, p.ID, p.Of, p.Name, p.State, failed, text))
+ // A part made smaller to fit the budget keeps the size and digest
+ // of what it carried, which both formats must give alike.
+ kept := ""
+ if p.State == model.ContentTooLarge || p.State == model.ContentTruncated {
+ kept = fmt.Sprintf("%d/%s", p.Bytes, p.SHA256)
+ }
+ parts = append(parts, fmt.Sprintf("%s|%s|%s|%s|%s|%s|%s|%s", p.Kind, p.ID, p.Of, p.Name, p.State, failed, text, kept))
}
return recordView{
Kind: string(hdr.Kind), Stream: hdr.Stream, Batch: hdr.Batch,
@@ -1254,6 +1297,57 @@ func RecordsWellFormed(root, session string) ([]string, error) {
return out, nil
}
+// FilesFitTheBudget checks that every landed file is at most budget bytes.
+// A collector cuts a file by what its records take once encoded, and a
+// record over the budget lands alone with its largest payloads left out, so
+// the one file allowed over the budget holds a single record with no payload
+// left to take out.
+func FilesFitTheBudget(root, session string, budget int64) ([]string, error) {
+ if budget <= 0 {
+ budget = 2 << 20
+ }
+ files, err := storage.LandedFiles(storage.NewZone(root), session)
+ if err != nil {
+ return nil, err
+ }
+ var out []string
+ for _, lf := range files {
+ data, err := os.ReadFile(lf.Path)
+ if err != nil {
+ return nil, err
+ }
+ if int64(len(data)) <= budget {
+ continue
+ }
+ hdr, recs, err := sessiondata.All(bytes.NewReader(data))
+ if err != nil {
+ return nil, err
+ }
+ if len(recs) == 1 && !holdsPayload(hdr.Kind, recs[0]) {
+ continue
+ }
+ out = append(out, fmt.Sprintf("files_fit_the_budget: %s is %d bytes in %d record(s), over the budget of %d",
+ filepath.Base(lf.Path), len(data), len(recs), budget))
+ }
+ return out, nil
+}
+
+// holdsPayload reports whether a record still holds a payload that could be
+// taken out, by Fit's own rule. A provider body's manifest, the last part of
+// a record in a provider_body file, is never taken out.
+func holdsPayload(kind sessiondata.Kind, rec *sessiondata.Record) bool {
+ manifest := -1
+ if _, err := providerbody.ManifestOf(rec); kind == sessiondata.KindProviderBody && err == nil {
+ manifest = len(rec.Parts) - 1
+ }
+ for i, p := range rec.Parts {
+ if i != manifest && sessiondata.Payload(p) > 0 {
+ return true
+ }
+ }
+ return false
+}
+
// ViewCoversTheSession checks the asz.view document holds the whole
// session: every round of the chain, verified; every landed file, with its
// digest as on disk; every talk, run and step of the fold in a tree, under
@@ -1610,10 +1704,14 @@ func ProviderBodiesRebuild(root, session string) ([]string, int, error) {
continue
}
body, err := held.Body(rec.ID)
+ // A body that holds or names a piece too large to land cannot be
+ // rebuilt from the root. parts_keep_source_bytes rebuilds it from
+ // the file the runtime wrote, where a format has one.
+ tooLarge := errors.Is(err, providerbody.ErrTooLarge)
switch {
- case err != nil:
+ case err != nil && !tooLarge:
bad("%s does not rebuild from the files up to %d: %v", rec.ID, r.Ref.Seq, err)
- case len(body) != m.Bytes || providerbody.Digest(body) != m.SHA256:
+ case !tooLarge && (len(body) != m.Bytes || providerbody.Digest(body) != m.SHA256):
bad("%s rebuilds to other bytes than its manifest states", rec.ID)
case m.Role != r.Role:
bad("%s is a %s, listed as a %s", rec.ID, m.Role, r.Role)
diff --git a/internal/scenario/run/run.go b/internal/scenario/run/run.go
index d7d6134..03f10d2 100644
--- a/internal/scenario/run/run.go
+++ b/internal/scenario/run/run.go
@@ -244,6 +244,9 @@ func checkFormat(sc *scenario.Scenario, ex *expect.File, f scenario.Format, out
{"fold_equals_parse", ex.Properties.FoldEqualsParse, func() ([]string, error) { return expect.FoldEqualsParse(out, session, session) }},
{"immutable_rounds", ex.Properties.ImmutableRounds, func() ([]string, error) { return expect.ImmutableRounds(out, session) }},
{"records_well_formed", ex.Properties.RecordsWellFormed, func() ([]string, error) { return expect.RecordsWellFormed(out, session) }},
+ {"files_fit_the_budget", ex.Properties.FilesFitTheBudget, func() ([]string, error) {
+ return expect.FilesFitTheBudget(out, session, ex.Collect.MaxDeltaBytes)
+ }},
{"repack_keeps_structure", ex.Properties.RepackKeepsStructure, func() ([]string, error) { return repackKeepsStructure(out, session, ex.Parse.MaxRoundBytes) }},
{"push_follows_the_wire", ex.Properties.PushFollowsTheWire, func() ([]string, error) {
return pushFollowsTheWire(out, session, f, ex.Push, expectedTokens(built.Plan), expectedMCP(built.Plan), expect.On(ex.Properties.MetricsMatchThePlan))
diff --git a/internal/scenario/run/sourcebytes.go b/internal/scenario/run/sourcebytes.go
index 75aa3ad..760c657 100644
--- a/internal/scenario/run/sourcebytes.go
+++ b/internal/scenario/run/sourcebytes.go
@@ -21,6 +21,8 @@ import (
"bytes"
"crypto/sha256"
"encoding/hex"
+ "encoding/json"
+ "errors"
"fmt"
"os"
"path/filepath"
@@ -28,6 +30,7 @@ import (
"github.com/apache/skywalking-ai-sessionizer/internal/storage"
"github.com/apache/skywalking-ai-sessionizer/pkg/changes"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/model"
"github.com/apache/skywalking-ai-sessionizer/pkg/providerbody"
"github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
)
@@ -114,18 +117,39 @@ func partsKeepSourceBytes(root, session string, sources map[string]string) ([]st
fail("%s row %d: sha %s, and its source's bytes give %s", l.name, row+1, rec.Sha, got)
continue
}
+ var values map[string]bool
for b, p := range rec.Parts {
+ // A changes/1 record an adapter derives beside a transcript's
+ // result is asz's own encoding, whole or made smaller, and no
+ // slice of the source.
+ if l.hdr.Kind == sessiondata.KindTranscript && p.Kind == sessiondata.PartData {
+ if _, derived := changes.Decode(p.Data); derived {
+ continue
+ }
+ }
+ if p.State == model.ContentTooLarge || p.State == model.ContentTruncated {
+ // What landed is not the source's bytes, or nothing of
+ // them, so the digest of what the part carried is checked
+ // instead: it must be the digest of a value of its source
+ // record. A part taken out keeps no payload, apart from an
+ // unknown part's reason; a part shrunk keeps a smaller one.
+ if values == nil {
+ values = valueDigests(line)
+ }
+ takenOut := p.State == model.ContentTooLarge
+ kept := len(p.Data) > 0 || (p.Text != "" && p.Kind != sessiondata.PartUnknown)
+ if !values[p.SHA256] || (takenOut && kept) {
+ fail("%s row %d block %d: the %s part made smaller to fit says %s, keeps a payload %v, and names a digest its source has %v",
+ l.name, row+1, b, p.Kind, p.State, kept, values[p.SHA256])
+ }
+ continue
+ }
var kept []byte
switch p.Kind {
case sessiondata.PartCall, sessiondata.PartResult, sessiondata.PartData:
if len(p.Data) == 0 {
continue
}
- if l.hdr.Kind == sessiondata.KindTranscript && p.Kind == sessiondata.PartData {
- if _, derived := changes.Decode(p.Data); derived {
- continue
- }
- }
kept = p.Data
case sessiondata.PartUnknown:
raw, err := p.Raw()
@@ -167,6 +191,10 @@ func providerKeepsSourceBytes(name, dir string, recs []*sessiondata.Record, held
fail("%s row %d: sha %s and %d bytes, and its file %s gives %s and %d", name, row+1, rec.Sha, rec.Bytes, m.Src, got, len(src))
}
for b, p := range rec.Parts[:len(rec.Parts)-1] {
+ if p.State == model.ContentTooLarge {
+ // Checked below, where the body is rebuilt from its file.
+ continue
+ }
kept := []byte(p.Data)
if p.Kind == sessiondata.PartUnknown {
if kept, err = p.Raw(); err != nil {
@@ -183,6 +211,11 @@ func providerKeepsSourceBytes(name, dir string, recs []*sessiondata.Record, held
continue
}
body, err := held.Body(rec.ID)
+ if errors.Is(err, providerbody.ErrTooLarge) {
+ // A piece of it was too large to land. Its digest names it, so
+ // the file the runtime wrote gives it back.
+ body, err = held.BodyFrom(rec.ID, src)
+ }
if err != nil {
fail("%s row %d: %v", name, row+1, err)
continue
@@ -192,3 +225,49 @@ func providerKeepsSourceBytes(name, dir string, recs []*sessiondata.Record, held
}
}
}
+
+// valueDigests returns the SHA-256 of every value in a source record: each
+// string as its text, and every value both as the source wrote it and as
+// compact JSON, and of the whole line, which may not be JSON at all. A part
+// omitted as too large names the digest of one of them: data as the writer
+// lands it, compacted, and an unknown part's bytes as the source had them.
+func valueDigests(line []byte) map[string]bool {
+ out := map[string]bool{digestHex(line): true}
+ var walk func(raw json.RawMessage)
+ walk = func(raw json.RawMessage) {
+ out[digestHex(bytes.TrimSpace(raw))] = true
+ var compact bytes.Buffer
+ if json.Compact(&compact, raw) != nil {
+ return
+ }
+ out[digestHex(compact.Bytes())] = true
+ switch v := bytes.TrimSpace(raw); {
+ case len(v) > 0 && v[0] == '{':
+ var m map[string]json.RawMessage
+ if json.Unmarshal(v, &m) == nil {
+ for _, x := range m {
+ walk(x)
+ }
+ }
+ case len(v) > 0 && v[0] == '[':
+ var a []json.RawMessage
+ if json.Unmarshal(v, &a) == nil {
+ for _, x := range a {
+ walk(x)
+ }
+ }
+ case len(v) > 0 && v[0] == '"':
+ var text string
+ if json.Unmarshal(v, &text) == nil {
+ out[digestHex([]byte(text))] = true
+ }
+ }
+ }
+ walk(line)
+ return out
+}
+
+func digestHex(b []byte) string {
+ sum := sha256.Sum256(b)
+ return hex.EncodeToString(sum[:])
+}
diff --git a/internal/scenario/run/sourcebytes_test.go b/internal/scenario/run/sourcebytes_test.go
new file mode 100644
index 0000000..1b62864
--- /dev/null
+++ b/internal/scenario/run/sourcebytes_test.go
@@ -0,0 +1,84 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package run
+
+import (
+ "bytes"
+ "crypto/sha256"
+ "encoding/hex"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/storage"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/model"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+)
+
+// A source line that is not JSON lands as an unknown part, whose text is
+// the reason, not content. Taken out as too large, it keeps the reason and
+// names the digest of the line, and the check accepts it; with another
+// digest it does not.
+func TestAnUnknownPartTakenOutKeepsItsReason(t *testing.T) {
+ line := []byte("{not json " + strings.Repeat("z", 10_000))
+ sum := sha256.Sum256(line)
+ for _, tc := range []struct {
+ digest string
+ ok bool
+ }{{hex.EncodeToString(sum[:]), true}, {strings.Repeat("0", 64), false}} {
+ root, src := t.TempDir(), t.TempDir()
+ if err := os.MkdirAll(filepath.Join(src, "proj"), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(filepath.Join(src, "proj", "s.jsonl"), append(line, '\n'), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ z := storage.NewZone(root)
+ hdr := &sessiondata.Header{Seq: 1, At: "2026-01-01T00:00:00Z", Kind: sessiondata.KindTranscript,
+ Adapter: "claude-code-local/0.1.0", Dialect: "claude-code/1", Src: "proj/s.jsonl", Session: "s1", Stream: "main"}
+ var buf bytes.Buffer
+ w, err := sessiondata.NewWriter(&buf, hdr)
+ if err != nil {
+ t.Fatal(err)
+ }
+ rec := &sessiondata.Record{Ord: 1, Off: 0, Sha: hex.EncodeToString(sum[:])[:12], Bytes: len(line),
+ Parts: []sessiondata.Part{{Kind: sessiondata.PartUnknown, Text: "the record is not valid JSON",
+ State: model.ContentTooLarge, Bytes: len(line), SHA256: tc.digest}}}
+ if err := w.Write(rec); err != nil {
+ t.Fatal(err)
+ }
+ if err := w.Close(); err != nil {
+ t.Fatal(err)
+ }
+ dir := z.StreamDir("s1", "main")
+ if err := os.MkdirAll(dir, 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(filepath.Join(dir, storage.LandedName("transcript", "20260101T000000.000000000Z", 1)), buf.Bytes(), 0o444); err != nil {
+ t.Fatal(err)
+ }
+ out, err := partsKeepSourceBytes(root, "s1", map[string]string{"claude-code-local": src})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if (len(out) == 0) != tc.ok {
+ t.Fatalf("digest %s…: the check says %v, want accepted %v", tc.digest[:8], out, tc.ok)
+ }
+ }
+}
diff --git a/internal/scenario/scenario.go b/internal/scenario/scenario.go
index b3feaa3..76ddad0 100644
--- a/internal/scenario/scenario.go
+++ b/internal/scenario/scenario.go
@@ -36,6 +36,7 @@ import (
"sort"
"strings"
"time"
+ "unicode/utf8"
"gopkg.in/yaml.v3"
)
@@ -139,6 +140,23 @@ type Step struct {
type Queued struct {
Text string `yaml:"text"`
Mode string `yaml:"mode"`
+ // Pad repeats the text until it is this many bytes long, as a result's
+ // Pad does: a message typed while the agent works can carry a pasted
+ // screenshot, measured at 4.6 MB.
+ Pad int `yaml:"pad"`
+}
+
+// UnmarshalYAML applies Pad as the step is read, so every writer gets the
+// same text.
+func (q *Queued) UnmarshalYAML(n *yaml.Node) error {
+ type plain Queued
+ var p plain
+ if err := n.Decode(&p); err != nil {
+ return err
+ }
+ *q = Queued(p)
+ q.Text = padded(q.Text, q.Pad)
+ return nil
}
// Inject is material the harness put into model context, of a named type.
@@ -216,6 +234,25 @@ type Change struct {
Path string `yaml:"path"`
Before *string `yaml:"before"`
After *string `yaml:"after"`
+ // Pad repeats After until it is this many bytes long, as a result's Pad
+ // does, so a change record can be larger than a landed file's budget.
+ Pad int `yaml:"pad"`
+}
+
+// UnmarshalYAML applies Pad as the step is read, so every writer gets the
+// same file.
+func (c *Change) UnmarshalYAML(n *yaml.Node) error {
+ type plain Change
+ var p plain
+ if err := n.Decode(&p); err != nil {
+ return err
+ }
+ *c = Change(p)
+ if c.After != nil && c.Pad > 0 {
+ after := padded(*c.After, c.Pad)
+ c.After = &after
+ }
+ return nil
}
// Result is what a tool returned. In a Tool it may be written as a plain
@@ -231,6 +268,11 @@ type Result struct {
String bool `yaml:"string"`
// Lost says the result never reached the file, while the call did.
Lost bool `yaml:"lost"`
+ // Pad repeats the text until it is this many bytes long, so a scenario
+ // can return more than a landed file's byte budget without spelling it
+ // out. It is applied as the scenario is read, so every writer gets the
+ // same text.
+ Pad int `yaml:"pad"`
}
// Synthetic is an assistant-role message the client fabricated. It may be
@@ -269,9 +311,29 @@ func (r *Result) UnmarshalYAML(n *yaml.Node) error {
return err
}
*r = Result(p)
+ r.Text = padded(r.Text, r.Pad)
return nil
}
+// padded repeats text until it is n bytes long, or returns it as it is when
+// it is already that long. The cut stays on a character boundary, so the
+// text stays valid UTF-8 and both formats write the same bytes.
+func padded(text string, n int) string {
+ if n <= len(text) {
+ return text
+ }
+ unit := text
+ if unit == "" {
+ unit = "x"
+ }
+ long := strings.Repeat(unit, n/len(unit)+1)
+ end := n
+ for end > 0 && !utf8.RuneStart(long[end]) {
+ end--
+ }
+ return long[:end]
+}
+
// Agent is a request to start a child agent: the call, a launch
// acknowledgement, the child's own stream, and, when Notify is set, the
// runtime reporting the child finished, which continues the talk in a new
diff --git a/internal/scenario/write_sd.go b/internal/scenario/write_sd.go
index 98a6c80..626bd87 100644
--- a/internal/scenario/write_sd.go
+++ b/internal/scenario/write_sd.go
@@ -184,31 +184,54 @@ func (w *sdWriter) land(dir, prefix, cursorName string, hdr sessiondata.Header,
return false, nil
}
fresh := recs[cur.Ord:]
- seq := w.state.Take()
- hdr.H, hdr.Schema, hdr.Seq = 1, "", seq
+ hdr.H, hdr.Schema = 1, ""
hdr.At = w.now.UTC().Format(time.RFC3339Nano)
hdr.Adapter, hdr.Dialect, hdr.Session = mock.Name+"/"+mock.Version, mock.Dialect, w.p.Session
- path := filepath.Join(dir, storage.LandedName(prefix, storage.Stamp(w.now), seq))
- err = storage.WriteAtomic(path, storage.PermLanded, func(out io.Writer) error {
- rw, err := sessiondata.NewWriter(out, &hdr)
+ // Files are cut as a collector cuts them, by what the records take once
+ // encoded, and a record over the budget loses its largest payloads, so
+ // both formats land the same parts.
+ for len(fresh) > 0 {
+ seq := w.state.Take()
+ hdr.Seq = seq
+ cut, err := sessiondata.NewCut(&hdr, w.maxDelta)
if err != nil {
- return err
+ return false, err
+ }
+ n := 0
+ for n < len(fresh) {
+ ok, err := cut.Take(fresh[n])
+ if err != nil {
+ return false, err
+ }
+ if !ok {
+ break
+ }
+ n++
}
- for _, r := range fresh {
- if err := rw.Write(r); err != nil {
+ path := filepath.Join(dir, storage.LandedName(prefix, storage.Stamp(w.now), seq))
+ err = storage.WriteAtomic(path, storage.PermLanded, func(out io.Writer) error {
+ rw, err := sessiondata.NewWriter(out, &hdr)
+ if err != nil {
return err
}
+ rw.Budget(w.maxDelta)
+ for _, r := range fresh[:n] {
+ if err := rw.Write(r); err != nil {
+ return err
+ }
+ }
+ return rw.Close()
+ })
+ if err != nil {
+ return false, err
}
- return rw.Close()
- })
- if err != nil {
- return false, err
- }
- cur.Ord, cur.LastSeq, cur.State = uint64(len(recs)), seq, storage.CursorActive
- if err := cur.Save(cursorPath, w.now); err != nil {
- return false, err
+ cur.Ord, cur.LastSeq, cur.State = cur.Ord+uint64(n), seq, storage.CursorActive
+ if err := cur.Save(cursorPath, w.now); err != nil {
+ return false, err
+ }
+ w.written = append(w.written, path)
+ fresh = fresh[n:]
}
- w.written = append(w.written, path)
return true, nil
}
@@ -431,6 +454,10 @@ func (w *sdWriter) record(e *Event) *sessiondata.Record {
data = map[string]any{"runId": e.Launch.Run, "status": "async_launched", "transcriptDir": "subagents/workflows/" + e.Launch.Run, "workflowName": e.Launch.Name}
case e.StringEnrichment:
data = "a bare string, not an object"
+ case len(e.Changes) > 0 && isEditingTool(e.ToolName):
+ // The runtime's own record of an editing tool's result, which
+ // the adapter keeps as the result's data.
+ data = w.p.runtimeResult(e)
}
if !e.Replayed {
part.Data = jsLine(data)
@@ -513,6 +540,7 @@ func (w *sdWriter) provider() error {
return err
}
held := providerbody.NewSession()
+ held.Budget(w.maxDelta)
files, err := storage.LandedFiles(w.z, w.p.Session)
if err != nil {
return err
diff --git a/internal/verify/provider.go b/internal/verify/provider.go
index 40a2ed2..17c5958 100644
--- a/internal/verify/provider.go
+++ b/internal/verify/provider.go
@@ -46,7 +46,8 @@ func (p BodyProblem) String() string {
// a body that no longer rebuilds: a piece or an earlier body it names is not
// there, or the bytes it rebuilds to do not match its digest. So every body is
// rebuilt, in landing order, and compared. A body landed twice with the same
-// digest is a repeat, as in a stream.
+// digest is a repeat, as in a stream. A body that holds or refers to a piece
+// too large to land cannot be rebuilt, and is counted, not reported.
func Provider(dir string) (*StreamReport, []BodyProblem, error) {
files, err := landedFiles(dir, string(sessiondata.KindProviderBody))
if err != nil {
@@ -75,7 +76,16 @@ func Provider(dir string) (*StreamReport, []BodyProblem, error) {
return nil, nil, fmt.Errorf("verify: %s row %d: %w", path, row, rerr)
}
rep.Records++
+ rep.TooLarge += tooLarge(rec)
rep.FirstOrd, rep.LastOrd = 1, 1
+ // A rebuilt body has the manifest's size and digest, so the record
+ // is held to the manifest, which a body that cannot be rebuilt,
+ // and a repeat, still have.
+ if m, err := providerbody.ManifestOf(rec); err == nil &&
+ (rec.Bytes != m.Bytes || len(m.SHA256) < 12 || rec.Sha != m.SHA256[:12]) {
+ problems = append(problems, BodyProblem{File: path, Row: row,
+ Why: "the record's size or digest is not the body's"})
+ }
if err := s.Add(rec); err != nil {
if errors.Is(err, providerbody.ErrRepeat) {
rep.Relanded++
@@ -84,14 +94,12 @@ func Provider(dir string) (*StreamReport, []BodyProblem, error) {
problems = append(problems, BodyProblem{File: path, Row: row, Why: err.Error()})
continue
}
- m, _ := providerbody.ManifestOf(rec)
- body, err := s.Body(rec.ID)
+ _, err := s.Body(rec.ID)
switch {
+ case errors.Is(err, providerbody.ErrTooLarge):
+ rep.Unbuilt++
case err != nil:
problems = append(problems, BodyProblem{File: path, Row: row, Why: err.Error()})
- case rec.Bytes != len(body) || len(m.SHA256) < 12 || rec.Sha != m.SHA256[:12]:
- problems = append(problems, BodyProblem{File: path, Row: row,
- Why: "the record's size or digest is not the body's"})
}
}
f.Close()
diff --git a/internal/verify/provider_test.go b/internal/verify/provider_test.go
new file mode 100644
index 0000000..632f8d9
--- /dev/null
+++ b/internal/verify/provider_test.go
@@ -0,0 +1,138 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package verify_test
+
+import (
+ "bytes"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/verify"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/providerbody"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+)
+
+// landBodies cuts bodies under a budget and lands them in one provider_body
+// file, as a collector does, letting edit change a record before it lands.
+func landBodies(t *testing.T, dir string, budget int64, edit func(*sessiondata.Record), bodies ...[]byte) {
+ t.Helper()
+ s := providerbody.NewSession()
+ s.Budget(budget)
+ var buf bytes.Buffer
+ w, err := sessiondata.NewWriter(&buf, &sessiondata.Header{Seq: 1, At: "2026-01-01T00:00:00Z",
+ Kind: sessiondata.KindProviderBody, Adapter: "a/1", Dialect: "d", Src: ".", Session: "s1"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ for i, b := range bodies {
+ rec, err := s.Encode(providerbody.Body{ID: "body" + string(rune('a'+i)), Role: providerbody.RoleResponse, Src: "x.json", Bytes: b})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if edit != nil {
+ edit(rec)
+ }
+ if err := w.Write(rec); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if err := w.Close(); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.MkdirAll(dir, 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(filepath.Join(dir, "provider_body-20260101T000000.000000000Z-000001.sd"), buf.Bytes(), 0o444); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func screenshotBody() []byte {
+ return []byte(`{"content":[{"type":"text","text":"` + strings.Repeat("iVBORw0KGgo", 4_000) + `"}],"id":"msg_1"}`)
+}
+
+// A body that holds a piece too large to land is counted, not reported: the
+// collector left the piece out on purpose.
+func TestABodyThatCannotBeRebuiltForAPieceTooLargeIsCounted(t *testing.T) {
+ dir := t.TempDir()
+ landBodies(t, dir, 16<<10, nil, screenshotBody())
+ rep, problems, err := verify.Provider(dir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(problems) != 0 || rep.Unbuilt != 1 || rep.TooLarge != 1 {
+ t.Fatalf("problems %v, unbuilt %d, too large %d; want none, 1 and 1", problems, rep.Unbuilt, rep.TooLarge)
+ }
+}
+
+// Such a body is still held to its manifest: a record whose size is not the
+// body's is a problem whether or not the body can be rebuilt.
+func TestABodyThatCannotBeRebuiltIsStillHeldToItsManifest(t *testing.T) {
+ dir := t.TempDir()
+ landBodies(t, dir, 16<<10, func(r *sessiondata.Record) { r.Bytes++ }, screenshotBody())
+ _, problems, err := verify.Provider(dir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(problems) != 1 || !strings.Contains(problems[0].Why, "size or digest") {
+ t.Fatalf("problems %v, want the record's size reported", problems)
+ }
+}
+
+// A repeat is held to its manifest too: the same record landed again with
+// another size is a problem, though the session holds the body already.
+func TestARepeatIsStillHeldToItsManifest(t *testing.T) {
+ dir := t.TempDir()
+ landBodies(t, dir, 16<<10, nil, screenshotBody())
+ first := filepath.Join(dir, "provider_body-20260101T000000.000000000Z-000001.sd")
+ f, err := os.Open(first)
+ if err != nil {
+ t.Fatal(err)
+ }
+ hdr, recs, err := sessiondata.All(f)
+ f.Close()
+ if err != nil {
+ t.Fatal(err)
+ }
+ again := *recs[0]
+ again.Bytes++
+ hdr.Seq = 2
+ var buf bytes.Buffer
+ w, err := sessiondata.NewWriter(&buf, &hdr)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := w.Write(&again); err != nil {
+ t.Fatal(err)
+ }
+ if err := w.Close(); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(filepath.Join(dir, "provider_body-20260101T000000.000000000Z-000002.sd"), buf.Bytes(), 0o444); err != nil {
+ t.Fatal(err)
+ }
+ rep, problems, err := verify.Provider(dir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if rep.Relanded != 1 || len(problems) != 1 || !strings.Contains(problems[0].Why, "size or digest") {
+ t.Fatalf("relanded %d, problems %v; want the repeat counted and its size reported", rep.Relanded, problems)
+ }
+}
diff --git a/internal/verify/session.go b/internal/verify/session.go
index 2b3d1c4..4476ecf 100644
--- a/internal/verify/session.go
+++ b/internal/verify/session.go
@@ -40,6 +40,10 @@ type SessionReport struct {
// Relanded counts records an interrupted pass caused to be landed twice.
// It is reported so it is visible, not because it is wrong.
Relanded int
+ // TooLarge counts the parts whose payload was too large to land, and
+ // Unbuilt the provider bodies that cannot be rebuilt for it. They are
+ // reported so they are visible, not because they are wrong.
+ TooLarge, Unbuilt int
// Bodies lists the provider bodies that do not rebuild. They are counted
// in Problems too.
Bodies []BodyProblem
@@ -66,6 +70,7 @@ func Session(z *storage.Zone, session string) (*SessionReport, error) {
rep.Streams = append(rep.Streams, sr)
rep.Records += sr.Records
rep.Relanded += sr.Relanded
+ rep.TooLarge += sr.TooLarge
rep.Problems += len(sr.OrdGaps) + len(sr.ByteGaps) + len(sr.ShaBad)
if sr.End != nil {
rep.Problems++
@@ -107,6 +112,8 @@ func Session(z *storage.Zone, session string) (*SessionReport, error) {
rep.Streams = append(rep.Streams, pr)
rep.Records += pr.Records
rep.Relanded += pr.Relanded
+ rep.TooLarge += pr.TooLarge
+ rep.Unbuilt += pr.Unbuilt
rep.Bodies = bodies
rep.Problems += len(bodies)
}
diff --git a/internal/verify/verify.go b/internal/verify/verify.go
index 2483448..cd6201d 100644
--- a/internal/verify/verify.go
+++ b/internal/verify/verify.go
@@ -34,6 +34,7 @@ import (
"strings"
"github.com/apache/skywalking-ai-sessionizer/internal/storage"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/model"
"github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
)
@@ -91,6 +92,14 @@ type StreamReport struct {
// cursor says the collector read to.
End *EndGap
+ // TooLarge counts the parts whose payload was too large to land, and
+ // Unbuilt the provider bodies that cannot be rebuilt because a piece of
+ // them was. Neither is a problem: a collector omits such a payload on
+ // purpose, keeping its size and digest, so the landed file fits its
+ // byte budget. They are counted so the loss is visible.
+ TooLarge int
+ Unbuilt int
+
// Relanded counts records that repeat a range already landed.
//
// This is not a problem. The collector lands data BEFORE it commits the
@@ -174,6 +183,7 @@ func Stream(dir, kind string) (*StreamReport, error) {
return nil, fmt.Errorf("verify: %s row %d: %w", path, row, rerr)
}
rep.Records++
+ rep.TooLarge += tooLarge(rec)
if first {
rep.FirstOrd = rec.Ord
@@ -218,6 +228,17 @@ func Stream(dir, kind string) (*StreamReport, error) {
return rep, nil
}
+// tooLarge counts a record's parts whose payload was too large to land.
+func tooLarge(rec *sessiondata.Record) int {
+ n := 0
+ for _, p := range rec.Parts {
+ if p.State == model.ContentTooLarge {
+ n++
+ }
+ }
+ return n
+}
+
// landedFiles lists a directory's landed files of one kind, in sequence order.
func landedFiles(dir, kind string) ([]string, error) {
items, err := os.ReadDir(dir)
diff --git a/internal/view/changes.go b/internal/view/changes.go
index f6f13b9..d9bed17 100644
--- a/internal/view/changes.go
+++ b/internal/view/changes.go
@@ -48,7 +48,7 @@ func (c *Conversation) workspaceChanges(landed []storage.LandedFile, recs map[[2
var out []sessionview.WorkspaceChange
seen := map[string]bool{}
- add := func(r *changes.Record, ref sessionflow.Ref) {
+ add := func(r *changes.Record, ref sessionflow.Ref, state string) {
// The id is the tool-use id, so two producers observing one call
// share it; who captured the record tells them apart, and a line
// landed twice by an interrupted pass is the same on both counts.
@@ -57,7 +57,10 @@ func (c *Conversation) workspaceChanges(landed []storage.LandedFile, recs map[[2
return
}
seen[key] = true
- out = append(out, sessionview.WorkspaceChange{Step: stepOf[r.Tool], Ref: ref, Record: *r})
+ if state == model.ContentAvailable {
+ state = ""
+ }
+ out = append(out, sessionview.WorkspaceChange{Step: stepOf[r.Tool], Ref: ref, State: state, Record: *r})
}
// The runtime's own patches, on the result records the tool steps read.
@@ -77,7 +80,7 @@ func (c *Conversation) workspaceChanges(landed []storage.LandedFile, recs map[[2
}
if r, ok := changes.Decode(p.Data); ok {
block := b
- add(r, sessionflow.Ref{Seq: ref.Seq, Row: ref.Row, Block: &block})
+ add(r, sessionflow.Ref{Seq: ref.Seq, Row: ref.Row, Block: &block}, p.State)
}
}
}
@@ -108,7 +111,7 @@ func (c *Conversation) workspaceChanges(landed []storage.LandedFile, recs map[[2
}
if r, ok := changes.Decode(p.Data); ok {
block := b
- add(r, sessionflow.Ref{Seq: lf.Seq, Row: row, Block: &block})
+ add(r, sessionflow.Ref{Seq: lf.Seq, Row: row, Block: &block}, p.State)
}
}
}
diff --git a/internal/view/fill_test.go b/internal/view/fill_test.go
new file mode 100644
index 0000000..35db149
--- /dev/null
+++ b/internal/view/fill_test.go
@@ -0,0 +1,50 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package view
+
+import (
+ "testing"
+
+ "github.com/apache/skywalking-ai-sessionizer/pkg/model"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+)
+
+// A step that stands on a whole record says too_large when a part of the
+// record was too large to land, as a screenshot pasted beside a sentence is.
+// Its text is the sentence, which landed. A record of one part reads through
+// that part, so this is the case of several.
+func TestAWholeRecordStepSaysAPartWasTooLarge(t *testing.T) {
+ rec := &sessiondata.Record{Parts: []sessiondata.Part{
+ {Kind: sessiondata.PartText, Text: "look at this", State: model.ContentAvailable, Bytes: 12},
+ {Kind: sessiondata.PartMedia, Media: "image/png", State: model.ContentTooLarge, Bytes: 4_628_610, SHA256: "ab"},
+ }}
+ var s step
+ fill(&s, rec, nil)
+ if s.State != model.ContentTooLarge || s.Text != "look at this" {
+ t.Fatalf("the step says %q with text %q, want too_large and the sentence", s.State, s.Text)
+ }
+ whole := &sessiondata.Record{Parts: []sessiondata.Part{
+ {Kind: sessiondata.PartText, Text: "look at this", State: model.ContentAvailable},
+ {Kind: sessiondata.PartText, Text: "and this", State: model.ContentAvailable},
+ }}
+ var w step
+ fill(&w, whole, nil)
+ if w.State != "" {
+ t.Fatalf("a whole record that landed whole says %q", w.State)
+ }
+}
diff --git a/internal/view/step.go b/internal/view/step.go
index e5b2c76..0e33104 100644
--- a/internal/view/step.go
+++ b/internal/view/step.go
@@ -245,6 +245,14 @@ func fill(s *step, rec *sessiondata.Record, block *int) {
p := partOf(rec, block)
if p == nil {
s.Text = clip(readable(rec))
+ // A step that stands on a whole record reads every part of it, so
+ // one part too large to land, such as a screenshot pasted beside a
+ // sentence, makes the step say so. Its text is what landed.
+ for _, part := range rec.Parts {
+ if part.State == model.ContentTooLarge {
+ s.State = model.ContentTooLarge
+ }
+ }
return
}
s.Name, s.Failed, s.State, s.Bytes = p.Name, p.Failed, p.State, p.Bytes
diff --git a/pkg/changes/changes.go b/pkg/changes/changes.go
index 03e3c30..52a93f2 100644
--- a/pkg/changes/changes.go
+++ b/pkg/changes/changes.go
@@ -294,6 +294,45 @@ func (r *Record) Validate() error {
return nil
}
+// DropHunks takes the hunks out of the file whose hunks are largest, and
+// says its diff is too_large, as a diff too large to compute does. Its
+// additions and deletions stay. It reports false when no file has hunks
+// left. A landed record too large for its file is made smaller this way
+// before anything else is left out, so its identity and its counts survive.
+func (r *Record) DropHunks() bool {
+ best, size := -1, 0
+ for i, c := range r.Changes {
+ n := 0
+ for _, h := range c.Hunks {
+ for _, l := range h.Lines {
+ n += len(l)
+ }
+ n += len(h.Lines) + 1
+ }
+ if len(c.Hunks) > 0 && n > size {
+ best, size = i, n
+ }
+ }
+ if best < 0 {
+ return false
+ }
+ r.Changes[best].Hunks, r.Changes[best].Diff = nil, DiffTooLarge
+ return true
+}
+
+// DropFiles takes out the lists that grow with the workspace, once no file
+// has hunks left: the files, the overlaps and the gaps. The record keeps its
+// identity, its tool, its window and changed_files, so a reader still sees
+// that the call changed files, and how many. It reports false when nothing
+// is left to take out.
+func (r *Record) DropFiles() bool {
+ if len(r.Changes)+len(r.Overlaps)+len(r.Gaps) == 0 {
+ return false
+ }
+ r.Changes, r.Overlaps, r.Gaps = nil, nil, nil
+ return true
+}
+
// Marshal writes a record as one JSON line without a trailing newline.
func (r *Record) Marshal() ([]byte, error) {
if r.Changes == nil {
diff --git a/pkg/model/model.go b/pkg/model/model.go
index 0042cb6..8843c9e 100644
--- a/pkg/model/model.go
+++ b/pkg/model/model.go
@@ -171,6 +171,10 @@ const (
ContentSizeOnly = "size_only"
ContentTruncated = "truncated"
ContentUnavailable = "unavailable"
+ // ContentTooLarge is a payload that was not landed because its record
+ // would not fit a landed file's byte budget on its own. Its size and
+ // digest stay.
+ ContentTooLarge = "too_large"
)
// StreamRole distinguishes lineages within one session.
diff --git a/pkg/model/term.go b/pkg/model/term.go
index 8563992..d92c9d8 100644
--- a/pkg/model/term.go
+++ b/pkg/model/term.go
@@ -131,7 +131,7 @@ func Vocabulary() []string {
// qualification
ExactUnique, ExactAmbiguous, StrongInference, WeakInference, Unresolved, Conflict,
ContentAvailable, ContentRedacted, ContentOmitted, ContentHashOnly,
- ContentSizeOnly, ContentTruncated, ContentUnavailable,
+ ContentSizeOnly, ContentTruncated, ContentUnavailable, ContentTooLarge,
TriggerExternal, TriggerNotification, TriggerUnknown,
}
}
diff --git a/pkg/providerbody/session.go b/pkg/providerbody/session.go
index 531e3e9..01587c7 100644
--- a/pkg/providerbody/session.go
+++ b/pkg/providerbody/session.go
@@ -22,8 +22,10 @@ import (
"encoding/json"
"errors"
"fmt"
+ "sort"
"unicode/utf8"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/model"
"github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
)
@@ -31,6 +33,11 @@ import (
// file landed again after an interrupted pass.
var ErrRepeat = errors.New("providerbody: the session already holds this body")
+// ErrTooLarge reports a body that cannot be rebuilt because a piece of it was
+// too large to land: its record, or an earlier one it refers to, holds the
+// piece's size and digest and not its bytes.
+var ErrTooLarge = errors.New("providerbody: a piece of the body was too large to land")
+
// maxTips bounds how many chain tips a new body is compared against. A tip is
// one per chain that is still growing: the main stream and each child running
// at the time.
@@ -47,6 +54,10 @@ const cacheLimit = 64 << 20
// and rebuilds any of them. Records must be added in the order they landed,
// because a record may refer only to what came before it.
type Session struct {
+ // maxRecord is the largest line a record Encode returns may take in a
+ // landed file, its newline included; zero means no limit. See Budget.
+ maxRecord int64
+
records map[string]*held
pieces map[string]pieceAt
// tips holds the latest request body of each chain, by chain, and
@@ -74,6 +85,17 @@ func NewSession() *Session {
return &Session{records: map[string]*held{}, pieces: map[string]pieceAt{}, tips: map[string]string{}, cache: map[string][]byte{}}
}
+// Budget makes Encode fit every record into a landed file of budget bytes on
+// its own, the file's header and closing line included. A record over it has
+// its largest pieces taken out until it fits. Zero or less means no limit; a
+// budget no record can fit still takes out every piece it can.
+func (s *Session) Budget(budget int64) {
+ s.maxRecord = 0
+ if budget > 0 {
+ s.maxRecord = max(budget-FileOverhead, 1)
+ }
+}
+
// ManifestOf returns the manifest a provider_body record carries as its last part.
func ManifestOf(rec *sessiondata.Record) (*Manifest, error) {
if len(rec.Parts) == 0 {
@@ -155,8 +177,9 @@ func (s *Session) Add(rec *sessiondata.Record) error {
return fmt.Errorf("providerbody: %s is held with digest %s and landed again with %s", rec.ID, h.m.SHA256, m.SHA256)
}
// A repeat is a repeat only if it rebuilds to what it claims; a
- // record that says so and does not is damage.
- if _, err := s.rebuild(rec, m); err != nil {
+ // record that says so and does not is damage. One that cannot be
+ // rebuilt for a piece too large to land is still a repeat.
+ if _, err := s.rebuild(rec, m, nil); err != nil && !errors.Is(err, ErrTooLarge) {
return err
}
return ErrRepeat
@@ -166,7 +189,13 @@ func (s *Session) Add(rec *sessiondata.Record) error {
if p.Kind != sessiondata.PartData {
continue
}
- if d := digest(p.Data); !hasPiece(s.pieces, d) {
+ // A piece taken out is still a piece: a later body that holds it
+ // refers to it rather than taking it out again.
+ d := p.SHA256
+ if p.State != model.ContentTooLarge {
+ d = digest(p.Data)
+ }
+ if !hasPiece(s.pieces, d) {
s.pieces[d] = pieceAt{id: rec.ID, part: i}
}
}
@@ -207,6 +236,29 @@ func (s *Session) Body(id string) ([]byte, error) {
return append([]byte(nil), b...), nil
}
+// BodyFrom rebuilds the body held under id, as Body does, taking the bytes of
+// any piece that was too large to land from source: another copy of the
+// body, such as the file the runtime wrote. A piece is matched by its
+// digest, so a source that is not the same body gives nothing it can use.
+func (s *Session) BodyFrom(id string, source []byte) ([]byte, error) {
+ h, ok := s.records[id]
+ if !ok {
+ return nil, fmt.Errorf("providerbody: no body %s", id)
+ }
+ fill := map[string][]byte{digest(source): source}
+ if spans, _, err := pieces(source); err == nil {
+ for _, sp := range spans {
+ piece := source[sp.a:sp.z]
+ fill[digest(piece)] = piece
+ }
+ }
+ b, err := s.rebuild(h.rec, h.m, fill)
+ if err != nil {
+ return nil, err
+ }
+ return append([]byte(nil), b...), nil
+}
+
// body is Body without the copy, for this package only: a later body's copy
// must read the base as it was rebuilt, and nothing here changes it.
func (s *Session) body(id string) ([]byte, error) {
@@ -217,7 +269,7 @@ func (s *Session) body(id string) ([]byte, error) {
if !ok {
return nil, fmt.Errorf("providerbody: no body %s", id)
}
- b, err := s.rebuild(h.rec, h.m)
+ b, err := s.rebuild(h.rec, h.m, nil)
if err != nil {
return nil, err
}
@@ -238,8 +290,10 @@ func (s *Session) remember(id string, b []byte) {
}
// rebuild joins a record's segments. The record need not be added yet, so a
-// record can be checked before it lands.
-func (s *Session) rebuild(rec *sessiondata.Record, m *Manifest) ([]byte, error) {
+// record can be checked before it lands. fill holds, by digest, the bytes of
+// pieces that were too large to land, when the caller has them: the body it
+// is encoding. Without them such a piece fails with ErrTooLarge.
+func (s *Session) rebuild(rec *sessiondata.Record, m *Manifest, fill map[string][]byte) ([]byte, error) {
if m.Bytes < 0 || m.Bytes > MaxBytes {
return nil, fmt.Errorf("providerbody: %s claims %d bytes", rec.ID, m.Bytes)
}
@@ -260,7 +314,7 @@ func (s *Session) rebuild(rec *sessiondata.Record, m *Manifest) ([]byte, error)
if *seg.Part < 0 || *seg.Part >= len(rec.Parts)-1 {
return nil, fmt.Errorf("providerbody: %s names part %d it does not have", rec.ID, *seg.Part)
}
- b, err := partBytes(rec.Parts[*seg.Part])
+ b, err := partBytes(rec.Parts[*seg.Part], fill)
if err != nil {
return nil, err
}
@@ -273,7 +327,7 @@ func (s *Session) rebuild(rec *sessiondata.Record, m *Manifest) ([]byte, error)
if !ok {
return nil, fmt.Errorf("providerbody: %s names piece %s, which no earlier record holds", rec.ID, seg.Piece)
}
- b, err := partBytes(s.records[at.id].rec.Parts[at.part])
+ b, err := partBytes(s.records[at.id].rec.Parts[at.part], fill)
if err != nil {
return nil, err
}
@@ -301,8 +355,14 @@ func (s *Session) rebuild(rec *sessiondata.Record, m *Manifest) ([]byte, error)
return out, nil
}
-func partBytes(p sessiondata.Part) ([]byte, error) {
- if p.Kind == sessiondata.PartUnknown {
+func partBytes(p sessiondata.Part, fill map[string][]byte) ([]byte, error) {
+ switch {
+ case p.State == model.ContentTooLarge:
+ if b, ok := fill[p.SHA256]; ok {
+ return b, nil
+ }
+ return nil, fmt.Errorf("%w: %d bytes with digest %s", ErrTooLarge, p.Bytes, p.SHA256)
+ case p.Kind == sessiondata.PartUnknown:
return p.Raw()
}
return p.Data, nil
@@ -338,10 +398,19 @@ func (s *Session) Encode(b Body) (*sessiondata.Record, error) {
rec.Call = k.Call
}
- parts, why := s.cut(body, m)
+ // The bytes of every piece of this body, by digest, so the round trip
+ // can check a record whose pieces did not fit.
+ fill := map[string][]byte{sum: body}
+ parts, why := s.cut(body, m, fill)
if why == "" {
rec.Parts = append(parts, manifestPart(m))
- if err := s.roundTrip(rec, m); err != nil {
+ s.fit(rec, fill)
+ // What is left after the pieces go is the manifest, and a body of
+ // many short strings keeps nearly all of itself there. Whole, the
+ // body is one part, which can be taken out.
+ if s.maxRecord > 0 && RecordBytes(rec) > s.maxRecord {
+ why = "the record is larger than the budget with every piece taken out"
+ } else if err := s.roundTrip(rec, fill); err != nil {
why = err.Error()
}
}
@@ -353,7 +422,8 @@ func (s *Session) Encode(b Body) (*sessiondata.Record, error) {
// does not repeat it.
m.Depth, m.Segments = 0, []Segment{{Part: &zero}}
rec.Parts = []sessiondata.Part{whole, manifestPart(m)}
- if err := s.roundTrip(rec, m); err != nil {
+ s.fit(rec, fill)
+ if err := s.roundTrip(rec, fill); err != nil {
return nil, err
}
}
@@ -363,8 +433,9 @@ func (s *Session) Encode(b Body) (*sessiondata.Record, error) {
return rec, nil
}
-// cut builds the parts and segments of a body, or says why it cannot.
-func (s *Session) cut(body []byte, m *Manifest) ([]sessiondata.Part, string) {
+// cut builds the parts and segments of a body, or says why it cannot. It
+// adds every piece of the body to fill.
+func (s *Session) cut(body []byte, m *Manifest, fill map[string][]byte) ([]sessiondata.Part, string) {
if !utf8.Valid(body) {
return nil, "the body is not valid UTF-8"
}
@@ -396,6 +467,7 @@ func (s *Session) cut(body []byte, m *Manifest) ([]sessiondata.Part, string) {
}
piece := body[sp.a:sp.z]
d := digest(piece)
+ fill[d] = piece
switch i, ok := local[d]; {
case ok:
m.Segments = append(m.Segments, Segment{Part: &i})
@@ -513,10 +585,48 @@ func manifestPart(m *Manifest) sessiondata.Part {
return sessiondata.Part{Kind: sessiondata.PartData, Data: data, State: "available", Bytes: len(data)}
}
+// fit takes out the largest parts a record holds of its body, never its
+// manifest, until the record is a line of at most the budget's bytes. A record
+// of one body is cut by leaving out what earlier records hold, so a body whose
+// new pieces are larger than the budget is the one case left: the first
+// request after a person pasted a screenshot, measured at 4.6 MB. A part taken
+// out keeps its place and its size, and its sha256 is the piece's digest, so
+// every segment still names it and later bodies refer to it as to any piece.
+// The body then cannot be rebuilt, and a reader is told why with ErrTooLarge.
+func (s *Session) fit(rec *sessiondata.Record, fill map[string][]byte) {
+ if s.maxRecord <= 0 || RecordBytes(rec) <= s.maxRecord {
+ return
+ }
+ own := len(rec.Parts) - 1
+ order := make([]int, 0, own)
+ for i := range rec.Parts[:own] {
+ if len(rec.Parts[i].Data) > 0 {
+ order = append(order, i)
+ }
+ }
+ sort.SliceStable(order, func(a, b int) bool { return len(rec.Parts[order[a]].Data) > len(rec.Parts[order[b]].Data) })
+ for _, i := range order {
+ p := &rec.Parts[i]
+ b, err := partBytes(*p, fill)
+ if err != nil {
+ continue
+ }
+ d := digest(b)
+ fill[d] = b
+ p.Data, p.Encoding = nil, ""
+ p.State, p.SHA256, p.Bytes = model.ContentTooLarge, d, len(b)
+ if RecordBytes(rec) <= s.maxRecord {
+ return
+ }
+ }
+}
+
// roundTrip writes the record with the Session Data writer, reads it back,
// and rebuilds the body from what was read. Only that proves the bytes a
-// reader will get, since the writer compacts every data part.
-func (s *Session) roundTrip(rec *sessiondata.Record, m *Manifest) error {
+// reader will get, since the writer compacts every data part. fill gives the
+// bytes of the pieces that were too large to land, so a record that omits
+// them is still proven to rebuild the body once they are given.
+func (s *Session) roundTrip(rec *sessiondata.Record, fill map[string][]byte) error {
var buf bytes.Buffer
hdr := &sessiondata.Header{Kind: sessiondata.KindProviderBody, Session: "check", Src: ".", Dialect: "check"}
w, err := sessiondata.NewWriter(&buf, hdr)
@@ -541,10 +651,9 @@ func (s *Session) roundTrip(rec *sessiondata.Record, m *Manifest) error {
if err != nil {
return err
}
- if _, err := s.rebuild(back, bm); err != nil {
+ if _, err := s.rebuild(back, bm, fill); err != nil {
return fmt.Errorf("the record does not rebuild the body after a round trip: %w", err)
}
- _ = m
return nil
}
diff --git a/pkg/providerbody/toolarge_test.go b/pkg/providerbody/toolarge_test.go
new file mode 100644
index 0000000..776840c
--- /dev/null
+++ b/pkg/providerbody/toolarge_test.go
@@ -0,0 +1,253 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package providerbody
+
+import (
+ "bytes"
+ "errors"
+ "strings"
+ "testing"
+
+ "github.com/apache/skywalking-ai-sessionizer/pkg/model"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+)
+
+const testBudget = 16 << 10
+
+// landed writes a record as a collector does and reads it back, so a test
+// checks what a reader gets.
+func landed(t *testing.T, rec *sessiondata.Record) *sessiondata.Record {
+ t.Helper()
+ var buf bytes.Buffer
+ w, err := sessiondata.NewWriter(&buf, &sessiondata.Header{Kind: sessiondata.KindProviderBody, Session: "s1", Src: ".", Dialect: "d"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := w.Write(rec); err != nil {
+ t.Fatal(err)
+ }
+ if err := w.Close(); err != nil {
+ t.Fatal(err)
+ }
+ if int64(buf.Len()) > testBudget {
+ t.Fatalf("the landed file is %d bytes, over the budget of %d", buf.Len(), testBudget)
+ }
+ _, recs, err := sessiondata.All(&buf)
+ if err != nil || len(recs) != 1 {
+ t.Fatalf("read back %d records: %v", len(recs), err)
+ }
+ return recs[0]
+}
+
+// omitted returns the record's parts that were too large to land.
+func omitted(rec *sessiondata.Record) []sessiondata.Part {
+ var out []sessiondata.Part
+ for _, p := range rec.Parts {
+ if p.State == model.ContentTooLarge {
+ out = append(out, p)
+ }
+ }
+ return out
+}
+
+// pasted is a chain whose first message holds a 40 KB string, as a pasted
+// screenshot's base64 or a long tool result is.
+func pasted() (*chain, string) {
+ c := newChain("s1", "p1", strings.Repeat("You are an agent. ", 30), 2)
+ big := "look at this: " + strings.Repeat("iVBORw0KGgo", 40_000/11)
+ c.add("user", big)
+ return c, big
+}
+
+func TestABodyOverTheBudgetOmitsItsLargestPiece(t *testing.T) {
+ s := NewSession()
+ s.Budget(testBudget)
+ c, big := pasted()
+ req := c.request()
+ rec, err := encode(s, "00000000-0000-4000-8000-000000000001.request.json", req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ piece := compact(big)
+ gone := omitted(rec)
+ if len(gone) != 1 || gone[0].SHA256 != Digest(piece) || gone[0].Bytes != len(piece) || len(gone[0].Data) != 0 {
+ t.Fatalf("omitted %d part(s), want the one 40 KB piece with its size and digest", len(gone))
+ }
+ if _, err := ManifestOf(rec); err != nil {
+ t.Fatalf("the manifest did not survive: %v", err)
+ }
+ // The cutter cannot rebuild what it did not land, and says why; the file
+ // the runtime wrote gives the piece back.
+ if _, err := s.Body(rec.ID); !errors.Is(err, ErrTooLarge) {
+ t.Fatalf("Body returned %v, want ErrTooLarge", err)
+ }
+ if got, err := s.BodyFrom(rec.ID, req); err != nil || !bytes.Equal(got, req) {
+ t.Fatalf("BodyFrom did not give the body back: %v", err)
+ }
+ // A reader of the landed file sees the same.
+ r := NewSession()
+ back := landed(t, rec)
+ if err := r.Add(back); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := r.Body(back.ID); !errors.Is(err, ErrTooLarge) {
+ t.Fatalf("a reader's Body returned %v, want ErrTooLarge", err)
+ }
+ if got, err := r.BodyFrom(back.ID, req); err != nil || !bytes.Equal(got, req) {
+ t.Fatalf("a reader's BodyFrom did not give the body back: %v", err)
+ }
+ // Landing the file again is a repeat, not damage.
+ if err := r.Add(back); !errors.Is(err, ErrRepeat) {
+ t.Fatalf("landed again: %v, want ErrRepeat", err)
+ }
+}
+
+func TestALaterBodyNamesTheOmittedPiece(t *testing.T) {
+ s := NewSession()
+ s.Budget(testBudget)
+ c, big := pasted()
+ first, err := encode(s, "00000000-0000-4000-8000-000000000001.request.json", c.request())
+ if err != nil {
+ t.Fatal(err)
+ }
+ c.add("assistant", "I see a screenshot.")
+ c.add("user", "what is in it?")
+ req := c.request()
+ rec, err := encode(s, "00000000-0000-4000-8000-000000000002.request.json", req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ m, _ := ManifestOf(rec)
+ names := false
+ for _, seg := range m.Segments {
+ names = names || seg.Piece == Digest(compact(big))
+ // The first body cannot be rebuilt, so nothing copies from it.
+ if seg.Copy != nil && seg.Copy.From == first.ID {
+ t.Fatalf("copies from %s, which cannot be rebuilt", first.ID)
+ }
+ }
+ if !names || len(omitted(rec)) != 0 {
+ t.Fatalf("the later body omits %d part(s) and names the piece %v; want it named, not omitted again", len(omitted(rec)), names)
+ }
+ if _, err := s.Body(rec.ID); !errors.Is(err, ErrTooLarge) {
+ t.Fatalf("Body returned %v, want ErrTooLarge", err)
+ }
+ if got, err := s.BodyFrom(rec.ID, req); err != nil || !bytes.Equal(got, req) {
+ t.Fatalf("BodyFrom did not give the body back: %v", err)
+ }
+ r := NewSession()
+ for _, x := range []*sessiondata.Record{first, rec} {
+ if err := r.Add(landed(t, x)); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if got, err := r.BodyFrom(rec.ID, req); err != nil || !bytes.Equal(got, req) {
+ t.Fatalf("a reader's BodyFrom did not give the later body back: %v", err)
+ }
+}
+
+func TestABodyLandedWholeIsOmittedWhole(t *testing.T) {
+ s := NewSession()
+ s.Budget(testBudget)
+ raw := []byte(`{"model":"claude-opus-5","messages":[` + strings.Repeat(`"x",`, 10_000))
+ rec, err := encode(s, "00000000-0000-4000-8000-000000000001.request.json", raw)
+ if err != nil {
+ t.Fatal(err)
+ }
+ whole := rec.Parts[0]
+ if whole.Kind != sessiondata.PartUnknown || whole.State != model.ContentTooLarge || whole.SHA256 != Digest(raw) || whole.Text == "" {
+ t.Fatalf("the whole part is %s %s with digest %q, want an unknown part omitted with the body's digest and its reason kept", whole.Kind, whole.State, whole.SHA256)
+ }
+ if got, err := s.BodyFrom(rec.ID, raw); err != nil || !bytes.Equal(got, raw) {
+ t.Fatalf("BodyFrom did not give the body back: %v", err)
+ }
+ landed(t, rec)
+}
+
+// Under a budget no record fits, every piece goes, the manifest stays, and
+// the body lands whole in one part left out, so the manifest is one segment.
+// A budget of a few bytes still limits: it does not mean no limit.
+func TestTheManifestIsNeverOmitted(t *testing.T) {
+ for _, budget := range []int64{FileOverhead + 1, 100} {
+ s := NewSession()
+ s.Budget(budget)
+ c, _ := pasted()
+ req := c.request()
+ rec, err := encode(s, "00000000-0000-4000-8000-000000000001.request.json", req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ m, err := ManifestOf(rec)
+ if err != nil {
+ t.Fatalf("budget %d: the manifest was omitted: %v", budget, err)
+ }
+ if whole := rec.Parts[0]; len(rec.Parts) != 2 || whole.Kind != sessiondata.PartUnknown || whole.State != model.ContentTooLarge || len(m.Segments) != 1 {
+ t.Fatalf("budget %d: %d parts, the first %s %s, %d segments; want the whole body left out", budget, len(rec.Parts), whole.Kind, whole.State, len(m.Segments))
+ }
+ if got, err := s.BodyFrom(rec.ID, req); err != nil || !bytes.Equal(got, req) {
+ t.Fatalf("budget %d: BodyFrom did not give the body back: %v", budget, err)
+ }
+ }
+}
+
+// A body of many short strings has no piece to leave out: they stay in the
+// manifest as literals. Whole, it is one part, which can be.
+func TestABodyOfShortStringsLandsWholeAndLeftOut(t *testing.T) {
+ s := NewSession()
+ s.Budget(testBudget)
+ var b strings.Builder
+ b.WriteString(`{"model":"claude-opus-5","messages":[`)
+ for i := 0; i < 2_000; i++ {
+ if i > 0 {
+ b.WriteByte(',')
+ }
+ b.WriteString(`{"role":"user","content":"a short message of no length"}`)
+ }
+ b.WriteString(`]}`)
+ req := []byte(b.String())
+ rec, err := encode(s, "00000000-0000-4000-8000-000000000001.request.json", req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if RecordBytes(rec) > testBudget-FileOverhead {
+ t.Fatalf("the record is %d bytes, over the budget", RecordBytes(rec))
+ }
+ if whole := rec.Parts[0]; whole.Kind != sessiondata.PartUnknown || whole.State != model.ContentTooLarge || whole.SHA256 != Digest(req) {
+ t.Fatalf("the first part is %s %s, want the whole body left out with its digest", whole.Kind, whole.State)
+ }
+ if got, err := s.BodyFrom(rec.ID, req); err != nil || !bytes.Equal(got, req) {
+ t.Fatalf("BodyFrom did not give the body back: %v", err)
+ }
+ landed(t, rec)
+}
+
+func TestWithoutABudgetNothingIsOmitted(t *testing.T) {
+ s := NewSession()
+ c, _ := pasted()
+ req := c.request()
+ rec, err := encode(s, "00000000-0000-4000-8000-000000000001.request.json", req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if n := len(omitted(rec)); n != 0 {
+ t.Fatalf("omitted %d part(s) with no budget", n)
+ }
+ if got, err := s.Body(rec.ID); err != nil || !bytes.Equal(got, req) {
+ t.Fatalf("Body did not give the body back: %v", err)
+ }
+}
diff --git a/pkg/sessiondata/fit.go b/pkg/sessiondata/fit.go
new file mode 100644
index 0000000..a7746ca
--- /dev/null
+++ b/pkg/sessiondata/fit.go
@@ -0,0 +1,247 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package sessiondata
+
+import (
+ "bytes"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+
+ "github.com/apache/skywalking-ai-sessionizer/pkg/changes"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/execution"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/model"
+)
+
+// endLineMax is the longest closing line a file can have, its newline
+// included: the record count takes at most twenty digits.
+const endLineMax = len(`{"t":"end","records":,"digest":""}`) + 20 + 64 + 1
+
+// Fit makes r a line of at most limit bytes by taking out its largest
+// payloads, one at a time, and reports how many parts it changed. It changes
+// r in place. A limit of zero or less fits everything. A part loses its data
+// before its text, so a result keeps its readable text, saying truncated,
+// while the runtime's structured copy of it goes.
+//
+// A landed file travels whole, so its size is what a receiver has to store
+// at once. A file is cut by narrowing what goes into it, which works down to
+// one record. One record can still be larger than the budget: a pasted
+// screenshot measured 4.6 MB in a transcript line. So the record lands, and
+// its largest payloads do not. A part taken out keeps its place, its kind and
+// every identifier, so a reader still sees the step, and it says why its
+// content is missing: its state is too_large, or truncated while its text is
+// left, its bytes the size of the original, and its sha256 the digest of what
+// was taken out first.
+//
+// A change record and an execution record are evidence a step joins to by
+// its tool call, so neither is ever taken out whole: a reader that met
+// nothing would read that the call changed no file or ran nowhere. A change
+// record is made smaller in its own format instead: its largest files lose
+// their hunks and say their diff is too_large, keeping their counts, and then
+// its file list goes, keeping its identity and changed_files. Its part says
+// truncated. An execution record keeps only sizes and digests, never content.
+//
+// A record that is still too large with every payload taken out lands as it
+// is: its identifiers, label and part list remain.
+func Fit(r *Record, limit int) (int, error) {
+ _, n, err := fit(r, limit)
+ return n, err
+}
+
+// fit is Fit, also returning the record's line, so a writer encodes it once.
+func fit(r *Record, limit int) ([]byte, int, error) {
+ line, err := encodeRecord(r)
+ if err != nil || limit <= 0 {
+ return line, 0, err
+ }
+ changed := map[int]bool{}
+ for len(line) > limit {
+ i := largest(r.Parts)
+ if i < 0 {
+ break
+ }
+ if !shrink(&r.Parts[i]) {
+ omit(&r.Parts[i])
+ }
+ changed[i] = true
+ if line, err = encodeRecord(r); err != nil {
+ break
+ }
+ }
+ return line, len(changed), err
+}
+
+// largest is the part with the most payload Fit can take out, the earlier of
+// two of the same size, so the same record always loses the same parts. It
+// is -1 when nothing is left to take out.
+func largest(parts []Part) int {
+ best, size := -1, 0
+ for i, p := range parts {
+ if n := Payload(p); n > size {
+ best, size = i, n
+ }
+ }
+ return best
+}
+
+// Payload is how much Fit can take out of a part in its next step: its data
+// while it has some, and then its text, unless it is an unknown part, whose
+// text is the reason its bytes were not described rather than content. An
+// execution record, and a change record with nothing left to shrink, have
+// none, so a record whose parts all have none is as small as Fit makes it.
+func Payload(p Part) int {
+ if p.State == model.ContentTooLarge {
+ return 0
+ }
+ if len(p.Data) > 0 {
+ if p.Kind == PartData {
+ if _, ok := execution.Decode(p.Data); ok {
+ return 0
+ }
+ if rec, ok := changes.Decode(p.Data); ok && !shrinkable(rec) {
+ return 0
+ }
+ }
+ return len(p.Data)
+ }
+ if p.Kind == PartUnknown {
+ return 0
+ }
+ return len(p.Text)
+}
+
+func shrinkable(rec *changes.Record) bool {
+ return len(rec.Changes)+len(rec.Overlaps)+len(rec.Gaps) > 0
+}
+
+// shrink makes a change record smaller by one step: the hunks of its
+// largest file, or, once no file has hunks, its file list. It reports false
+// for any other part.
+func shrink(p *Part) bool {
+ if p.Kind != PartData {
+ return false
+ }
+ rec, ok := changes.Decode(p.Data)
+ if !ok || (!rec.DropHunks() && !rec.DropFiles()) {
+ return false
+ }
+ data, err := rec.Marshal()
+ if err != nil {
+ return false
+ }
+ remember(p)
+ p.Data, p.State = data, model.ContentTruncated
+ return true
+}
+
+// omit takes one payload out of a part and says so: its data first, while it
+// has some, so a result keeps its readable text and says truncated, and then
+// its text. A part with nothing left says too_large.
+func omit(p *Part) {
+ remember(p)
+ if len(p.Data) > 0 {
+ p.Data, p.Encoding = nil, ""
+ } else {
+ p.Text = ""
+ }
+ p.State = model.ContentTooLarge
+ if p.Text != "" && p.Kind != PartUnknown {
+ p.State = model.ContentTruncated
+ }
+}
+
+// remember keeps the size and the digest of what a part carried, before Fit
+// first changes it: of its data when it had some, which goes first.
+func remember(p *Part) {
+ if p.SHA256 != "" {
+ return
+ }
+ b := carried(p)
+ sum := sha256.Sum256(b)
+ p.SHA256 = hex.EncodeToString(sum[:])
+ if p.Bytes == 0 {
+ p.Bytes = len(b)
+ }
+}
+
+// carried is what a part holds: an unknown part's bytes, other data as the
+// writer lands it, compacted, or the text of a part with no data.
+func carried(p *Part) []byte {
+ if len(p.Data) == 0 {
+ return []byte(p.Text)
+ }
+ if p.Kind == PartUnknown {
+ if raw, err := p.Raw(); err == nil {
+ return raw
+ }
+ }
+ var compact bytes.Buffer
+ if json.Compact(&compact, p.Data) != nil {
+ return p.Data
+ }
+ return compact.Bytes()
+}
+
+// Cut decides which records go into one landed file of a byte budget. A
+// collector reads a source by its own bytes, and a record is larger than the
+// line it came from, by its envelope, so a file is cut by what its records
+// take once encoded.
+type Cut struct {
+ // room is what the file has left for record lines, newlines included,
+ // and limit what it had before the first.
+ limit, room int
+}
+
+// NewCut starts a file with header h under a budget of bytes. A budget of
+// zero or less takes every record.
+func NewCut(h *Header, budget int64) (*Cut, error) {
+ c := &Cut{}
+ if budget <= 0 {
+ return c, nil
+ }
+ hh := *h
+ hh.H, hh.Schema = 1, Schema
+ head, err := encodeLine(&hh)
+ if err != nil {
+ return nil, err
+ }
+ c.limit = max(int(budget)-len(head)-1-endLineMax, 1)
+ c.room = c.limit
+ return c, nil
+}
+
+// Take fits r as a Writer under the same budget does, and reports whether it
+// goes into this file: the first record always does, a later one only while
+// the file stays within the budget. A record not taken starts the next file.
+func (c *Cut) Take(r *Record) (bool, error) {
+ if c.limit <= 0 {
+ _, _, err := fit(r, 0)
+ return err == nil, err
+ }
+ // A line alone in its file takes its newline too, as Writer.Budget counts.
+ line, _, err := fit(r, max(c.limit-1, 1))
+ if err != nil {
+ return false, err
+ }
+ need := len(line) + 1
+ if c.room < c.limit && need > c.room {
+ return false, nil
+ }
+ c.room -= need
+ return true, nil
+}
diff --git a/pkg/sessiondata/fit_test.go b/pkg/sessiondata/fit_test.go
new file mode 100644
index 0000000..2edb806
--- /dev/null
+++ b/pkg/sessiondata/fit_test.go
@@ -0,0 +1,369 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package sessiondata
+
+import (
+ "bytes"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "fmt"
+ "strings"
+ "testing"
+
+ "github.com/apache/skywalking-ai-sessionizer/pkg/changes"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/model"
+)
+
+func hexDigest(b []byte) string {
+ sum := sha256.Sum256(b)
+ return hex.EncodeToString(sum[:])
+}
+
+func lineOf(t *testing.T, r *Record) int {
+ t.Helper()
+ b, err := encodeRecord(r)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return len(b)
+}
+
+func TestFitOmitsTheLargestPayloadFirst(t *testing.T) {
+ r := &Record{Ord: 1, Sha: "x", Bytes: 1, Parts: []Part{
+ {Kind: PartText, Text: strings.Repeat("a", 5_000), State: "available", Bytes: 5_000},
+ {Kind: PartText, Text: strings.Repeat("b", 30_000), State: "available", Bytes: 30_000},
+ {Kind: PartData, Data: json.RawMessage(`"` + strings.Repeat("c", 20_000) + `"`), State: "available", Bytes: 20_002},
+ }}
+ n, err := Fit(r, 30_000)
+ if err != nil || n != 1 {
+ t.Fatalf("omitted %d part(s), %v; want the one largest", n, err)
+ }
+ if r.Parts[1].State != model.ContentTooLarge || r.Parts[0].State != "available" || r.Parts[2].State != "available" {
+ t.Fatalf("states %s %s %s, want only the 30 KB text omitted", r.Parts[0].State, r.Parts[1].State, r.Parts[2].State)
+ }
+ if got := lineOf(t, r); got > 30_000 {
+ t.Fatalf("the record is %d bytes after Fit, over 30000", got)
+ }
+}
+
+func TestFitOmitsEqualPayloadsInOrder(t *testing.T) {
+ r := &Record{Ord: 1, Sha: "x", Bytes: 1, Parts: []Part{
+ {Kind: PartText, Text: strings.Repeat("a", 10_000)},
+ {Kind: PartText, Text: strings.Repeat("b", 10_000)},
+ }}
+ if n, _ := Fit(r, 15_000); n != 1 || r.Parts[0].State != model.ContentTooLarge || r.Parts[1].State == model.ContentTooLarge {
+ t.Fatalf("omitted %d; the first of two equal parts must go first", n)
+ }
+}
+
+func TestFitKeepsSizeAndDigestOfEachPayload(t *testing.T) {
+ text := strings.Repeat("t", 4_000)
+ spaced := json.RawMessage("{ \"k\" : \"" + strings.Repeat("d", 4_000) + "\" }")
+ var compact bytes.Buffer
+ _ = json.Compact(&compact, spaced)
+ unknown := Part{Kind: PartUnknown, Text: "why it was not described"}
+ raw := []byte("\xff" + strings.Repeat("u", 4_000))
+ unknown.SetRaw(raw)
+ r := &Record{Ord: 1, Sha: "x", Bytes: 1, Parts: []Part{
+ {Kind: PartText, Text: text, Bytes: 4_000},
+ {Kind: PartResult, Of: "toolu_1", Text: "the readable form", Data: spaced},
+ unknown,
+ }}
+ if n, err := Fit(r, 100); err != nil || n != 3 {
+ t.Fatalf("omitted %d part(s), %v; want all three", n, err)
+ }
+ for i, want := range []struct {
+ sha string
+ bytes int
+ }{
+ {hexDigest([]byte(text)), 4_000},
+ // A part with data and text loses both, and its digest is the data's.
+ {hexDigest(compact.Bytes()), compact.Len()},
+ // An unknown part's digest is of the bytes it kept, not their base64.
+ {hexDigest(raw), len(raw)},
+ } {
+ p := r.Parts[i]
+ if p.State != model.ContentTooLarge || len(p.Data) != 0 || p.Encoding != "" {
+ t.Fatalf("part %d still holds a payload: state %s, %d data, encoding %q", i, p.State, len(p.Data), p.Encoding)
+ }
+ // An unknown part's text is the reason its bytes were not described,
+ // not content, so it stays.
+ if keep := p.Kind == PartUnknown; (p.Text != "") != keep {
+ t.Fatalf("part %d has %d bytes of text left", i, len(p.Text))
+ }
+ if p.SHA256 != want.sha || p.Bytes != want.bytes {
+ t.Fatalf("part %d has digest %s and %d bytes, want %s and %d", i, p.SHA256, p.Bytes, want.sha, want.bytes)
+ }
+ }
+ // Identifiers stay, so a reader still joins the result to its call.
+ if r.Parts[1].Of != "toolu_1" || r.Parts[1].Kind != PartResult {
+ t.Fatalf("the result lost its identity: %+v", r.Parts[1])
+ }
+}
+
+func TestFitLandsARecordThatCannotShrinkAsItIs(t *testing.T) {
+ r := &Record{Ord: 1, Sha: "x", Bytes: 1, Label: strings.Repeat("l", 10_000),
+ Parts: []Part{{Kind: PartText, Text: "short"}}}
+ n, err := Fit(r, 1_000)
+ if err != nil || n != 1 || r.Label == "" {
+ t.Fatalf("omitted %d, %v; want the one payload omitted and the record kept", n, err)
+ }
+}
+
+func TestFitChangesNothingWithoutALimitOrWhenItFits(t *testing.T) {
+ for _, max := range []int{0, -1, 1 << 20} {
+ r := &Record{Ord: 1, Sha: "x", Bytes: 1, Parts: []Part{{Kind: PartText, Text: strings.Repeat("a", 5_000)}}}
+ if n, _ := Fit(r, max); n != 0 || r.Parts[0].Text == "" {
+ t.Fatalf("max %d omitted %d part(s)", max, n)
+ }
+ }
+}
+
+func TestWriterBudgetFitsEachRecordIntoItsFile(t *testing.T) {
+ const budget = 8 << 10
+ var buf bytes.Buffer
+ w, err := NewWriter(&buf, &Header{Kind: KindTranscript, Session: "s", Src: "a.jsonl", Dialect: "d", Stream: "main"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ w.Budget(budget)
+ rec := &Record{Ord: 1, Sha: "x", Bytes: 1, Parts: []Part{{Kind: PartText, Text: strings.Repeat("a", 20_000)}}}
+ if err := w.Write(rec); err != nil {
+ t.Fatal(err)
+ }
+ if err := w.Close(); err != nil {
+ t.Fatal(err)
+ }
+ if buf.Len() > budget {
+ t.Fatalf("a file of one record is %d bytes, over the budget of %d", buf.Len(), budget)
+ }
+ // The caller's record is the one that landed, for its index.
+ if rec.Parts[0].State != model.ContentTooLarge {
+ t.Fatalf("the caller's record still says %s", rec.Parts[0].State)
+ }
+ _, recs, err := All(&buf)
+ if err != nil || len(recs) != 1 || recs[0].Parts[0].SHA256 != hexDigest([]byte(strings.Repeat("a", 20_000))) {
+ t.Fatalf("read back %d records, %v", len(recs), err)
+ }
+}
+
+func TestWriterBudgetLeavesProviderBodiesToTheirCutter(t *testing.T) {
+ var buf bytes.Buffer
+ w, err := NewWriter(&buf, &Header{Kind: KindProviderBody, Session: "s", Src: ".", Dialect: "d"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ w.Budget(1 << 10)
+ rec := &Record{Ord: 1, Sha: "x", Bytes: 1, Parts: []Part{{Kind: PartData, Data: json.RawMessage(`"` + strings.Repeat("p", 5_000) + `"`)}}}
+ if err := w.Write(rec); err != nil {
+ t.Fatal(err)
+ }
+ if rec.Parts[0].State == model.ContentTooLarge {
+ t.Fatal("the writer omitted a provider body's part; only its cutter may")
+ }
+}
+
+// changeRecord is a changes/1 record of files, each with one hunk of n lines.
+func changeRecord(t *testing.T, files, n int) json.RawMessage {
+ t.Helper()
+ rec := changes.Record{Schema: changes.Schema, ID: "c1", CapturedBy: changes.CapturedByASZPlugin,
+ Session: "s", Stream: "main", Tool: "toolu_1", Time: "2026-01-01T00:00:00Z", Basis: "tool_window"}
+ for f := 0; f < files; f++ {
+ lines := make([]string, n)
+ for i := range lines {
+ lines[i] = "+" + strings.Repeat("x", 60)
+ }
+ rec.Changes = append(rec.Changes, changes.FileChange{Path: fmt.Sprintf("f%d.go", f), Operation: changes.OpModify,
+ Diff: changes.DiffAvailable, Additions: changes.Int(n), Deletions: changes.Int(0),
+ Hunks: []changes.Hunk{{NewStart: 1, NewLines: n, Lines: lines}}})
+ }
+ data, err := rec.Marshal()
+ if err != nil {
+ t.Fatal(err)
+ }
+ return data
+}
+
+// A change record keeps its identity, files and counts: its largest files
+// lose their hunks first, and only what is still too large is left out.
+func TestFitShrinksAChangeRecordBeforeLeavingItOut(t *testing.T) {
+ data := changeRecord(t, 3, 100)
+ r := &Record{Ord: 1, Sha: "x", Bytes: 1, Tool: "toolu_1", Parts: []Part{{Kind: PartData, Data: data, State: "available", Bytes: len(data)}}}
+ if _, err := Fit(r, len(data)/2); err != nil {
+ t.Fatal(err)
+ }
+ p := r.Parts[0]
+ got, ok := changes.Decode(p.Data)
+ if !ok || p.State != model.ContentTruncated || p.SHA256 != hexDigest(data) || p.Bytes != len(data) {
+ t.Fatalf("the change record is %s, decodes %v, digest %q; want truncated, decodable, with the original's digest", p.State, ok, p.SHA256)
+ }
+ dropped := 0
+ for _, c := range got.Changes {
+ if c.Diff == changes.DiffTooLarge && c.Hunks == nil && c.Additions != nil && *c.Additions == 100 {
+ dropped++
+ }
+ }
+ if dropped == 0 || dropped == len(got.Changes) || len(got.Changes) != 3 {
+ t.Fatalf("%d of %d files lost their hunks; want some, with their counts kept", dropped, len(got.Changes))
+ }
+ // Under a budget even the record without hunks does not fit, its file
+ // list goes too, and the record stays: who captured it, its id and its
+ // tool, so a reader still sees that the call changed files. It is never
+ // taken out whole, and one part changed several times counts once.
+ r2 := &Record{Ord: 1, Sha: "x", Bytes: 1, Parts: []Part{{Kind: PartData, Data: data}}}
+ n, err := Fit(r2, 100)
+ if err != nil || n != 1 {
+ t.Fatalf("Fit changed %d part(s), %v; want the one", n, err)
+ }
+ p2 := r2.Parts[0]
+ left, ok := changes.Decode(p2.Data)
+ if !ok || p2.State != model.ContentTruncated || p2.SHA256 != hexDigest(data) || left.ID != "c1" || left.Tool != "toolu_1" || len(left.Changes) != 0 {
+ t.Fatalf("the change record is %s, decodes %v, with %d files; want it truncated to its identity", p2.State, ok, len(left.Changes))
+ }
+}
+
+// An execution record keeps sizes and digests only, and is never taken out,
+// however small the budget: a step joins to it by its tool call.
+func TestFitNeverTakesOutAnExecutionRecord(t *testing.T) {
+ data := json.RawMessage(`{"schema":"execution/1","id":"e1","observed_by":"asz-plugin","boundary":"mcp","session":"s","stream":"main","tool":"toolu_1","protocol":"mcp","time":"2026-01-01T00:00:00Z","outcome":"returned","arguments":{"state":"size_only","bytes":10,"sha256":"ab"}}`)
+ r := &Record{Ord: 1, Sha: "x", Bytes: 1, Tool: "toolu_1", Parts: []Part{{Kind: PartData, Data: data, State: "available"}}}
+ if n, err := Fit(r, 50); err != nil || n != 0 || r.Parts[0].State != "available" || len(r.Parts[0].Data) == 0 {
+ t.Fatalf("Fit changed %d part(s), %v, state %s", n, err, r.Parts[0].State)
+ }
+}
+
+// A record exactly at the limit lands whole; one byte over, it does not.
+func TestFitAtTheExactLimit(t *testing.T) {
+ r := &Record{Ord: 1, Sha: "x", Bytes: 1, Parts: []Part{{Kind: PartText, Text: strings.Repeat("a", 5_000)}}}
+ n := lineOf(t, r)
+ if c, _ := Fit(r, n); c != 0 {
+ t.Fatalf("a record of exactly %d bytes lost %d part(s)", n, c)
+ }
+ if c, _ := Fit(r, n-1); c != 1 || r.Parts[0].State != model.ContentTooLarge {
+ t.Fatalf("a record one byte over its limit lost %d part(s)", c)
+ }
+}
+
+// A file is cut by what its records take once encoded: small records, each
+// far under the budget, fill files of at most the budget, and the first
+// record of a file is always taken.
+func TestCutPacksByEncodedSize(t *testing.T) {
+ const budget = 2 << 10
+ h := &Header{Kind: KindTranscript, Session: "s", Src: "a.jsonl", Dialect: "d", Stream: "main", Seq: 1, At: "2026-01-01T00:00:00Z"}
+ var recs []*Record
+ for i := 0; i < 100; i++ {
+ recs = append(recs, &Record{Ord: uint64(i + 1), Sha: "x", Bytes: 2, ID: fmt.Sprintf("r%03d", i), Parts: []Part{{Kind: PartText, Text: "{}"}}})
+ }
+ files := 0
+ for len(recs) > 0 {
+ cut, err := NewCut(h, budget)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var buf bytes.Buffer
+ w, err := NewWriter(&buf, h)
+ if err != nil {
+ t.Fatal(err)
+ }
+ w.Budget(budget)
+ n := 0
+ for n < len(recs) {
+ ok, err := cut.Take(recs[n])
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !ok {
+ break
+ }
+ if err := w.Write(recs[n]); err != nil {
+ t.Fatal(err)
+ }
+ n++
+ }
+ if err := w.Close(); err != nil {
+ t.Fatal(err)
+ }
+ if n == 0 || buf.Len() > budget {
+ t.Fatalf("file %d took %d records in %d bytes, budget %d", files, n, buf.Len(), budget)
+ }
+ recs = recs[n:]
+ files++
+ }
+ if files < 2 {
+ t.Fatalf("100 records went into %d file(s) of %d bytes", files, budget)
+ }
+ // One record larger than the budget is still taken, fitted.
+ cut, _ := NewCut(h, budget)
+ big := &Record{Ord: 1, Sha: "x", Bytes: 1, Parts: []Part{{Kind: PartText, Text: strings.Repeat("a", 10_000)}}}
+ if ok, err := cut.Take(big); !ok || err != nil || big.Parts[0].State != model.ContentTooLarge {
+ t.Fatalf("a large first record was taken %v, %v, state %s", ok, err, big.Parts[0].State)
+ }
+}
+
+// A part with text and data loses its data first, the larger and the
+// structured copy, and keeps its readable text, saying truncated; only a
+// tighter limit takes the text too. The digest stays the data's.
+func TestFitTakesDataBeforeText(t *testing.T) {
+ data := json.RawMessage(`{"content":"` + strings.Repeat("c", 20_000) + `"}`)
+ mk := func() *Record {
+ return &Record{Ord: 1, Sha: "x", Bytes: 1, Parts: []Part{
+ {Kind: PartResult, Of: "toolu_1", Text: "File created successfully", Data: data, State: "available", Bytes: 25},
+ }}
+ }
+ r := mk()
+ if _, err := Fit(r, 1_000); err != nil {
+ t.Fatal(err)
+ }
+ if p := r.Parts[0]; p.State != model.ContentTruncated || p.Text != "File created successfully" || len(p.Data) != 0 || p.SHA256 != hexDigest(data) {
+ t.Fatalf("the result is %s with text %q and digest %q; want truncated, its text kept, the data's digest", p.State, p.Text, p.SHA256)
+ }
+ r = mk()
+ if _, err := Fit(r, 50); err != nil {
+ t.Fatal(err)
+ }
+ if p := r.Parts[0]; p.State != model.ContentTooLarge || p.Text != "" || p.SHA256 != hexDigest(data) {
+ t.Fatalf("under a tighter limit the result is %s with text %q and digest %q; want too_large, the data's digest", p.State, p.Text, p.SHA256)
+ }
+}
+
+// A file of one record is at most the budget, whether the record fits as it
+// is or is fitted, for every text size around the budget. Cut fits a record
+// as Writer.Budget does, its newline counted.
+func TestACutRecordAloneFitsItsFile(t *testing.T) {
+ h := &Header{Kind: KindTranscript, Session: "s", Src: "a.jsonl", Dialect: "d", Stream: "main", Seq: 1, At: "2026-01-01T00:00:00Z"}
+ const budget = 1024
+ for n := 400; n < 1200; n++ {
+ cut, err := NewCut(h, budget)
+ if err != nil {
+ t.Fatal(err)
+ }
+ r := &Record{Ord: 1, Sha: "x", Bytes: 1, Parts: []Part{{Kind: PartText, Text: strings.Repeat("a", n)}}}
+ if ok, err := cut.Take(r); !ok || err != nil {
+ t.Fatal(ok, err)
+ }
+ var buf bytes.Buffer
+ w, _ := NewWriter(&buf, h)
+ w.Budget(budget)
+ _ = w.Write(r)
+ _ = w.Close()
+ if buf.Len() > budget {
+ t.Fatalf("a text of %d bytes made a file of %d bytes", n, buf.Len())
+ }
+ }
+}
diff --git a/pkg/sessiondata/sessiondata.go b/pkg/sessiondata/sessiondata.go
index b1e13b8..9587a19 100644
--- a/pkg/sessiondata/sessiondata.go
+++ b/pkg/sessiondata/sessiondata.go
@@ -161,10 +161,14 @@ type Part struct {
Encoding string `json:"encoding,omitempty"`
// State says how much of the original is here: available, truncated,
- // redacted, omitted or unavailable. A reader is always told.
+ // redacted, omitted, unavailable or too_large. A reader is always told.
State string `json:"state,omitempty"`
// Bytes is the size of the original, even when this part holds less of it.
Bytes int `json:"bytes,omitempty"`
+ // SHA256 is set on a part whose payload was too large to land: the digest
+ // of the data it would have carried, or of its text when it had no data.
+ // A reader can still match the payload against another copy of it.
+ SHA256 string `json:"sha256,omitempty"`
}
// EncodingBase64 is the Encoding of an unknown part whose bytes are not valid
diff --git a/pkg/sessiondata/writer.go b/pkg/sessiondata/writer.go
index 125f26f..3f26b95 100644
--- a/pkg/sessiondata/writer.go
+++ b/pkg/sessiondata/writer.go
@@ -48,6 +48,12 @@ type Writer struct {
bw *bufio.Writer
h hash.Hash
n int
+
+ kind Kind
+ // head is the header line's size, its newline included, and max the
+ // largest record line Budget allows: zero until it is set.
+ head int
+ max int
}
// End is the last line of a .sd file.
@@ -75,12 +81,29 @@ func NewWriter(w io.Writer, h *Header) (*Writer, error) {
if err := bw.WriteByte('\n'); err != nil {
return nil, err
}
- return &Writer{bw: bw, h: sum}, nil
+ return &Writer{bw: bw, h: sum, kind: h.Kind, head: len(enc) + 1}, nil
+}
+
+// Budget makes Write fit every record into a file of budget bytes on its
+// own, header and closing line included, by Fit. A collector cuts its files
+// at the budget by narrowing what goes into them, and this is the one case
+// that cannot narrow: a single record over the budget.
+//
+// A provider_body record is never changed here. Its last part is the
+// manifest that rebuilds the body, and its cutter fits it instead, knowing
+// which of its parts later bodies refer to.
+func (w *Writer) Budget(budget int64) {
+ if budget <= 0 || w.kind == KindProviderBody {
+ w.max = 0
+ return
+ }
+ w.max = max(int(budget)-w.head-endLineMax-1, 1)
}
-// Write appends one record.
+// Write appends one record. Under a Budget, it may first omit the record's
+// largest payloads, in place, so the caller's index sees what landed.
func (w *Writer) Write(r *Record) error {
- enc, err := encodeRecord(r)
+ enc, _, err := fit(r, w.max)
if err != nil {
return fmt.Errorf("sessiondata: encode record %d: %w", r.Ord, err)
}
diff --git a/pkg/sessionview/sessionview.go b/pkg/sessionview/sessionview.go
index 49f2249..4862fa0 100644
--- a/pkg/sessionview/sessionview.go
+++ b/pkg/sessionview/sessionview.go
@@ -109,6 +109,12 @@ type ProviderBody struct {
type WorkspaceChange struct {
Step string `json:"step"`
Ref sessionflow.Ref `json:"ref"`
+ // State is the state of the part the record landed in: truncated when
+ // it was made smaller to fit its landed file, its hunks or its file list
+ // taken out, and empty when it landed whole. A record whose file list
+ // was taken out keeps changed_files, so an empty list with a positive
+ // count is files changed, not none.
+ State string `json:"state,omitempty"`
changes.Record
}
diff --git a/tests/langchain_reset_test.go b/tests/langchain_reset_test.go
index d509c31..290a228 100644
--- a/tests/langchain_reset_test.go
+++ b/tests/langchain_reset_test.go
@@ -395,12 +395,26 @@ func TestABodyNamesThePromptItsCallWasPlacedUnder(t *testing.T) {
// conversation has to come out whole: the boundary is kept once and the
// summary joins it.
func TestAResetRecoversWhenOnlyItsBoundaryLanded(t *testing.T) {
+ // Smaller than a boundary or a summary as it was read, 5,322 bytes each,
+ // so each lands in a file of its own, and larger than the largest file of
+ // one record, 4,449 bytes, with room for a provider body file's header,
+ // so no record has its payload omitted.
+ const perRecord = 5120
zone := storage.NewZone(t.TempDir())
- session := landIntoBudget(t, zone, "summarized", 1)[0]
+ session := landIntoBudget(t, zone, "summarized", perRecord)[0]
files, err := storage.LandedFiles(zone, session)
if err != nil {
t.Fatal(err)
}
+ for _, f := range files {
+ eachRecordIn(t, f.Path, func(rec sessiondata.Record) {
+ for _, p := range rec.Parts {
+ if p.State == model.ContentTooLarge {
+ t.Fatalf("%s lost a part to the budget of %d; the corpus changed", rec.ID, perRecord)
+ }
+ }
+ })
+ }
crashAt := uint64(0)
for _, f := range files {
eachRecordIn(t, f.Path, func(rec sessiondata.Record) {
@@ -434,7 +448,7 @@ func TestAResetRecoversWhenOnlyItsBoundaryLanded(t *testing.T) {
}
t.Logf("crashed after file %d, %d files lost", crashAt, removed)
- landIntoBudget(t, zone, "summarized", 1)
+ landIntoBudget(t, zone, "summarized", perRecord)
view := parsed(t, zone, session)
resetsAre(t, view, 2)
// Each summary is a record that exists again, with the summariser's words.
diff --git a/tests/scenarios/too-large.expect.yaml b/tests/scenarios/too-large.expect.yaml
new file mode 100644
index 0000000..bc6323f
--- /dev/null
+++ b/tests/scenarios/too-large.expect.yaml
@@ -0,0 +1,30 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+checkpoints:
+ final:
+ kinds: {llm.call: 5}
+ view:
+ state: verified
+ provider_bodies: 10
+ captured_prompts: 5
+ states: {too_large: 2, truncated: 1}
+ changes: 2
+ changes_joined: 2
+ change_states: {truncated: 2}
+collect:
+ # Smaller than the result, larger than every other record.
+ max_delta_bytes: 16384
diff --git a/tests/scenarios/too-large.yaml b/tests/scenarios/too-large.yaml
new file mode 100644
index 0000000..a24fed2
--- /dev/null
+++ b/tests/scenarios/too-large.yaml
@@ -0,0 +1,62 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+# A tool returns more than a landed file may hold. The record of its result
+# is larger than the budget on its own, so it lands with the result omitted
+# as too large: the step stays, with the result's size and digest. The
+# request that sends the result back to the model is too large as well, and
+# its piece of the result is omitted the same way. The next request names
+# that piece rather than carrying it again. A message typed while the agent
+# works, as long as a pasted screenshot, is a step of a whole record, and
+# says too_large too. A command writes a file larger than the budget, and
+# its change record lands shrunk: its hunks go, its identity and counts stay,
+# and it still joins the command. The runtime's own Write tool does the same,
+# and the change record the adapter derives from the runtime's patch lands
+# shrunk beside its result. Every other record lands whole.
+title: a result too large to land
+provider_bodies: true
+steps:
+ - input: read the log
+ - call:
+ text: Reading it.
+ tool:
+ name: Read
+ input: {file_path: /Users/dev/scenario/app.log}
+ result: {text: "one line of the log\n", pad: 40000, after: 200ms}
+ - call:
+ text: Writing it out.
+ tool:
+ name: Bash
+ input: {command: ./dump > copy.log}
+ result: dumped
+ changes:
+ - path: copy.log
+ after: "one line of the log\n"
+ pad: 30000
+ - call:
+ text: Saving a copy.
+ tool:
+ name: Write
+ input: {file_path: /Users/dev/scenario/saved.log}
+ result: File created successfully
+ changes:
+ - path: saved.log
+ after: "one line of the log\n"
+ pad: 30000
+ - call: {text: The log repeats one line.}
+ - queued: {text: "and this screenshot: ", pad: 30000, mode: prompt}
+ - input: thanks
+ - call: {text: You are welcome.}