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/18 13:47:53 UTC
(skywalking-ai-sessionizer) branch main updated: Fix repack, OTLP export and container image issues from the release review (#81)
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 da56fb1 Fix repack, OTLP export and container image issues from the release review (#81)
da56fb1 is described below
commit da56fb1fecaaed02f5130f776e810b2c703f96dc
Author: 吴晟 Wu Sheng <wu...@foxmail.com>
AuthorDate: Sun Oct 18 21:47:48 2026 +0800
Fix repack, OTLP export and container image issues from the release review (#81)
asz repack moved the idle clock that retention reads from the header of
the file with the highest sequence. A file it cut kept the time of the
first file its records came from, and files were numbered stream by
stream, so the highest sequence need not be the newest file. Retention
could remove or retire a session in the new root that the source root
kept. Each new file now has the header of the newest file its records came
from, and files are numbered in the order their records landed.
asz repack counted only record lines against the budget, so a file of
several records could be larger than max_delta_bytes. It now cuts with
sessiondata.Cut, header and closing line included, through the new
Cut.TakeRaw.
asz repack carried only <kind>.cursor. A second source of a stream, such
as a second recorder directory, keeps <kind>-<hash>.cursor, so collection
into the new root read it again from its start. Every cursor is carried
now. Each names the new file holding the last record its own source
landed, and an origin relative to the storage root, a recorder inside it,
is written again with its cursor's name as a collector into the new root
computes them; the rule for both moves to storage, where the change
collector and repack share it. Records of two files are cut into one only
when the files are next to each other and their headers are the same apart
from seq and at, so every file names its own source and no record moves
past another.
A data part holding a byte that is not valid UTF-8 landed as it was, and
the OTLP export could never send its file, since a protocol buffer string
must be valid UTF-8. The writer now takes such data out: the part says
omitted, or truncated while its text is left, a change or execution record
is written again with U+FFFD, and the record's dropped list says why, so
it is not taken for a payload too large to land. The export reports a file
that cannot be encoded on its own and keeps it out of every request, and a
failed request no longer leaves out of the pass the file being batched
when it failed.
The OTLP HTTP client read a 2xx answer whose body was cut off as a full
success, and an answer longer than the 1 MiB it reads as its first MiB,
and recorded every file of the request as sent. Both are errors now, and
the request goes again.
The container image declared /asz/data without making it, so Docker made a
new named or anonymous volume owned by root, and the first collection
failed with permission denied. The image now holds the directory, owned by
the user it runs as, 65532.
---
CLAUDE.md | 4 +
Dockerfile | 8 +
docs/en/changes/changes.md | 42 +++
docs/en/concepts-and-designs/glossary.md | 3 +-
docs/en/formats/otlp.md | 14 +-
docs/en/formats/session-data.md | 32 +-
docs/en/formats/storage-root.md | 16 +-
docs/en/guides/command-line-internals.md | 18 +-
docs/en/setup/container-image.md | 11 +
internal/adapters/claudecodechanges/collect.go | 20 +-
internal/export/otlp/client.go | 23 +-
internal/export/otlp/client_test.go | 90 +++++
internal/export/otlp/push.go | 15 +-
internal/export/otlp/unsendable_test.go | 150 ++++++++
internal/repack/repack.go | 373 ++++++++++++++------
internal/storage/cursor.go | 25 ++
pkg/sessiondata/fit.go | 110 ++++--
pkg/sessiondata/fit_test.go | 134 ++++++++
pkg/sessiondata/writer_test.go | 56 ++-
tests/chain/invalid_utf8_test.go | 271 +++++++++++++++
tests/chain/repack_test.go | 453 +++++++++++++++++++++++++
21 files changed, 1699 insertions(+), 169 deletions(-)
diff --git a/CLAUDE.md b/CLAUDE.md
index 5714d1a..b807a46 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -204,6 +204,10 @@ 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.
+ The one exception: JSON holding a byte that is not valid UTF-8 is left out, its size and SHA-256
+ kept and a `dropped` entry saying why, and a change or execution record is written again with
+ U+FFFD in the byte's place. Every landed line is valid UTF-8, since an OTLP receiver
+ refuses anything else.
- **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
diff --git a/Dockerfile b/Dockerfile
index 2d927a2..84b2e61 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -30,6 +30,9 @@ RUN go mod download
COPY . .
RUN CGO_ENABLED=0 GOOS=$TARGETOS GOARCH=$TARGETARCH \
go build -trimpath -ldflags="-s -w -X main.version=${VERSION}" -o /out/asz ./cmd/asz
+# The storage root's directory, empty, for the image to own. The final image
+# has no shell to make it.
+RUN mkdir -p /out/data
# Static binary, no shell, non-root. The image reads and writes only its
# storage root, mounted at /asz/data, which is also the default the binary
@@ -45,6 +48,11 @@ COPY --from=build /out/asz /usr/local/bin/asz
# a binary distribution must carry them.
COPY dist-material/LICENSE dist-material/NOTICE /licenses/
COPY dist-material/licenses /licenses/licenses
+# The storage root belongs to the user the image runs as, 65532. Docker gives
+# a new named or anonymous volume the owner of the directory it is mounted
+# on, and makes it root's when the image has no such directory, so the first
+# collection failed to write to it.
+COPY --from=build --chown=65532:65532 /out/data /asz/data
WORKDIR /asz
VOLUME ["/asz/data"]
EXPOSE 8787
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index dc28764..9aa49ad 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -319,6 +319,44 @@
closed them, such as a test receiver shutting down, could break a request the exporter had just
started: `transport connection broken: http: CloseIdleConnections called`, seen in CI with
scenarios pushing in parallel.
+- No landed line holds a byte that is not valid UTF-8. A JSON decoder accepts such a byte inside a
+ string, so a tool input or other source JSON could bring one, and it landed as it was. A protocol
+ buffer string must be valid UTF-8, so the OTLP export could never send that file, over HTTP or
+ gRPC, and every request it was in failed on every pass. Such data is now taken out: `omitted`, or
+ `truncated` while its text is left, with its size and SHA-256 kept. A change or execution record
+ is written again in its own format, with U+FFFD in place of the byte. Either way the record's
+ `dropped` says which part lost what and why, so it is not taken for a payload too large to land.
+ See [Reading a record](../formats/session-data.md#reading-a-record).
+- One file the OTLP export cannot send no longer holds back the others. It went in a request with
+ them, the request failed, and none of them was sent; and the file being batched when that request
+ failed was left out of the pass, under the request's error. A file that cannot be encoded is now
+ reported on its own, under its name, on every pass, and the files beside it go. A failed request
+ is reported as its own error, and the file being batched goes in the next request.
+- The OTLP exporter over HTTP takes a 2xx answer whose body does not arrive whole as an error, such
+ as one whose connection closes before the length it promised, and one longer than the 1 MiB it
+ reads of an answer. It read the first as an empty answer, a full success, and the second as its
+ first MiB, and recorded every file of the request as sent, though nothing said the receiver took
+ them. They now go again on the next pass.
+- `asz repack` keeps a session's idle clock. A file it cut kept the time of the first file its
+ records came from, and the files were numbered stream by stream, so the file with the highest
+ sequence, where retention reads how long a session has been idle, need not be the newest.
+ Retention could then remove a session's files in the new root, or retire it, while the root it was
+ repacked from kept everything. Each new file now has the header of the newest file its records
+ came from, and the files are numbered in the order their records landed.
+- `asz repack` cuts a file by what it takes once written, its header and closing line included, as
+ the collectors do. It counted only the record lines, so a file of several records could be larger
+ than `max_delta_bytes`: at a budget of 1,024 bytes, three records made a file of 1,338.
+- `asz repack` carries every cursor. A stream can have more than one source, such as two recorder
+ directories, and each source after the first keeps a cursor of its own, `<kind>-<hash>.cursor`.
+ Only `<kind>.cursor` was carried, so collection into the new root read the other sources again
+ from their start and landed their records again. Each cursor's `last_seq` now names the new file
+ holding the last record its own source landed. A recorder inside the storage root writes its
+ cursors' `origin` relative to it; a repack writes that origin, and the name of a cursor made from
+ it, as a collector into the new root computes them, or that recorder was read from its start.
+- `asz repack` cuts the records of two files into one only when the files are next to each other in
+ their directory and their headers are the same apart from `seq` and `at`. A file took the header
+ of the first file its records came from, so records of another source or another adapter version
+ landed under the wrong header.
## Retention
@@ -619,6 +657,10 @@
- `candidate` checks for `cmp` and `publish` for `perl` before they change anything, as they do
for every other tool they run. `publish` lists the upload to PyPI with the `pypi` skill among the
steps it prints after the move, between apt and Scoop, as the release guide orders them.
+- The container image owns its storage root, `/asz/data`, as the user it runs as, 65532. The image
+ had no such directory, so Docker made a new named or anonymous volume root's, and the first
+ collection failed with `mkdir /asz/data/_root: permission denied`. A new volume now belongs to the
+ image's user. A bind mount works as before.
## Documentation
diff --git a/docs/en/concepts-and-designs/glossary.md b/docs/en/concepts-and-designs/glossary.md
index 8ac2fe4..ea2587f 100644
--- a/docs/en/concepts-and-designs/glossary.md
+++ b/docs/en/concepts-and-designs/glossary.md
@@ -162,7 +162,8 @@ parts, and its identifiers take role names. It does not always keep the source r
keeps its line number counting from 1, byte offset, size, and the first 12 hexadecimal characters
of the SHA-256 of its bytes. Where a part carries the source's own JSON, such as a call's input,
it keeps those bytes apart from whitespace between tokens (see
-[What data holds](../formats/session-data.md#what-data-holds)). Content the dialect cannot describe
+[What data holds](../formats/session-data.md#what-data-holds)), unless they hold a byte that is not
+valid UTF-8: such JSON is taken out, keeping its size and digest. Content the dialect cannot describe
keeps every one of its bytes in an `unknown` part, so nothing is guessed or dropped. When those
bytes are not valid UTF-8, all of them are kept as base64 (see
[Parts](../formats/session-data.md#parts)). A reader is handed parts, so it never meets a
diff --git a/docs/en/formats/otlp.md b/docs/en/formats/otlp.md
index f6d5ff8..a0ef3c7 100644
--- a/docs/en/formats/otlp.md
+++ b/docs/en/formats/otlp.md
@@ -21,7 +21,13 @@ is a small record, and the same conversation is 306 records.
Landed files and rounds are both write-once, so each is sent once. `push.state` in the storage
root lists what was sent, with the digest each file had; a file is recorded only after the request
-carrying it succeeded, so a failed request leaves it for the next pass.
+carrying it succeeded, so a failed request leaves it for the next pass. A failed request is
+reported once, as its own error, and the files after it are still tried in the same pass, unless
+the receiver asked the sender to slow down, as [Rate](#rate) describes. A file that
+cannot be encoded as a log record never goes in a request: a protocol buffer string must be valid
+UTF-8, and one file that is not would fail every request it was in, on every pass, with the files
+beside it. It is reported on its own, under its name, on every pass. asz never lands such a file
+(see [Reading a record](session-data.md#reading-a-record)), but a writer outside asz can.
`push.state` also names the receivers the files went to, with one `endpoint` line each:
`grpc://host:port`, `grpcs://host:port` with `tls: true`, or the HTTP base URL. A `rejected` line
@@ -90,8 +96,10 @@ which records were rejected. A scenario session with such a file is never remove
Over HTTP, a request counts as taken only when the answer has a 2xx status and a body that is empty
or protobuf: `application/x-protobuf`, as OTLP names it, or `application/protobuf`, whatever the
-parameters. A 2xx answer with a body of any other type is an error. Nothing in the request is
-recorded as sent, so it goes again on the next pass:
+parameters. A 2xx answer with a body of any other type is an error, and so is one whose body does
+not arrive whole, such as a connection that closes before the length the answer promised, and one
+whose body is longer than 1 MiB, which is all the sender reads of an answer. Nothing
+in the request is recorded as sent, so it goes again on the next pass:
```text
otlp: http://127.0.0.1:12800/v1/logs answered 200 OK with text/html, not an OTLP response; nothing is recorded as sent
diff --git a/docs/en/formats/session-data.md b/docs/en/formats/session-data.md
index b80d94b..7464252 100644
--- a/docs/en/formats/session-data.md
+++ b/docs/en/formats/session-data.md
@@ -39,8 +39,8 @@ stops at the first longer line.
| --- | --- |
| `h` | envelope version, `1` |
| `schema` | `sd/1` |
-| `seq` | the landed sequence number, monotonic per session |
-| `at` | when the file was collected |
+| `seq` | the landed sequence number, monotonic per session, in the order files landed |
+| `at` | when the file was collected. A file `asz repack` cut says when the newest file its records came from was collected |
| `kind` | what it was collected from: `transcript`, `agent_meta`, `journal`, `workflow_manifest`, `workflow_script`, `changes`, `execution` or `provider_body`. `otlp_log` and `otlp_span` are reserved for a push transport, and no adapter writes either. |
| `adapter` | how the records were acquired, with its contract version |
| `dialect` | whose schema they were read as. A push receiver and a local reader for one runtime share a dialect and nothing else. |
@@ -202,6 +202,22 @@ which leaves out a field the format does not name, and says `truncated`. Real li
near it. Measured over 423,944 lines of 94 real conversations, the deepest held 16, a model request
carrying a tool's JSON Schema in its data.
+Every landed line is valid UTF-8. JSON sent between systems must be, a reader may refuse a line
+that is not, and a protocol buffer string must be too, so a file holding such a byte could never be
+sent over [OTLP](otlp.md). A JSON decoder can accept a byte that is not valid UTF-8 inside a
+string, so source JSON can bring one, such as a write cut short in the middle of a character. A
+string the writer encodes itself, such as `text` or `label`, holds U+FFFD in its place. A part's
+data that holds such a byte is taken out, whatever its size. The part keeps its place, its kind and
+its identifiers, its `state` is `omitted`, or `truncated` while its text is left, and its `bytes`
+and `sha256` are those of the source JSON it held, as for a payload
+[too large to land](#too-large-to-land). A `changes/1` or `execution/1` record is written again in
+its own format, with U+FFFD in place of the byte, and says `truncated`. Data that is also nested
+too deep is taken out as such data is, above, and says `too_large`. In every case the record's
+[`dropped`](#dropped) gains one entry for the part, so a reader knows the byte was there: `what`
+is `the data of part N`, with N its block, `bytes` the size of the JSON taken out, and `why` is
+`it holds a byte that is not valid UTF-8, which no landed line may hold`. An
+`unknown` part keeps such bytes whole, in base64, as [Parts](#parts) describes.
+
A `time` is written as RFC 3339: a four-digit year, the letter `T`, the time of day to the
second with a fraction of up to nine digits or none, and `Z` or an offset such as `+08:00`. Times
compare as the instants they name, never as text. As text, `2026-01-01T12:00:00.11Z` sorts before
@@ -455,7 +471,9 @@ takes nothing out.
not text, an attachment with no prose, or a record or document with no message. A journal line,
sidecar, manifest, system record or line the asz plugin wrote can land whole this way. The
writer inserts that JSON after `json.Compact`, keeping its bytes apart from whitespace between
- tokens. This keeps each landed record on one line. Whitespace inside strings is unchanged.
+ tokens. This keeps each landed record on one line. Whitespace inside strings is unchanged. JSON
+ that holds a byte that is not valid UTF-8 is taken out instead, as
+ [Reading a record](#reading-a-record) describes.
- Bytes the dialect could not describe: an `unknown` part holds them inside one JSON string,
using base64 when needed, as described below. `Part.Raw` returns the original bytes whatever
escapes that string uses.
@@ -493,8 +511,9 @@ 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 `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
+No built-in adapter writes `omitted` itself, and a part that lands as `omitted` or `truncated` is
+one made smaller to fit the budget or to hold only valid UTF-8, so no part 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
@@ -577,7 +596,8 @@ runs after every call to an MCP server. Measured on Claude Code 2.1.282:
```
`dropped` exists so a loss is stated rather than silent. It lists only what the dialect understood
-and chose not to carry. Everything it did not understand travels as an `unknown` part, so a later
+and chose not to carry, and the data the writer took out because it held a byte that is not valid
+UTF-8 (see [Reading a record](#reading-a-record)). Everything it did not understand travels as an `unknown` part, so a later
version of the dialect can interpret it without collecting again. In an entry, `what` and `why`
are strings and `bytes` is an integer.
diff --git a/docs/en/formats/storage-root.md b/docs/en/formats/storage-root.md
index 8d3678e..b4e86b5 100644
--- a/docs/en/formats/storage-root.md
+++ b/docs/en/formats/storage-root.md
@@ -145,7 +145,12 @@ addresses every record by file and line and binds itself to the file digests, so
that a round references would break every reference. A change of budget applies to new files
only. To bring an existing root under a new budget, `asz repack DEST` re-cuts every file into a
new root, keeping each record's bytes and order, carries the cursors over, and builds the chains
-again on the new files, which describe the same conversation at new positions.
+again on the new files, which describe the same conversation at new positions. It cuts a file by
+what it takes once written, header and closing line included, as the collector does. It cuts
+records of two files into one only when the files are next to each other in their directory and
+their headers are the same apart from `seq` and `at`, so every file still names its own source and
+every record keeps its place. Each new file has the header of the newest file its records
+came from, `at` included, and the files are numbered in the order their records landed.
## Cursors
@@ -186,7 +191,7 @@ state active
| Field | What it is for |
| --- | --- |
| `source` | the source path, relative to the adapter's root, with forward slashes |
-| `origin` | the adapter's root the source is under, with forward slashes: relative to the storage root when the source root lies inside it, so a root copied whole keeps its cursors, and absolute otherwise. Written since 0.5.0 by an adapter that reads more than one root, since two roots can hold one relative path. A cursor without it was written before 0.5.0 and belongs to the root whose file it was read from - the file with its `dev` and `ino` and the bytes of its `tail_sha256`, since t [...]
+| `origin` | the adapter's root the source is under, with forward slashes: relative to the storage root when the source root lies inside it, so a root copied whole keeps its cursors, and absolute otherwise. `asz repack` writes a relative origin again for its new root, which the source root is outside, and renames the cursor named for it to match. Written since 0.5.0 by an adapter that reads more than one root, since two roots can hold one relative path. A cursor without it was written be [...]
| `dev`, `ino` | the device and the inode the file had when it was read, or on Windows its volume serial number and file index. Resuming a cursor compares `ino` alone, and it is what this page calls the file's identity: a change means the file was replaced, or is being reached another way, and the bytes decide which, see below. Claiming a cursor that carries no `origin` compares both, since that decides whether the cursor was read from this file at all and not from another with the same [...]
| `offset` | the bytes consumed, always at the end of a line, and where the next read starts. Compared with the file's size, it says whether anything is new. A size below it means the file was cut short. |
| `ord` | the last source line consumed, so line numbers continue without counting again from byte 0 |
@@ -433,8 +438,11 @@ The store had also outlived its source: it held 48 sessions where the source dir
`storage.retention` says how long an idle conversation keeps what in the root, at two levels on one
clock. A conversation is idle for as long as nothing new has landed in any of its streams, by asz's
-clock: the time is read from the header of its newest landed file or skeleton, never from a
-record's time, a file's time or when a file was removed. Whether it was sent, whether an endpoint
+clock: the time is read from the header of its newest landed file or skeleton, the one with the
+highest sequence, never from a record's time, a file's time or when a file was removed. A root
+`asz repack` wrote has the same clock as the root it read: each file it cuts says when the newest
+file its records came from was collected, and its files are numbered in the order their records
+landed. Whether it was sent, whether an endpoint
is set, and whether its source is still there do not change that. Each setting is `-1`, off, by
default, and only both off keeps everything.
diff --git a/docs/en/guides/command-line-internals.md b/docs/en/guides/command-line-internals.md
index fc6374c..ef220b2 100644
--- a/docs/en/guides/command-line-internals.md
+++ b/docs/en/guides/command-line-internals.md
@@ -203,11 +203,19 @@ status instead of the log.
## repack
Re-cuts every landed file of the storage root, or of one session, into `DEST` so that no file
-exceeds the configured `max_delta_bytes`, except a file of one record larger than it, which is
-carried whole. Then it assembles `DEST`'s chains. Every record keeps its bytes and its order; the
-cursors come along so collection can continue into `DEST`; the index and the chains are rebuilt
-there, because a round addresses records by file and line and the old
-references name positions that no longer exist. `DEST` must be a new or empty directory outside
+exceeds the configured `max_delta_bytes`, header and closing line included, then assembles
+`DEST`'s chains. The one exception is a record larger than the budget: it is carried whole, in a
+file of its own, which is then larger than the budget. Every record keeps its bytes and its order,
+and records of two files are cut into one only when the files are next to each other and their
+headers are the same apart from `seq` and `at`. Every cursor comes along so collection can
+continue into `DEST`, the one a second source of a stream keeps too, such as a second recorder
+directory. Each names the new file that holds the last record its source landed. A cursor whose
+`origin` is relative to the storage root, a recorder inside it, gets the origin and the name a
+collector into `DEST` gives it. The index and the chains are rebuilt there, because a round
+addresses records by file and line and the old references name positions that no longer exist.
+Each new file keeps the time the newest file its records came from was collected, and the files
+are numbered in the order their records landed, so retention finds a session as idle in `DEST` as
+in the root it was repacked from. `DEST` must be a new or empty directory outside
the storage root. A populated destination is refused before any session is written, because its
chains bind to the original landed files. One inside the root is refused too, because every
command would read it as a session of the root. If a repack is interrupted, retry into another
diff --git a/docs/en/setup/container-image.md b/docs/en/setup/container-image.md
index 8698039..91124b8 100644
--- a/docs/en/setup/container-image.md
+++ b/docs/en/setup/container-image.md
@@ -51,6 +51,17 @@ docker run --rm -p 8787:8787 \
The default command, `server`, collects every 10 minutes and serves the page. On Linux, add
`--user "$(id -u):$(id -g)"` so the container can write to `./data`.
+To keep the data in a Docker volume instead of `./data`, name the volume, and add no `--user`. A
+new volume belongs to the user the image runs as:
+
+```sh
+docker run --rm -p 8787:8787 \
+ -v "$HOME/.claude/projects:/claude/projects:ro" \
+ -v "$PWD/asz.yaml:/asz/asz.yaml:ro" \
+ -v asz-data:/asz/data \
+ ghcr.io/apache/skywalking-ai-sessionizer:latest
+```
+
## Run any command
Put the command after the image name:
diff --git a/internal/adapters/claudecodechanges/collect.go b/internal/adapters/claudecodechanges/collect.go
index a7f3ed0..4c3dc90 100644
--- a/internal/adapters/claudecodechanges/collect.go
+++ b/internal/adapters/claudecodechanges/collect.go
@@ -18,14 +18,11 @@
package claudecodechanges
import (
- "crypto/sha256"
- "encoding/hex"
"errors"
"fmt"
"io"
"os"
"path/filepath"
- "strings"
"time"
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecode"
@@ -246,8 +243,7 @@ func cursorFor(dir, root string, src Source) (string, error) {
case held.Source == src.Rel && held.Origin == "" && describes(held, src.Path):
return plain, nil
}
- sum := sha256.Sum256([]byte(root + "\n" + src.Rel))
- return filepath.Join(dir, prefix+"-"+hex.EncodeToString(sum[:])[:12]+".cursor"), nil
+ return filepath.Join(dir, storage.SourceCursorName(prefix, root, src.Rel)), nil
}
// describes reports whether a cursor that names no root was read from this
@@ -277,18 +273,10 @@ func describes(cur *storage.Cursor, path string) bool {
return err == nil && got == cur.TailSHA256
}
-// origin is how this collector's root is written into a cursor, with
-// forward slashes so it reads the same on every pass. A root inside the
-// storage root - a scenario's, which a test copies whole to another place
-// - is written relative to it, so the copy keeps its cursors; any other
-// root is written absolute, since a source directory that moves is a
-// different source.
+// origin is how this collector's root is written into a cursor. The rule
+// lives in storage, because asz repack writes it again for a new root.
func (c *Collector) origin() string {
- root := filepath.Clean(c.SourceRoot)
- if rel, err := filepath.Rel(c.Zone.Root(), root); err == nil && rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
- return filepath.ToSlash(rel)
- }
- return filepath.ToSlash(root)
+ return storage.CursorOrigin(c.Zone.Root(), c.SourceRoot)
}
// collectSource lands one window of one file, and says whether more waits.
diff --git a/internal/export/otlp/client.go b/internal/export/otlp/client.go
index cec6e90..e86e357 100644
--- a/internal/export/otlp/client.go
+++ b/internal/export/otlp/client.go
@@ -250,12 +250,16 @@ func (c *httpClient) ExportMetrics(req *collmetricspb.ExportMetricsServiceReques
return out.GetPartialSuccess().GetRejectedDataPoints(), nil
}
+// maxAnswer is the most of an answer's body the client reads.
+const maxAnswer = 1 << 20
+
// post sends one request as protobuf and returns the answer's body.
//
// A 2xx answer with an empty body is a full success, whatever its type,
// since that is how a proxy in front of a receiver often answers. A 2xx
-// answer with a body is an OTLP response only when its type is protobuf.
-// Anything else is an error, and nothing is recorded as sent. Taking such an
+// answer with a body is an OTLP response only when its type is protobuf, and
+// only when the whole body arrived. Anything else is an error, and nothing is
+// recorded as sent. Taking such an
// answer as success was measured against a real listener: an HTML page
// answered with 200, and a redirect to a login page, both left the files
// marked sent while the receiver stored nothing.
@@ -278,7 +282,9 @@ func (c *httpClient) post(url string, req proto.Message) ([]byte, error) {
return nil, err
}
defer resp.Body.Close()
- answer, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
+ // One byte past the most an answer may be is read, so an answer that
+ // goes on past it is told from one that ends there.
+ answer, readErr := io.ReadAll(io.LimitReader(resp.Body, maxAnswer+1))
if resp.StatusCode == http.StatusTooManyRequests {
t := &Throttled{Detail: url + " answered " + resp.Status}
if secs, err := strconv.Atoi(strings.TrimSpace(resp.Header.Get("Retry-After"))); err == nil && secs > 0 {
@@ -293,6 +299,17 @@ func (c *httpClient) post(url string, req proto.Message) ([]byte, error) {
}
return nil, fmt.Errorf("otlp: %s answered %s: %s", url, resp.Status, strings.TrimSpace(string(snippet)))
}
+ // A success cut off before its body arrived says nothing about what the
+ // receiver took. Read as an empty answer, it marked every file of the
+ // request sent, and none was sent again.
+ if readErr != nil {
+ return nil, fmt.Errorf("otlp: %s answered %s, and its body did not arrive whole: %w; nothing is recorded as sent", url, resp.Status, readErr)
+ }
+ // What comes after the first MiB is never read, so an answer that long
+ // is not known to have arrived whole. No OTLP response comes near it.
+ if len(answer) > maxAnswer {
+ return nil, fmt.Errorf("otlp: %s answered %s with more than %d bytes, longer than any OTLP response; nothing is recorded as sent", url, resp.Status, maxAnswer)
+ }
if len(answer) == 0 {
return nil, nil
}
diff --git a/internal/export/otlp/client_test.go b/internal/export/otlp/client_test.go
index f024d54..00ab54b 100644
--- a/internal/export/otlp/client_test.go
+++ b/internal/export/otlp/client_test.go
@@ -18,6 +18,7 @@
package otlp_test
import (
+ "fmt"
"io"
"net/http"
"net/http/httptest"
@@ -119,6 +120,95 @@ func TestAPageAnsweredWith200IsNotASuccess(t *testing.T) {
}
}
+// A success whose body is cut off before it all arrives is not a success:
+// what the receiver took is unknown. It is an error, whatever the type, and
+// a pass marks nothing sent, so the files go again.
+func TestASuccessCutOffBeforeItsBodyIsNotASuccess(t *testing.T) {
+ for _, media := range []string{"application/x-protobuf", "text/html"} {
+ t.Run(media, func(t *testing.T) {
+ // The answer promises 100 bytes, and the connection closes
+ // before the first of them.
+ c := httpReceiver(t, func(w http.ResponseWriter, r *http.Request) {
+ _, _ = io.Copy(io.Discard, r.Body)
+ conn, buf, err := w.(http.Hijacker).Hijack()
+ if err != nil {
+ t.Error(err)
+ return
+ }
+ _, _ = buf.WriteString("HTTP/1.1 200 OK\r\nContent-Type: " + media + "\r\nContent-Length: 100\r\n\r\n")
+ _ = buf.Flush()
+ _ = conn.Close()
+ })
+ if _, err := c.Export(&collogspb.ExportLogsServiceRequest{}); err == nil || !strings.Contains(err.Error(), "did not arrive whole") {
+ t.Fatalf("logs: a cut answer gave %v; want an error saying its body did not arrive whole", err)
+ }
+ if _, err := c.ExportMetrics(&collmetricspb.ExportMetricsServiceRequest{}); err == nil || !strings.Contains(err.Error(), "did not arrive whole") {
+ t.Fatalf("metrics: a cut answer gave %v; want an error saying its body did not arrive whole", err)
+ }
+
+ z, _ := zoneWithOneSession(t)
+ p := &otlp.Pusher{Zone: z, Client: c, Version: "test", ServiceName: "Claude Code", Endpoint: "http://receiver"}
+ st, err := p.Pass()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if st.Files != 0 || len(st.Errors) == 0 {
+ t.Fatalf("files=%d errors=%v; want nothing marked and the error reported", st.Files, st.Errors)
+ }
+ sent, err := otlp.LoadSent(z.Root())
+ if err != nil {
+ t.Fatal(err)
+ }
+ for rel := range sentPaths(t, z) {
+ if _, ok := sent.Digest(rel); ok {
+ t.Fatalf("%s was recorded as sent", rel)
+ }
+ }
+ })
+ }
+}
+
+// The client reads at most 1 MiB of an answer, far more than any OTLP
+// response holds. An answer cut off just after that much is not a success
+// either, though what arrived is a whole OTLP response, and nor is a whole
+// answer longer than that: what came after the first MiB was never read.
+func TestASuccessLongerThanTheClientReadsIsNotASuccess(t *testing.T) {
+ const limit = 1 << 20
+ logs := &collogspb.ExportLogsServiceResponse{PartialSuccess: &collogspb.ExportLogsPartialSuccess{}}
+ for n := limit - 16; proto.Size(logs) != limit; n++ {
+ logs.PartialSuccess.ErrorMessage = strings.Repeat("x", n)
+ }
+ prefix, err := proto.Marshal(logs)
+ if err != nil || len(prefix) != limit {
+ t.Fatalf("a response of %d bytes, %v", len(prefix), err)
+ }
+ for name, promised := range map[string]int{"cut off after the limit": limit + 100, "longer than the limit": limit + 1} {
+ t.Run(name, func(t *testing.T) {
+ c := httpReceiver(t, func(w http.ResponseWriter, r *http.Request) {
+ _, _ = io.Copy(io.Discard, r.Body)
+ conn, buf, err := w.(http.Hijacker).Hijack()
+ if err != nil {
+ t.Error(err)
+ return
+ }
+ _, _ = fmt.Fprintf(buf, "HTTP/1.1 200 OK\r\nContent-Type: application/x-protobuf\r\nContent-Length: %d\r\n\r\n", promised)
+ _, _ = buf.Write(prefix)
+ if promised == limit+1 {
+ _ = buf.WriteByte(0)
+ }
+ _ = buf.Flush()
+ _ = conn.Close()
+ })
+ if _, err := c.Export(&collogspb.ExportLogsServiceRequest{}); err == nil {
+ t.Fatal("logs: the answer was taken as a success")
+ }
+ if _, err := c.ExportMetrics(&collmetricspb.ExportMetricsServiceRequest{}); err == nil {
+ t.Fatal("metrics: the answer was taken as a success")
+ }
+ })
+ }
+}
+
// A redirect is not followed. A receiver that moved is configured again, and
// a redirect to a login page, followed, would answer 200 with a page.
func TestARedirectIsNotFollowed(t *testing.T) {
diff --git a/internal/export/otlp/push.go b/internal/export/otlp/push.go
index 0657cfa..3169799 100644
--- a/internal/export/otlp/push.go
+++ b/internal/export/otlp/push.go
@@ -691,13 +691,26 @@ func (b *batch) group(session string) *logspb.ResourceLogs {
}
func (b *batch) add(session string, r *logspb.LogRecord, rel, digest string) error {
+ // A record that cannot be encoded fails every request it is in, on every
+ // pass, and every file batched beside it with it. A protocol buffer
+ // string must be valid UTF-8, and a file written outside asz's writer can
+ // hold a byte that is not. So it is reported on its own, and stays out.
+ if _, err := proto.Marshal(r); err != nil {
+ return fmt.Errorf("the file cannot be sent as a log record: %w", err)
+ }
size := int64(len(r.GetBody().GetStringValue()))
// A file that does not fit next to what is already batched goes after
// it. A file larger than the whole budget therefore travels alone: the
// batch before it is sent first, and the file after it starts a new one.
+ // A batch that fails is reported as the request's error, not as this
+ // file's, and this file still goes in the next request: before, it was
+ // left out of the pass under the request's error.
if b.bytes > 0 && b.bytes+size > b.p.BatchBytes {
if err := b.flush(); err != nil {
- return err
+ b.st.Errors = append(b.st.Errors, err)
+ if b.stop != nil {
+ return nil
+ }
}
}
g := b.group(session)
diff --git a/internal/export/otlp/unsendable_test.go b/internal/export/otlp/unsendable_test.go
new file mode 100644
index 0000000..7dceb80
--- /dev/null
+++ b/internal/export/otlp/unsendable_test.go
@@ -0,0 +1,150 @@
+// 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 otlp_test
+
+import (
+ "fmt"
+ "io"
+ "net/http"
+ "path/filepath"
+ "strings"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/export/otlp"
+ "github.com/apache/skywalking-ai-sessionizer/internal/storage"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+)
+
+// zoneOfThree lands three files in one stream of one session, in sequence
+// order, and returns their paths relative to the root. The file at bad holds
+// a byte that is not valid UTF-8, written raw, as the writer never lands it:
+// a writer outside asz could.
+func zoneOfThree(t *testing.T, bad int) (*storage.Zone, []string) {
+ t.Helper()
+ z := storage.NewZone(t.TempDir())
+ dir := z.StreamDir("sess1", "main")
+ var rels []string
+ for seq := 1; seq <= 3; seq++ {
+ path := filepath.Join(dir, storage.LandedName("transcript", storage.Stamp(time.Unix(0, 0)), uint64(seq)))
+ err := storage.WriteAtomic(path, storage.PermLanded, func(w io.Writer) error {
+ hdr := &sessiondata.Header{Seq: uint64(seq), At: "2026-09-04T00:00:00Z", Kind: sessiondata.KindTranscript,
+ Adapter: "test/0", Dialect: "test/1", Src: "sess1.jsonl", Session: "sess1", Stream: "main"}
+ sw, err := sessiondata.NewWriter(w, hdr)
+ if err != nil {
+ return err
+ }
+ text := fmt.Sprintf("record %d", seq)
+ if seq == bad {
+ text = "record \xff"
+ }
+ line := fmt.Sprintf(`{"ord":%d,"off":0,"sha":"a","bytes":1,"time":"2026-09-04T01:00:00Z","parts":[{"k":"text","text":"%s"}]}`, seq, text)
+ if err := sw.WriteRaw([]byte(line)); err != nil {
+ return err
+ }
+ return sw.Close()
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ rel, _ := filepath.Rel(z.Root(), path)
+ rels = append(rels, filepath.ToSlash(rel))
+ }
+ return z, rels
+}
+
+// A file that cannot be encoded as a log record, since a protocol buffer
+// string must be valid UTF-8, fails every request it is in. It is reported on
+// its own, under its name, on every pass, and stays out of the request: the
+// files beside it go, whether they share one request or each has its own.
+func TestAFileThatCannotBeSentIsReportedOnItsOwn(t *testing.T) {
+ for _, protocol := range protocols {
+ for _, batch := range []int64{0, 1} {
+ t.Run(fmt.Sprintf("%s/batch %d", protocol, batch), func(t *testing.T) {
+ z, rels := zoneOfThree(t, 2)
+ rcv := startReceiver(t)
+ p := &otlp.Pusher{Zone: z, Client: clientFor(t, rcv, protocol), Version: "test", ServiceName: "Claude Code",
+ Endpoint: protocol, BatchBytes: batch}
+ for pass := 1; pass <= 2; pass++ {
+ st, err := p.Pass()
+ if err != nil {
+ t.Fatal(err)
+ }
+ want := 2
+ if pass == 2 {
+ want = 0
+ }
+ if st.Files != want || len(st.Errors) != 1 || !strings.HasPrefix(st.Errors[0].Error(), rels[1]+": ") {
+ t.Fatalf("pass %d: files=%d errors=%v; want %d sent and one error naming %s", pass, st.Files, st.Errors, want, rels[1])
+ }
+ }
+ _, logs := parseRequests(rcv.Requests())
+ if len(logs) != 2 || !strings.Contains(logs[0].body, "record 1") || !strings.Contains(logs[1].body, "record 3") {
+ t.Fatalf("the receiver holds %d records; want the first and the third file", len(logs))
+ }
+ sent, err := otlp.LoadSent(z.Root())
+ if err != nil {
+ t.Fatal(err)
+ }
+ for i, rel := range rels {
+ if _, ok := sent.Digest(rel); ok != (i != 1) {
+ t.Fatalf("%s recorded sent: %v", rel, ok)
+ }
+ }
+ })
+ }
+ }
+}
+
+// A request that fails is reported as the request's error, and the file
+// that was being batched when it went goes in the next request of the same
+// pass. It was left out of the pass, under the failed request's error with
+// its own name on it.
+func TestAFailedRequestDoesNotSkipTheFileAfterIt(t *testing.T) {
+ var requests atomic.Int32
+ c := httpReceiver(t, func(w http.ResponseWriter, r *http.Request) {
+ _, _ = io.Copy(io.Discard, r.Body)
+ if requests.Add(1) == 1 {
+ http.Error(w, "busy", http.StatusInternalServerError)
+ return
+ }
+ w.WriteHeader(http.StatusOK)
+ })
+ z, rels := zoneOfThree(t, 0)
+ p := &otlp.Pusher{Zone: z, Client: c, Version: "test", ServiceName: "Claude Code", Endpoint: "http://receiver", BatchBytes: 1}
+ st, err := p.Pass()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if st.Files != 2 || st.Requests != 3 || len(st.Errors) != 1 || strings.HasPrefix(st.Errors[0].Error(), rels[1]) {
+ t.Fatalf("files=%d requests=%d errors=%v; want the second and third file sent, and the error not named for the second", st.Files, st.Requests, st.Errors)
+ }
+ sent, err := otlp.LoadSent(z.Root())
+ if err != nil {
+ t.Fatal(err)
+ }
+ for i, rel := range rels {
+ if _, ok := sent.Digest(rel); ok != (i != 0) {
+ t.Fatalf("%s recorded sent: %v", rel, ok)
+ }
+ }
+ if st, err := p.Pass(); err != nil || st.Files != 1 || len(st.Errors) != 0 {
+ t.Fatalf("the next pass: files=%d errors=%v, %v; want the first file sent", st.Files, st.Errors, err)
+ }
+}
diff --git a/internal/repack/repack.go b/internal/repack/repack.go
index 00fbe9c..bba0579 100644
--- a/internal/repack/repack.go
+++ b/internal/repack/repack.go
@@ -26,11 +26,15 @@
package repack
import (
+ "encoding/json"
"errors"
"fmt"
"io"
+ "io/fs"
+ "math"
"os"
"path/filepath"
+ "sort"
"strings"
"time"
@@ -46,13 +50,38 @@ type Stats struct {
Bytes int64
}
-// slot is one landing directory and file prefix: a stream's transcripts, a
-// stream's sidecars, a run's journal, manifest or script. Records are re-cut
-// within a slot and never across slots.
+// slot is a run of files of one landing directory and file prefix, next to
+// each other in sequence order, whose headers are the same apart from the
+// sequence and the time: a stream's transcripts, a stream's sidecars, a run's
+// journal, manifest or script. Records are re-cut within a slot and never
+// across slots.
+//
+// A file's header says where its records came from. Two sources can write
+// the files of one directory and prefix, such as two recorder directories,
+// each behind its own cursor, and an upgrade can change an adapter's
+// version. Cut together, their records would land under the first file's
+// header. A header that changes and changes back starts a new slot each
+// time: cut together, the two runs with the first header would put the
+// records between them after both, and a provider body before a body it
+// copies.
type slot struct {
rel string // directory relative to the session directory
prefix string
+ id string // the header without its sequence and its time
files []storage.LandedFile
+ hdrs []sessiondata.Header // each file's header
+ outs []*output
+}
+
+// output is one file a repack writes, planned before any is written: how
+// many records it takes, in slot order, from where the file before it ended,
+// and which file of the slot its last record came from. It carries that
+// file's header, and so the time that file was collected.
+type output struct {
+ records int
+ last int // the index in the slot's files
+ landed uint64 // the sequence that file had in the source
+ seq uint64
}
// CheckSource refuses a root with a session retention removed files of, before
@@ -127,10 +156,16 @@ func CheckDestination(root string) error {
}
// Session re-cuts every landed file of session from src into dst so that no
-// file exceeds budget bytes, except one holding a single record larger than
-// that. Records keep their bytes and their order; the session's cursors are
-// carried over so a collector can continue into dst; the index and the chain
-// are not copied, because both are derived and are rebuilt on the new files.
+// file exceeds budget bytes, header and closing line included, except one
+// holding a single record larger than that. Records keep their bytes and their
+// order; the session's cursors are carried over so a collector can continue
+// into dst; the index and the chain are not copied, because both are derived
+// and are rebuilt on the new files.
+//
+// The new files keep the session's idle clock, which retention reads from the
+// header of the file with the highest sequence. Each file says when the newest
+// file its records came from was collected, and the files are numbered in the
+// order their records landed.
func Session(src, dst *storage.Zone, session string, budget int64, now time.Time) (*Stats, error) {
if budget <= 0 {
return nil, errors.New("repack: the budget must be positive")
@@ -175,133 +210,229 @@ func Session(src, dst *storage.Zone, session string, budget int64, now time.Time
return nil, fmt.Errorf("repack: destination session must be new: %w", err)
}
- // Group by slot, keeping slots in the order their first file landed and
- // files within a slot in sequence order, which is source order.
- var slots []*slot
- byKey := map[string]*slot{}
- for _, lf := range files {
- rel, err := filepath.Rel(srcDir, filepath.Dir(lf.Path))
- if err != nil {
+ slots, err := group(srcDir, files)
+ if err != nil {
+ return nil, err
+ }
+ // Every file is planned before any is written, because a file's
+ // sequence is its place among the files of every slot. landedIn maps each
+ // source file to the new file its last record goes into, which is where
+ // a cursor that named the source file points now.
+ landedIn := map[uint64]*output{}
+ for _, s := range slots {
+ if err := plan(s, budget, landedIn); err != nil {
return nil, err
}
- prefix := strings.SplitN(filepath.Base(lf.Path), "-", 2)[0]
- key := rel + "/" + prefix
- s, ok := byKey[key]
- if !ok {
- s = &slot{rel: rel, prefix: prefix}
- byKey[key] = s
- slots = append(slots, s)
- }
- s.files = append(s.files, lf)
}
-
state, err := storage.LoadSessionState(src.SessionStatePath(session), session)
if err != nil {
return nil, err
}
state.NextSeq = 1
+ number(slots, state)
st := &Stats{FilesIn: len(files)}
for _, s := range slots {
- last, err := repackSlot(s, filepath.Join(dstDir, s.rel), budget, state, st)
- if err != nil {
- return nil, err
- }
- if err := carryCursor(filepath.Join(srcDir, s.rel, s.prefix+".cursor"),
- filepath.Join(dstDir, s.rel, s.prefix+".cursor"), last, now); err != nil {
+ if err := write(s, filepath.Join(dstDir, s.rel), st); err != nil {
return nil, err
}
}
+ if err := carryCursors(src, dst, session, landedIn, now); err != nil {
+ return nil, err
+ }
if err := state.Save(dst.SessionStatePath(session), now); err != nil {
return nil, err
}
return st, nil
}
+// group puts every landed file in its slot, keeping slots in the order their
+// first file landed and files within a slot in sequence order, which is
+// source order. A file joins the slot of the file before it in its directory
+// and prefix only when their headers are the same.
+func group(srcDir string, files []storage.LandedFile) ([]*slot, error) {
+ var slots []*slot
+ current := map[string]*slot{}
+ for _, lf := range files {
+ rel, err := filepath.Rel(srcDir, filepath.Dir(lf.Path))
+ if err != nil {
+ return nil, err
+ }
+ hdr, err := readHeader(lf.Path)
+ if err != nil {
+ return nil, err
+ }
+ same := hdr
+ same.Seq, same.At = 0, ""
+ id, err := json.Marshal(same)
+ if err != nil {
+ return nil, err
+ }
+ prefix := strings.SplitN(filepath.Base(lf.Path), "-", 2)[0]
+ key := rel + "\n" + prefix
+ s := current[key]
+ if s == nil || s.id != string(id) {
+ s = &slot{rel: rel, prefix: prefix, id: string(id)}
+ current[key] = s
+ slots = append(slots, s)
+ }
+ s.files = append(s.files, lf)
+ s.hdrs = append(s.hdrs, hdr)
+ }
+ return slots, nil
+}
+
+// plan cuts a slot's records into files by what each file takes once
+// written, its header and closing line included, as a collector cuts its
+// files. The header is measured at its longest, since a file's sequence is
+// not known until every slot is planned and its time is that of the newest
+// file it takes records from. landedIn gets, for each source file, the new
+// file its last record goes into.
+func plan(s *slot, budget int64, landedIn map[uint64]*output) error {
+ longest := s.hdrs[0]
+ longest.Seq = math.MaxUint64
+ for _, h := range s.hdrs {
+ if len(h.At) > len(longest.At) {
+ longest.At = h.At
+ }
+ }
+ var cut *sessiondata.Cut
+ var cur *output
+ for i, lf := range s.files {
+ err := eachLine(lf.Path, func(line []byte) error {
+ if cur == nil || !cut.TakeRaw(line) {
+ c, err := sessiondata.NewCut(&longest, budget)
+ if err != nil {
+ return err
+ }
+ cut, cur = c, &output{}
+ cut.TakeRaw(line)
+ s.outs = append(s.outs, cur)
+ }
+ cur.records++
+ cur.last, cur.landed = i, lf.Seq
+ landedIn[lf.Seq] = cur
+ return nil
+ })
+ if err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+// number gives every planned file its sequence in the order the collector
+// landed its records: by the source sequence of the file its last record
+// came from. The file holding the newest records then has the highest
+// sequence and their time, which is where retention reads how long the
+// session has been idle. Within a slot the files keep their records' order.
+func number(slots []*slot, state *storage.SessionState) {
+ var all []*output
+ for _, s := range slots {
+ all = append(all, s.outs...)
+ }
+ sort.SliceStable(all, func(i, j int) bool { return all[i].landed < all[j].landed })
+ for _, o := range all {
+ o.seq = state.Take()
+ }
+}
+
// out is one landed file being written.
type out struct {
tmp *os.File
w *sessiondata.Writer
final string
- size int64
- n int
}
-func repackSlot(s *slot, outDir string, budget int64, state *storage.SessionState, st *Stats) (uint64, error) {
+// write writes a slot's planned files.
+func write(s *slot, outDir string, st *Stats) (err error) {
if err := os.MkdirAll(outDir, 0o755); err != nil {
- return 0, err
+ return err
}
var cur *out
- var last uint64
- finish := func() error {
- if cur == nil {
- return nil
- }
- err := closeOut(cur)
- cur = nil
- if err == nil {
- st.FilesOut++
+ k, n := 0, 0
+ defer func() {
+ if err != nil && cur != nil {
+ cur.tmp.Close()
+ os.Remove(cur.tmp.Name())
}
- return err
- }
+ }()
for _, lf := range s.files {
- f, err := sessiondata.Open(lf.Path)
- if err != nil {
- return 0, err
- }
- r, err := sessiondata.NewReader(f)
- if err != nil {
- f.Close()
- return 0, fmt.Errorf("repack: %s: %w", lf.Path, err)
- }
- hdr := r.Header()
- for {
- line, err := r.NextRaw()
- if errors.Is(err, io.EOF) {
- break
- }
- if err != nil {
- f.Close()
- return 0, fmt.Errorf("repack: %s: %w", lf.Path, err)
- }
- need := int64(len(line) + 1)
- // Cut before a record that would push the file past the budget,
- // unless the file is empty: a single record larger than the budget
- // is landed whole, as the collector does.
- if cur != nil && cur.n > 0 && cur.size+need > budget {
- if err := finish(); err != nil {
- f.Close()
- return 0, err
- }
- }
+ err := eachLine(lf.Path, func(line []byte) error {
if cur == nil {
- seq := state.Take()
- last = seq
- o, err := openOut(outDir, s.prefix, hdr, seq)
+ hdr := s.hdrs[s.outs[k].last]
+ hdr.Seq = s.outs[k].seq
+ o, err := openOut(outDir, s.prefix, hdr)
if err != nil {
- f.Close()
- return 0, err
+ return err
}
cur = o
}
if err := cur.w.WriteRaw(line); err != nil {
- f.Close()
- return 0, err
+ return err
}
- cur.size += need
- cur.n++
st.Records++
st.Bytes += int64(len(line))
+ if n++; n == s.outs[k].records {
+ o := cur
+ cur = nil
+ if err := closeOut(o); err != nil {
+ return err
+ }
+ st.FilesOut++
+ k, n = k+1, 0
+ }
+ return nil
+ })
+ if err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+// readHeader reads a landed file's header.
+func readHeader(path string) (sessiondata.Header, error) {
+ f, err := sessiondata.Open(path)
+ if err != nil {
+ return sessiondata.Header{}, err
+ }
+ defer f.Close()
+ r, err := sessiondata.NewReader(f)
+ if err != nil {
+ return sessiondata.Header{}, fmt.Errorf("repack: %s: %w", path, err)
+ }
+ return r.Header(), nil
+}
+
+// eachLine hands every record line of a landed file to fn, as it landed.
+func eachLine(path string, fn func(line []byte) error) error {
+ f, err := sessiondata.Open(path)
+ if err != nil {
+ return err
+ }
+ defer f.Close()
+ r, err := sessiondata.NewReader(f)
+ if err != nil {
+ return fmt.Errorf("repack: %s: %w", path, err)
+ }
+ for {
+ line, err := r.NextRaw()
+ if errors.Is(err, io.EOF) {
+ return nil
+ }
+ if err != nil {
+ return fmt.Errorf("repack: %s: %w", path, err)
+ }
+ if err := fn(line); err != nil {
+ return err
}
- f.Close()
}
- return last, finish()
}
-// openOut starts a landed file under a new sequence, with the header of the
-// file its first record came from. Everything in that header is constant for
-// the slot except the sequence.
-func openOut(dir, prefix string, hdr sessiondata.Header, seq uint64) (*out, error) {
- hdr.Seq = seq
+// openOut starts a landed file under the header given, the header of the
+// file its last record came from with the new sequence. Its name carries
+// that file's time too.
+func openOut(dir, prefix string, hdr sessiondata.Header) (*out, error) {
at, err := time.Parse(time.RFC3339Nano, hdr.At)
if err != nil {
at = time.Now()
@@ -316,7 +447,7 @@ func openOut(dir, prefix string, hdr sessiondata.Header, seq uint64) (*out, erro
os.Remove(tmp.Name())
return nil, err
}
- return &out{tmp: tmp, w: w, final: filepath.Join(dir, storage.LandedName(prefix, storage.Stamp(at), seq))}, nil
+ return &out{tmp: tmp, w: w, final: filepath.Join(dir, storage.LandedName(prefix, storage.Stamp(at), hdr.Seq))}, nil
}
// closeOut finishes a landed file the way the collector does: closing line,
@@ -348,16 +479,55 @@ func closeOut(o *out) (err error) {
return nil
}
-// carryCursor copies a slot's cursor into the new root with its last landed
-// sequence updated, so a collector pointed at the new root continues from
-// where the old one stopped instead of landing everything again. The rest of
-// the cursor describes the source file, which the repack did not touch.
-func carryCursor(srcPath, dstPath string, last uint64, now time.Time) error {
+// carryCursors copies every cursor of the session into the new root, so a
+// collector pointed at the new root continues each source from where the old
+// one stopped instead of landing it again. A directory holds one cursor per
+// source: <prefix>.cursor for the first, and <prefix>-<hash>.cursor for each
+// other source of the same files, such as a second recorder directory. Each
+// names the new file holding the last record of the file its source landed
+// last.
+//
+// An origin relative to the storage root names a source root inside it. A
+// collector into the new root writes that root as it writes any root
+// outside, and names the source's own cursor by it, so the origin and that
+// name are written again as it computes them. Left as they were, the
+// collector found no cursor of its own and read the source from its start.
+func carryCursors(src, dst *storage.Zone, session string, landedIn map[uint64]*output, now time.Time) error {
+ srcDir, dstDir := src.SessionDir(session), dst.SessionDir(session)
+ return filepath.WalkDir(srcDir, func(path string, d fs.DirEntry, err error) error {
+ if err != nil || d.IsDir() || !strings.HasSuffix(d.Name(), ".cursor") {
+ return err
+ }
+ rel, err := filepath.Rel(srcDir, filepath.Dir(path))
+ if err != nil {
+ return err
+ }
+ cur, err := storage.LoadCursor(path, storage.CursorAppend, "")
+ if err != nil {
+ return err
+ }
+ var last uint64
+ if o := landedIn[cur.LastSeq]; o != nil {
+ last = o.seq
+ }
+ name, origin := d.Name(), cur.Origin
+ if origin != "" && !filepath.IsAbs(filepath.FromSlash(origin)) {
+ origin = storage.CursorOrigin(dst.Root(), filepath.Join(src.Root(), filepath.FromSlash(cur.Origin)))
+ prefix := strings.SplitN(strings.TrimSuffix(name, ".cursor"), "-", 2)[0]
+ if name == storage.SourceCursorName(prefix, cur.Origin, cur.Source) {
+ name = storage.SourceCursorName(prefix, origin, cur.Source)
+ }
+ }
+ return carryCursor(path, filepath.Join(dstDir, rel, name), last, origin, now)
+ })
+}
+
+// carryCursor copies one cursor with its last landed sequence, its origin
+// and its time updated. The rest of the cursor describes the source file,
+// which the repack did not touch.
+func carryCursor(srcPath, dstPath string, last uint64, origin string, now time.Time) error {
data, err := os.ReadFile(srcPath)
if err != nil {
- if os.IsNotExist(err) {
- return nil
- }
return err
}
lines := strings.Split(strings.TrimRight(string(data), "\n"), "\n")
@@ -365,10 +535,15 @@ func carryCursor(srcPath, dstPath string, last uint64, now time.Time) error {
switch {
case strings.HasPrefix(l, "last_seq "):
lines[i] = fmt.Sprintf("last_seq %d", last)
+ case strings.HasPrefix(l, "origin "):
+ lines[i] = "origin " + origin
case strings.HasPrefix(l, "updated_at "):
lines[i] = "updated_at " + now.UTC().Format(time.RFC3339Nano)
}
}
+ if err := os.MkdirAll(filepath.Dir(dstPath), 0o755); err != nil {
+ return err
+ }
return storage.WriteAtomic(dstPath, storage.PermState, func(w io.Writer) error {
_, err := io.WriteString(w, strings.Join(lines, "\n")+"\n")
return err
diff --git a/internal/storage/cursor.go b/internal/storage/cursor.go
index e2b4f81..09bb25c 100644
--- a/internal/storage/cursor.go
+++ b/internal/storage/cursor.go
@@ -19,9 +19,12 @@ package storage
import (
"bufio"
+ "crypto/sha256"
+ "encoding/hex"
"fmt"
"io"
"os"
+ "path/filepath"
"strconv"
"strings"
"time"
@@ -204,3 +207,25 @@ func (c *Cursor) Save(path string, now time.Time) error {
return bw.Flush()
})
}
+
+// CursorOrigin is how a source root is written into a cursor's origin, with
+// forward slashes so it reads the same on every pass. A source root inside
+// the storage root, as a scenario's is, is written relative to it, so a copy
+// of the whole root keeps its cursors. Any other root is written as it is,
+// since a source directory that moves is a different source.
+func CursorOrigin(storageRoot, sourceRoot string) string {
+ root := filepath.Clean(sourceRoot)
+ if rel, err := filepath.Rel(storageRoot, root); err == nil && rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
+ return filepath.ToSlash(rel)
+ }
+ return filepath.ToSlash(root)
+}
+
+// SourceCursorName is the name of the cursor a source keeps when the plain
+// one, <prefix>.cursor, belongs to another source of the same files. It is
+// named for the source's origin and its path, so it is the same on every
+// pass.
+func SourceCursorName(prefix, origin, source string) string {
+ sum := sha256.Sum256([]byte(origin + "\n" + source))
+ return prefix + "-" + hex.EncodeToString(sum[:])[:12] + ".cursor"
+}
diff --git a/pkg/sessiondata/fit.go b/pkg/sessiondata/fit.go
index f0a4188..acd8411 100644
--- a/pkg/sessiondata/fit.go
+++ b/pkg/sessiondata/fit.go
@@ -22,6 +22,8 @@ import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
+ "fmt"
+ "unicode/utf8"
"github.com/apache/skywalking-ai-sessionizer/pkg/changes"
"github.com/apache/skywalking-ai-sessionizer/pkg/execution"
@@ -59,10 +61,12 @@ const endLineMax = len(`{"t":"end","records":,"digest":""}`) + 20 + 64 + 1
// A record that is still too large with every payload taken out lands as it
// is: its identifiers, label and part list remain.
//
-// Data nested deeper than a line may hold, MaxDataDepth, is taken out the
-// same way first, under any limit: a reader may refuse the line otherwise. A
-// change or execution record is written again in its own format instead,
-// which leaves out the fields its format does not name.
+// Data a reader may refuse is taken out first, under any limit: data nested
+// deeper than a line may hold, MaxDataDepth, the same way, and data holding a
+// byte that is not valid UTF-8, saying omitted and naming it in the record's
+// dropped list. A change or execution record is written again in its own
+// format instead, which leaves out the fields its format does not name and
+// holds U+FFFD in place of such a byte.
func Fit(r *Record, limit int) (int, error) {
_, n, err := fit(r, limit, true)
return n, err
@@ -105,42 +109,88 @@ func Depth(data []byte) int {
return deepest
}
-// level takes out the data of every part nested deeper than MaxDataDepth,
-// in place, and adds the parts it changed to changed. It changes r only when
-// the record then encodes, so a record that cannot be written stays as the
-// caller gave it.
+// WhyNotUTF8 is the reason a record's dropped list gives for the data of a
+// part taken out because it held a byte that is not valid UTF-8.
+const WhyNotUTF8 = "it holds a byte that is not valid UTF-8, which no landed line may hold"
+
+// level takes out the data of every part a reader may refuse, in place, and
+// adds the parts it changed to changed. It changes r only when the record
+// then encodes, so a record that cannot be written stays as the caller gave
+// it.
+//
+// Data nested deeper than MaxDataDepth is taken out as a payload too large to
+// carry is. Data holding a byte that is not valid UTF-8 is taken out too: JSON
+// sent between systems must be UTF-8, and a protocol buffer string must be,
+// so a landed file holding such a byte could never be sent over OTLP. The
+// JSON decoder accepts such a byte inside a string, so source JSON can bring
+// one. The byte says nothing about size, so its part says omitted rather than
+// too_large, and the record's dropped list says why. Data that is too deep as
+// well says too_large, as too deep data does, and the dropped list still
+// names the byte.
func level(r *Record, changed map[int]bool) error {
parts := append([]Part(nil), r.Parts...)
- deep := map[int]bool{}
+ taken := map[int]bool{}
+ var dropped []Drop
for i := range parts {
- if len(parts[i].Data) > 0 && Depth(parts[i].Data) > MaxDataDepth {
- if !reencode(&parts[i]) {
- omit(&parts[i])
- }
- deep[i] = true
+ p := &parts[i]
+ if len(p.Data) == 0 {
+ continue
}
+ deep, bad := Depth(p.Data) > MaxDataDepth, !utf8.Valid(p.Data)
+ if !deep && !bad {
+ continue
+ }
+ if bad {
+ dropped = append(dropped, Drop{What: fmt.Sprintf("the data of part %d", i), Bytes: len(carried(p)), Why: WhyNotUTF8})
+ }
+ switch {
+ case reencode(p):
+ case deep:
+ omit(p)
+ default:
+ leaveOut(p)
+ }
+ taken[i] = true
}
- if len(deep) == 0 {
+ if len(taken) == 0 {
return nil
}
leveled := *r
leveled.Parts = parts
+ if len(dropped) > 0 {
+ leveled.Dropped = append(append([]Drop(nil), r.Dropped...), dropped...)
+ }
if _, err := encodeRecord(&leveled); err != nil {
return err
}
copy(r.Parts, parts)
- for i := range deep {
+ r.Dropped = leveled.Dropped
+ for i := range taken {
changed[i] = true
}
return nil
}
+// leaveOut takes out a part's data that no landed line may hold, and says
+// omitted, or truncated while its text is left, with the size and the
+// digest of what it held.
+func leaveOut(p *Part) {
+ remember(p)
+ p.Data, p.Encoding = nil, ""
+ p.State = model.ContentOmitted
+ if p.Text != "" && p.Kind != PartUnknown {
+ p.State = model.ContentTruncated
+ }
+}
+
// reencode writes a change record or an execution record again in its own
// format, as its decoder reads it, so a field its format does not name, nested
// however deep, goes and the record stays: neither is ever taken out whole,
-// since a step joins to it by its tool call. Its part says truncated, with the
-// size and the digest of what it held. It reports false for any other data,
-// and for a record its format cannot write shallow enough.
+// since a step joins to it by its tool call. The decoder reads a byte that is
+// not valid UTF-8 as U+FFFD, so the record written again is valid UTF-8. Its
+// part says truncated, with the size and the digest of what it held. It
+// reports false for any other data, and for a record its format cannot write
+// shallow enough.
func reencode(p *Part) bool {
if p.Kind != PartData {
return false
@@ -345,10 +395,26 @@ func (c *Cut) Take(r *Record) (bool, error) {
if err != nil {
return false, err
}
- need := len(line) + 1
+ return c.take(len(line) + 1), nil
+}
+
+// TakeRaw reports whether a record line that is already encoded, without its
+// newline, goes into this file, by the rule Take follows. The line is never
+// changed, so one larger than the budget is taken only as the first record
+// of a file, and that file is larger than the budget.
+func (c *Cut) TakeRaw(line []byte) bool {
+ if c.limit <= 0 {
+ return true
+ }
+ return c.take(len(line) + 1)
+}
+
+// take counts a line of need bytes, its newline included, into the file, or
+// reports that the line starts the next file.
+func (c *Cut) take(need int) bool {
if c.room < c.limit && need > c.room {
- return false, nil
+ return false
}
c.room -= need
- return true, nil
+ return true
}
diff --git a/pkg/sessiondata/fit_test.go b/pkg/sessiondata/fit_test.go
index eb8c1d4..1e6f0a9 100644
--- a/pkg/sessiondata/fit_test.go
+++ b/pkg/sessiondata/fit_test.go
@@ -23,8 +23,10 @@ import (
"encoding/hex"
"encoding/json"
"fmt"
+ "reflect"
"strings"
"testing"
+ "unicode/utf8"
"github.com/apache/skywalking-ai-sessionizer/pkg/changes"
"github.com/apache/skywalking-ai-sessionizer/pkg/execution"
@@ -364,6 +366,61 @@ func TestCutPacksByEncodedSize(t *testing.T) {
}
}
+// A line that is already encoded is cut by the rule a record is: lines fill
+// files of at most the budget, header and closing line included, and the
+// first line of a file is always taken, unchanged, even when it alone is
+// larger than the budget.
+func TestCutTakesARawLineByTheSameRule(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 lines [][]byte
+ for i := 0; i < 100; i++ {
+ line, err := encodeRecord(&Record{Ord: uint64(i + 1), Sha: "x", Bytes: 2, ID: fmt.Sprintf("r%03d", i), Parts: []Part{{Kind: PartText, Text: "{}"}}})
+ if err != nil {
+ t.Fatal(err)
+ }
+ lines = append(lines, line)
+ }
+ files := 0
+ for len(lines) > 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)
+ }
+ n := 0
+ for n < len(lines) && cut.TakeRaw(lines[n]) {
+ if err := w.WriteRaw(lines[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 lines in %d bytes, budget %d", files, n, buf.Len(), budget)
+ }
+ lines = lines[n:]
+ files++
+ }
+ if files < 2 {
+ t.Fatalf("100 lines went into %d file(s) of %d bytes", files, budget)
+ }
+ cut, err := NewCut(h, budget)
+ if err != nil {
+ t.Fatal(err)
+ }
+ big := bytes.Repeat([]byte("a"), 10_000)
+ if !cut.TakeRaw(big) || cut.TakeRaw([]byte("{}")) {
+ t.Fatal("a large first line was not taken alone")
+ }
+}
+
// 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.
@@ -452,3 +509,80 @@ func TestADeepFieldLeavesAChangeOrExecutionRecordInPlace(t *testing.T) {
})
}
}
+
+// A change record or an execution record holding a byte that is not valid
+// UTF-8 is never taken out whole either. It is written again in its own
+// format, which reads the byte as U+FFFD, and says truncated, with the size
+// and the digest of what it held. The record's dropped list says why.
+func TestABadByteLeavesAChangeOrExecutionRecordInPlace(t *testing.T) {
+ change := strings.Replace(string(changeRecord(t, 1, 2)), `"f0.go"`, "\"f0\xff.go\"", 1)
+ observed := "{\"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\",\"cwd\":\"a\xffb\"}"
+ for name, data := range map[string]json.RawMessage{
+ "change": json.RawMessage(change),
+ "execution": json.RawMessage(observed),
+ } {
+ t.Run(name, func(t *testing.T) {
+ r := &Record{Ord: 1, Tool: "toolu_1", Parts: []Part{{Kind: PartData, Data: data}}}
+ line, n, err := fit(r, 0, true)
+ if err != nil || n != 1 {
+ t.Fatalf("fit changed %d part(s), %v", n, err)
+ }
+ if !utf8.Valid(line) {
+ t.Fatalf("the line holds a byte that is not valid UTF-8: %q", line)
+ }
+ p := r.Parts[0]
+ if p.State != model.ContentTruncated || p.SHA256 != hexDigest(data) || p.Bytes != len(data) {
+ t.Fatalf("the record is %s with digest %q and %d bytes: %s", p.State, p.SHA256, p.Bytes, p.Data)
+ }
+ if want := []Drop{{What: "the data of part 0", Bytes: len(data), Why: WhyNotUTF8}}; !reflect.DeepEqual(r.Dropped, want) {
+ t.Fatalf("the record drops %+v; want %+v", r.Dropped, want)
+ }
+ if name == "change" {
+ if rec, ok := changes.Decode(p.Data); !ok || rec.ID != "c1" || len(rec.Changes) != 1 || rec.Changes[0].Path != "f0\xef\xbf\xbd.go" {
+ t.Fatalf("the change record lost its identity or files: %s", p.Data)
+ }
+ } else if rec, ok := execution.Decode(p.Data); !ok || rec.ID != "e1" || rec.Tool != "toolu_1" || rec.Outcome != "returned" {
+ t.Fatalf("the execution record lost its identity: %s", p.Data)
+ }
+ })
+ }
+}
+
+// A byte that is not valid UTF-8 is named in the record's dropped list
+// whatever else is wrong with its data. Data nested too deep as well is taken
+// out as too deep data is, too_large, and a change or execution record with a
+// deep field as well is written again in its own format, truncated; each
+// record still says the byte was there.
+func TestABadByteIsNamedWhenItsDataIsAlsoTooDeep(t *testing.T) {
+ nest := func(inner string) string { return strings.Repeat("[", 300) + inner + strings.Repeat("]", 300) }
+ change := strings.Replace(string(changeRecord(t, 1, 2)), `"f0.go"`, "\"f0\xff.go\"", 1)
+ change = strings.TrimSuffix(change, "}") + `,"extra":` + nest("0") + `}`
+ observed := "{\"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\",\"cwd\":\"a\xffb\",\"extra\":" +
+ nest("0") + "}"
+ for name, c := range map[string]struct {
+ data string
+ state string
+ }{
+ "data": {nest("\"\xff\""), model.ContentTooLarge},
+ "change": {change, model.ContentTruncated},
+ "execution": {observed, model.ContentTruncated},
+ } {
+ t.Run(name, func(t *testing.T) {
+ data := json.RawMessage(c.data)
+ r := &Record{Ord: 1, Tool: "toolu_1", Parts: []Part{{Kind: PartData, Data: data}}}
+ line, n, err := fit(r, 0, true)
+ if err != nil || n != 1 {
+ t.Fatalf("fit changed %d part(s), %v", n, err)
+ }
+ if !utf8.Valid(line) || Depth(line) > MaxLineDepth {
+ t.Fatalf("the line is not valid UTF-8 or nests too deep: %q", line)
+ }
+ if p := r.Parts[0]; p.State != c.state || p.SHA256 != hexDigest(data) || p.Bytes != len(data) {
+ t.Fatalf("the part is %s with digest %q and %d bytes", p.State, p.SHA256, p.Bytes)
+ }
+ if want := []Drop{{What: "the data of part 0", Bytes: len(data), Why: WhyNotUTF8}}; !reflect.DeepEqual(r.Dropped, want) {
+ t.Fatalf("the record drops %+v; want %+v", r.Dropped, want)
+ }
+ })
+ }
+}
diff --git a/pkg/sessiondata/writer_test.go b/pkg/sessiondata/writer_test.go
index 63eba9c..10f47f8 100644
--- a/pkg/sessiondata/writer_test.go
+++ b/pkg/sessiondata/writer_test.go
@@ -28,6 +28,7 @@ import (
"reflect"
"strings"
"testing"
+ "unicode/utf8"
"github.com/apache/skywalking-ai-sessionizer/pkg/model"
"github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
@@ -46,7 +47,6 @@ func TestWriterDataBytesDoNotDependOnJSONEngine(t *testing.T) {
{"escape spelling", `{"s":"\u003C\u003e\u0026\/\u2028\u2029\u0001\u00e9\u00E9\ud800\ud83d\ude00"}`, `{"s":"\u003C\u003e\u0026\/\u2028\u2029\u0001\u00e9\u00E9\ud800\ud83d\ude00"}`},
{"number spelling", `[1E400,-0,1.0,1e+09,9007199254740993]`, `[1E400,-0,1.0,1e+09,9007199254740993]`},
{"duplicate keys and order", `{"z":1,"a":2,"z":3}`, `{"z":1,"a":2,"z":3}`},
- {"invalid utf8", "{\"text\":\"\xff\xfe\xc0\xaf\"}", "{\"text\":\"\xff\xfe\xc0\xaf\"}"},
{"compact whitespace", " \r\n { \"x\" : [ true , null , \" a b \\n\\t\" ] }\t", `{"x":[true,null," a b \n\t"]}`},
{"string", `"<text>&\u2028"`, `"<text>&\u2028"`},
{"null", `null`, `null`},
@@ -303,19 +303,57 @@ func TestWriterDoesNotHTMLEscapeItsStringFields(t *testing.T) {
}
}
-func TestWriterInvalidUTF8StringUsesGoReplacement(t *testing.T) {
- rec := &sessiondata.Record{Ord: 1, Label: "\xff", Parts: []sessiondata.Part{{Kind: sessiondata.PartData, Data: json.RawMessage("\"\xff\"")}}}
+// No landed line holds a byte that is not valid UTF-8: JSON sent between
+// systems must be UTF-8, and a protocol buffer string, which carries a file
+// over OTLP, must be too. A string the writer encodes holds U+FFFD in its
+// place. Data holding one is taken out, whatever its size: the part keeps its
+// place, its kind and its identifiers, says omitted, or truncated while its
+// text is left, and keeps the size and the SHA-256 of what it held. The
+// record's dropped list names each such part and why, so a reader tells it
+// from a payload too large to land. Data in valid UTF-8 lands as it is.
+func TestDataThatIsNotValidUTF8IsTakenOut(t *testing.T) {
+ input := json.RawMessage("{\"command\":\"echo \xff\"}")
+ rec := &sessiondata.Record{Ord: 1, Label: "\xff", Parts: []sessiondata.Part{
+ {Kind: sessiondata.PartCall, ID: "toolu_1", Name: "Bash", Data: input},
+ {Kind: sessiondata.PartResult, Of: "toolu_1", Text: "readable", Data: json.RawMessage("\"\xfe\"")},
+ {Kind: sessiondata.PartData, Data: json.RawMessage(`"é"`)},
+ }}
file := writeRecords(t, rec)
+ if !utf8.Valid(file) {
+ t.Fatalf("the file holds a byte that is not valid UTF-8: %q", file)
+ }
+ // Go's default engine writes the replacement character as it is, and the
+ // older engine escapes it. Both read back as the character.
+ replacement := "\xef\xbf\xbd"
line := bytes.Split(file, []byte{'\n'})[1]
- // Go's default engine writes the replacement character literally. The
- // older engine escapes it. This known difference applies to strings,
- // while data keeps its original bytes in both builds.
- if !bytes.Contains(line, []byte(`"label":"�"`)) && !bytes.Contains(line, []byte(`"label":"\ufffd"`)) {
+ if !bytes.Contains(line, []byte(`"label":"`+replacement+`"`)) && !bytes.Contains(line, []byte(`"label":"\`+`ufffd"`)) {
t.Fatalf("unexpected invalid UTF-8 string encoding: %q", line)
}
got := readRecords(t, file)[0]
- if got.Label != "�" || !bytes.Equal(got.Parts[0].Data, rec.Parts[0].Data) {
- t.Fatalf("unexpected replacement or data change: %+v", got)
+ if got.Label != replacement {
+ t.Fatalf("the label reads back as %q", got.Label)
+ }
+ call, result, kept := got.Parts[0], got.Parts[1], got.Parts[2]
+ sum := sha256.Sum256(input)
+ if len(call.Data) != 0 || call.State != model.ContentOmitted || call.Bytes != len(input) ||
+ call.SHA256 != hex.EncodeToString(sum[:]) || call.Kind != sessiondata.PartCall || call.ID != "toolu_1" || call.Name != "Bash" {
+ t.Fatalf("the call part is %+v", call)
+ }
+ if len(result.Data) != 0 || result.State != model.ContentTruncated || result.Text != "readable" || result.Of != "toolu_1" || result.Bytes != 3 {
+ t.Fatalf("the result part is %+v", result)
+ }
+ if string(kept.Data) != `"é"` || kept.State != "" {
+ t.Fatalf("data in valid UTF-8 did not land as it was: %+v", kept)
+ }
+ want := []sessiondata.Drop{
+ {What: "the data of part 0", Bytes: len(input), Why: sessiondata.WhyNotUTF8},
+ {What: "the data of part 1", Bytes: 3, Why: sessiondata.WhyNotUTF8},
+ }
+ if !reflect.DeepEqual(got.Dropped, want) || !strings.Contains(sessiondata.WhyNotUTF8, "UTF-8") {
+ t.Fatalf("the record drops %+v; want %+v", got.Dropped, want)
+ }
+ if len(rec.Parts[0].Data) != 0 || rec.Parts[0].State != model.ContentOmitted || !reflect.DeepEqual(rec.Dropped, want) {
+ t.Fatal("the caller's record does not say what landed")
}
}
diff --git a/tests/chain/invalid_utf8_test.go b/tests/chain/invalid_utf8_test.go
new file mode 100644
index 0000000..247e533
--- /dev/null
+++ b/tests/chain/invalid_utf8_test.go
@@ -0,0 +1,271 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package chain_test
+
+import (
+ "bytes"
+ "encoding/json"
+ "errors"
+ "io"
+ "io/fs"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+ "unicode/utf8"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecodechanges"
+ "github.com/apache/skywalking-ai-sessionizer/internal/export/otlp"
+ "github.com/apache/skywalking-ai-sessionizer/internal/export/otlp/otlptest"
+ "github.com/apache/skywalking-ai-sessionizer/internal/retention"
+ "github.com/apache/skywalking-ai-sessionizer/internal/scenario"
+ "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/sessiondata"
+)
+
+// A source line can hold a byte that is not valid UTF-8 inside a JSON
+// string: the JSON decoder accepts it, and a write cut short is one way to
+// get one. No landed file holds it, since a protocol buffer string must be
+// valid UTF-8 and a file holding one could never be sent. The tool input that
+// held it is taken out, keeping the call, its size and its digest, and says
+// omitted; a change record is written again with U+FFFD in its place; and
+// each record's dropped list says why. Collected and parsed twice, the second
+// pass lands nothing and changes nothing. The index rebuilt from the files'
+// skeletons is the index rebuilt from the files. Every file of the session
+// goes over both transports, beside a clean session in the same root.
+func TestASourceByteThatIsNotUTF8IsSent(t *testing.T) {
+ s := newStage(t, growing)
+ b, err := scenario.Build(s.sc, scenario.FormatClaudeCode, s.out, scenario.Options{
+ At: time.Date(2026, 2, 1, 10, 0, 0, 0, time.UTC), Through: "one",
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ s.session = b.Session
+ breakToolInput(t, s.out, s.session)
+ recorder := t.TempDir()
+ writeBrokenChange(t, recorder, s.session)
+ collectChanges := func() int {
+ t.Helper()
+ st, err := claudecodechanges.New(recorder, s.zone, 0).CollectAll(nil)
+ if err != nil || len(st.Errors) > 0 || st.Conflicts != 0 {
+ t.Fatalf("the change records: %+v, %v", st, err)
+ }
+ return st.Records
+ }
+
+ // A clean session in the same root, landed first, so its files are sent
+ // first too and the broken one lies between them and the rounds.
+ clean := newStage(t, strings.Replace(growing, "add1c7ed-0003-4000-8000-000000000003", "add1c7ed-0009-4000-8000-000000000009", 1))
+ if _, err := scenario.Build(clean.sc, scenario.FormatClaudeCode, clean.out, scenario.Options{
+ At: time.Date(2026, 1, 1, 10, 0, 0, 0, time.UTC),
+ }); err != nil {
+ t.Fatal(err)
+ }
+ clean.zone, clean.session = s.zone, "add1c7ed-0009-4000-8000-000000000009"
+ if st, err := clean.collector().CollectAll(nil); err != nil || len(st.Errors) > 0 {
+ t.Fatal(err, st.Errors)
+ }
+
+ st, err := s.collector().CollectAll(nil)
+ if err != nil || len(st.Errors) > 0 {
+ t.Fatal(err, st.Errors)
+ }
+ if n := collectChanges(); n != 1 {
+ t.Fatalf("%d change records landed; want one", n)
+ }
+ files, err := storage.LandedFiles(s.zone, s.session)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var call, change *sessiondata.Part
+ var callDrops, changeDrops []sessiondata.Drop
+ for _, lf := range files {
+ data, err := os.ReadFile(lf.Path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !utf8.Valid(data) {
+ t.Fatalf("%s holds a byte that is not valid UTF-8", lf.Path)
+ }
+ for _, rec := range records(t, lf.Path) {
+ for i, p := range rec.Parts {
+ switch {
+ case p.Kind == sessiondata.PartCall && p.Name == "Bash":
+ call, callDrops = &rec.Parts[i], rec.Dropped
+ case p.Kind == sessiondata.PartData && bytes.Contains(p.Data, []byte(`"changes/1"`)):
+ change, changeDrops = &rec.Parts[i], rec.Dropped
+ }
+ }
+ }
+ }
+ if call == nil || call.ID == "" || len(call.Data) != 0 || call.State != model.ContentOmitted || call.Bytes == 0 || call.SHA256 == "" ||
+ len(callDrops) != 1 || callDrops[0].Why != sessiondata.WhyNotUTF8 {
+ t.Fatalf("the call whose input held the byte landed as %+v, dropping %+v", call, callDrops)
+ }
+ if change == nil || change.State != model.ContentTruncated || !bytes.Contains(change.Data, []byte("f\xef\xbf\xbdo.go")) ||
+ len(changeDrops) != 1 || changeDrops[0].Why != sessiondata.WhyNotUTF8 {
+ t.Fatalf("the change record that held the byte landed as %+v, dropping %+v", change, changeDrops)
+ }
+ first := s.parse()
+ if first.Number == 0 {
+ t.Fatal("the first parse wrote no round")
+ }
+
+ again, err := s.collector().CollectAll(nil)
+ if err != nil || len(again.Errors) > 0 || again.SourcesLanded != 0 || again.Records != 0 {
+ t.Fatalf("the second pass landed %d sources and %d records: %v %v", again.SourcesLanded, again.Records, err, again.Errors)
+ }
+ if n := collectChanges(); n != 0 {
+ t.Fatalf("the second pass landed %d change records", n)
+ }
+ if r := s.parse(); r.Number != 0 {
+ t.Fatalf("the second parse wrote round %d over nothing new", r.Number)
+ }
+ clean.parse()
+
+ // Every file becomes a skeleton in a copy of the root, and the index
+ // rebuilt from the skeletons is the one rebuilt from the files.
+ removed := storage.NewZone(t.TempDir())
+ copyDir(t, s.zone.Root(), removed.Root())
+ rem := &retention.Remover{Zone: removed, MetadataOnlyAfter: time.Hour, Now: later}
+ if res := rem.Pass(retention.Input{}); res.Files == 0 || len(res.Errors) > 0 {
+ t.Fatalf("retention removed %d files: %v", res.Files, res.Errors)
+ }
+ sameIndex(t, s.zone, removed, s.session)
+ verified(t, removed, s.session)
+
+ for _, protocol := range []string{otlp.ProtocolHTTP, otlp.ProtocolGRPC} {
+ t.Run(protocol, func(t *testing.T) {
+ rcv, err := otlptest.Start()
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer rcv.Close()
+ c, err := rcv.Client(protocol)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer c.Close()
+ if err := os.Remove(filepath.Join(s.zone.Root(), otlp.StateFile)); err != nil && !errors.Is(err, os.ErrNotExist) {
+ t.Fatal(err)
+ }
+ p := &otlp.Pusher{Zone: s.zone, Client: c, Version: "test", ServiceName: "Claude Code", Endpoint: protocol}
+ st, err := p.Pass()
+ if err != nil || len(st.Errors) > 0 {
+ t.Fatalf("the push failed: %v %v", err, st.Errors)
+ }
+ cleanFiles, err := storage.LandedFiles(s.zone, clean.session)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if st.Files < len(files)+len(cleanFiles)+2 {
+ t.Fatalf("the push sent %d files; want every landed file of both sessions and their rounds", st.Files)
+ }
+ if st, err := p.Pass(); err != nil || len(st.Errors) > 0 || st.Files != 0 {
+ t.Fatalf("the next push sent %d files: %v %v", st.Files, err, st.Errors)
+ }
+ })
+ }
+}
+
+// breakToolInput puts a raw 0xff byte inside the tool input string of the
+// session's transcript.
+func breakToolInput(t *testing.T, out, session string) {
+ t.Helper()
+ var transcript string
+ err := filepath.WalkDir(filepath.Join(out, "_source"), func(p string, _ fs.DirEntry, err error) error {
+ if err == nil && strings.HasSuffix(p, session+".jsonl") {
+ transcript = p
+ }
+ return err
+ })
+ if err != nil || transcript == "" {
+ t.Fatalf("no transcript: %v", err)
+ }
+ src, err := os.ReadFile(transcript)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var changed []byte
+ for _, line := range bytes.SplitAfter(src, []byte("\n")) {
+ if bytes.Contains(line, []byte(`"echo one"`)) && bytes.Contains(line, []byte(`"tool_use"`)) {
+ line = bytes.Replace(line, []byte(`"echo one"`), []byte("\"echo \xff one\""), 1)
+ if !json.Valid(bytes.TrimSpace(line)) {
+ t.Fatal("the JSON decoder refuses the line; the test needs one it accepts")
+ }
+ }
+ changed = append(changed, line...)
+ }
+ if bytes.Equal(changed, src) {
+ t.Fatal("no line changed")
+ }
+ if err := os.WriteFile(transcript, changed, 0o644); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// writeBrokenChange writes the recorder's change record for the session's
+// Bash call, with a raw 0xff byte inside the path of the file it changed.
+func writeBrokenChange(t *testing.T, root, session string) {
+ t.Helper()
+ r := &changes.Record{Schema: changes.Schema, ID: "c1", CapturedBy: changes.CapturedByASZPlugin, Session: session,
+ Stream: "main", Tool: "toolu_1", ToolName: "Bash", Time: "2026-02-01T10:00:04Z", Basis: changes.BasisToolWindow,
+ Root: &changes.Root{Path: "/w"}, ChangedFiles: changes.Int(1),
+ Changes: []changes.FileChange{{Path: "placeholder.go", Operation: changes.OpModify, Diff: changes.DiffUnavailable}}}
+ line, err := r.Marshal()
+ if err != nil {
+ t.Fatal(err)
+ }
+ line = bytes.Replace(line, []byte("placeholder.go"), []byte("f\xffo.go"), 1)
+ path := filepath.Join(root, "file-changes-market", "output", session, "main.jsonl")
+ if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(path, append(line, '\n'), 0o644); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// records reads every record of a landed file.
+func records(t *testing.T, path string) []*sessiondata.Record {
+ t.Helper()
+ f, err := sessiondata.Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer f.Close()
+ r, err := sessiondata.NewReader(f)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var out []*sessiondata.Record
+ for {
+ rec, err := r.Next()
+ if errors.Is(err, io.EOF) {
+ return out
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ out = append(out, rec)
+ }
+}
diff --git a/tests/chain/repack_test.go b/tests/chain/repack_test.go
index ed80605..a0b9c99 100644
--- a/tests/chain/repack_test.go
+++ b/tests/chain/repack_test.go
@@ -18,17 +18,27 @@
package chain_test
import (
+ "errors"
+ "fmt"
+ "io"
"os"
"path/filepath"
"reflect"
"runtime"
+ "slices"
"testing"
"time"
+ "github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecodechanges"
"github.com/apache/skywalking-ai-sessionizer/internal/parse"
"github.com/apache/skywalking-ai-sessionizer/internal/repack"
+ "github.com/apache/skywalking-ai-sessionizer/internal/retention"
+ "github.com/apache/skywalking-ai-sessionizer/internal/scenario"
+ "github.com/apache/skywalking-ai-sessionizer/internal/scenario/expect"
"github.com/apache/skywalking-ai-sessionizer/internal/storage"
"github.com/apache/skywalking-ai-sessionizer/internal/verify"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/changes"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
"github.com/apache/skywalking-ai-sessionizer/pkg/sessionflow"
)
@@ -131,3 +141,446 @@ func TestRepackRefusesARetiredSession(t *testing.T) {
})
}
}
+
+// TestRepackKeepsTheIdleClock. Retention reads how long a conversation has
+// been idle from the header of its newest landed file, the one with the
+// highest sequence. A repack keeps that time exactly, whatever its budget:
+// a file it cuts says when the newest file its records came from was
+// collected, and the files are numbered in the order their records landed.
+// Before, a merged file kept its first source file's time, and files were
+// numbered stream by stream, so retention removed or retired the copy of a
+// conversation the source kept.
+func TestRepackKeepsTheIdleClock(t *testing.T) {
+ day := func(d int) time.Time { return time.Date(2026, 3, d, 10, 0, 0, 0, time.UTC) }
+ s := newStage(t, growing)
+ for _, step := range []struct {
+ checkpoint string
+ at time.Time
+ }{{"one", day(1)}, {"two", day(2)}, {"", day(30)}} {
+ b, err := scenario.Build(s.sc, scenario.FormatClaudeCode, s.out, scenario.Options{
+ At: time.Date(2026, 2, 1, 10, 0, 0, 0, time.UTC), Through: step.checkpoint,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ s.session = b.Session
+ col := s.collector()
+ col.Now = func() time.Time { return step.at }
+ if st, err := col.CollectAll(nil); err != nil || len(st.Errors) > 0 {
+ t.Fatal(err, st.Errors)
+ }
+ }
+ s.parse()
+ want := newestAt(t, s.zone, s.session)
+ if !want.Equal(day(30)) {
+ t.Fatalf("the source's newest file says %s; want day 30", want)
+ }
+ for _, budget := range []int64{64 << 20, 2 << 20, 1} {
+ t.Run(fmt.Sprint(budget), func(t *testing.T) {
+ dst := storage.NewZone(t.TempDir())
+ if _, err := repack.Session(s.zone, dst, s.session, budget, day(30)); err != nil {
+ t.Fatal(err)
+ }
+ if got := newestAt(t, dst, s.session); !got.Equal(want) {
+ t.Fatalf("the destination's newest file says %s; the source's says %s", got, want)
+ }
+ if _, err := parse.Session(dst, parse.Options{Conversation: s.session, Session: s.session}); err != nil {
+ t.Fatal(err)
+ }
+ for _, z := range []*storage.Zone{s.zone, dst} {
+ rem := &retention.Remover{Zone: z, MetadataOnlyAfter: 10 * 24 * time.Hour, RetireAfter: 10 * 24 * time.Hour,
+ Now: func() time.Time { return day(31) }}
+ if res := rem.Pass(retention.Input{}); res.Files != 0 || len(res.Retired) != 0 || len(res.Errors) != 0 {
+ t.Fatalf("%s: retention a day after the last collection removed %d files and retired %v: %v", z.Root(), res.Files, res.Retired, res.Errors)
+ }
+ }
+ })
+ }
+}
+
+// newestAt is the time the header of a session's highest sequence says,
+// the time retention reads.
+func newestAt(t *testing.T, z *storage.Zone, session string) time.Time {
+ t.Helper()
+ files, err := storage.LandedFiles(z, session)
+ if err != nil || len(files) == 0 {
+ t.Fatalf("no landed files: %v", err)
+ }
+ at, err := time.Parse(time.RFC3339Nano, landedHeader(t, files[len(files)-1].Path).At)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return at
+}
+
+func landedHeader(t *testing.T, path string) sessiondata.Header {
+ t.Helper()
+ f, err := sessiondata.Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer f.Close()
+ r, err := sessiondata.NewReader(f)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return r.Header()
+}
+
+// TestRepackFitsEveryFileInItsBudget. A file a repack writes fits its
+// budget with its header and closing line, as a file a collector cuts does,
+// unless it holds one record larger than the budget. Before, only the record
+// lines were counted.
+func TestRepackFitsEveryFileInItsBudget(t *testing.T) {
+ s := newStage(t, growing)
+ s.through("")
+ merged := false
+ for _, budget := range []int64{1024, 1536, 2048, 4096} {
+ dst := storage.NewZone(t.TempDir())
+ if _, err := repack.Session(s.zone, dst, s.session, budget, time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ files, err := storage.LandedFiles(dst, s.session)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, lf := range files {
+ info, err := os.Stat(lf.Path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ n := len(records(t, lf.Path))
+ if n > 1 && info.Size() > budget {
+ t.Fatalf("budget %d: %s holds %d records in %d bytes", budget, filepath.Base(lf.Path), n, info.Size())
+ }
+ merged = merged || n > 1
+ }
+ }
+ if !merged {
+ t.Fatal("no file held two records; the check exercised nothing")
+ }
+}
+
+// TestRepackCarriesTheCursorOfEveryRecorderRoot. The change records of one
+// stream can come from several sources: a recorder root outside the storage
+// root, and one inside it with two plugin directories, one of them at the
+// same relative path as the first root's. The first source keeps the plain
+// cursor, and each other keeps one of its own, named for its origin and its
+// path. A recorder inside the storage root writes its origin relative to it,
+// so a repack writes that origin, and the name of its cursor, as a collector
+// into the new root computes them. Each cursor names the file holding the
+// last record its own source landed. Collection into the new root goes on
+// from where each source stopped: it lands nothing, then only the records
+// appended since, and the new root parses and verifies. Records of two files
+// are cut into one only when the files' headers are the same, so every file
+// names its own source.
+func TestRepackCarriesTheCursorOfEveryRecorderRoot(t *testing.T) {
+ const session = "11111111-2222-4333-8444-555555555555"
+ line := func(id, tool string) string {
+ r := &changes.Record{Schema: changes.Schema, ID: id, CapturedBy: changes.CapturedByASZPlugin, Session: session,
+ Stream: "main", Tool: tool, ToolName: "Bash", Time: "2026-09-08T02:00:04Z", Basis: changes.BasisToolWindow,
+ Root: &changes.Root{Path: "/w"}, ChangedFiles: changes.Int(0)}
+ b, err := r.Marshal()
+ if err != nil {
+ t.Fatal(err)
+ }
+ return string(b) + "\n"
+ }
+ for _, budget := range []int64{1, 64 << 20} {
+ t.Run(fmt.Sprint(budget), func(t *testing.T) {
+ src := storage.NewZone(t.TempDir())
+ outside, inside := t.TempDir(), filepath.Join(src.Root(), "_recorder")
+ add := func(root, plugin string, lines ...string) {
+ t.Helper()
+ path := filepath.Join(root, plugin, "output", session, "main.jsonl")
+ if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, l := range lines {
+ if _, err := f.WriteString(l); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if err := f.Close(); err != nil {
+ t.Fatal(err)
+ }
+ }
+ add(outside, "file-changes-market", line("o/c1", "toolu_1"), line("o/c2", "toolu_2"))
+ add(inside, "file-changes-market", line("i/c1", "toolu_3"))
+ add(inside, "asz-changes-market", line("i/c2", "toolu_4"))
+ collect := func(z *storage.Zone) int {
+ t.Helper()
+ records := 0
+ for _, root := range []string{outside, inside} {
+ st, err := claudecodechanges.New(root, z, 0).CollectAll(nil)
+ if err != nil || len(st.Errors) > 0 || st.Conflicts != 0 {
+ t.Fatalf("%s: %+v, %v", root, st, err)
+ }
+ records += st.Records
+ }
+ return records
+ }
+ if n := collect(src); n != 4 {
+ t.Fatalf("the source landed %d records; want four", n)
+ }
+ dst := storage.NewZone(t.TempDir())
+ if _, err := repack.Session(src, dst, session, budget, time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ sources := func(z *storage.Zone) map[string]bool {
+ out := map[string]bool{}
+ files, err := storage.LandedFiles(z, session)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, lf := range files {
+ out[landedHeader(t, lf.Path).Src] = true
+ }
+ return out
+ }
+ if a, b := sources(src), sources(dst); !reflect.DeepEqual(a, b) || len(a) != 2 {
+ t.Fatalf("the files name the sources %v; the source root's name %v", b, a)
+ }
+ before, after := cursorsIn(t, src, session), cursorsIn(t, dst, session)
+ if len(after) != 3 {
+ t.Fatalf("the new root has %d cursors; want one for each of three sources", len(after))
+ }
+ for _, a := range after {
+ var b *storage.Cursor
+ for _, c := range before {
+ if c.Source == a.Source && c.Offset == a.Offset {
+ b = c
+ }
+ }
+ if b == nil {
+ t.Fatalf("the new root's cursor for %s at %d has none in the source root", a.Source, a.Offset)
+ }
+ landed := recordsOf(t, src, session, b.LastSeq)
+ named := recordsOf(t, dst, session, a.LastSeq)
+ if len(landed) == 0 || !slices.Contains(named, landed[len(landed)-1]) {
+ t.Fatalf("the cursor for %s at %d names a file holding %v; its source landed last %v", a.Source, a.Offset, named, landed)
+ }
+ }
+ if n := collect(dst); n != 0 {
+ t.Fatalf("collection into the new root landed %d records again", n)
+ }
+ parsedAndVerified(t, dst, session)
+ add(outside, "file-changes-market", line("o/c3", "toolu_5"))
+ add(inside, "file-changes-market", line("i/c3", "toolu_6"))
+ add(inside, "asz-changes-market", line("i/c4", "toolu_7"))
+ if n := collect(dst); n != 3 {
+ t.Fatalf("collection into the new root landed %d records; want the three appended", n)
+ }
+ parsedAndVerified(t, dst, session)
+ })
+ }
+}
+
+// cursorsIn reads every cursor of a session's main stream.
+func cursorsIn(t *testing.T, z *storage.Zone, session string) []*storage.Cursor {
+ t.Helper()
+ paths, err := filepath.Glob(filepath.Join(z.StreamDir(session, "main"), "*.cursor"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var out []*storage.Cursor
+ for _, p := range paths {
+ c, err := storage.LoadCursor(p, storage.CursorAppend, "")
+ if err != nil {
+ t.Fatal(err)
+ }
+ out = append(out, c)
+ }
+ return out
+}
+
+// recordsOf is the line, the byte offset and the id of every record of a
+// session's file with sequence seq, in order.
+func recordsOf(t *testing.T, z *storage.Zone, session string, seq uint64) []string {
+ t.Helper()
+ files, err := storage.LandedFiles(z, session)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var out []string
+ for _, lf := range files {
+ if lf.Seq == seq {
+ for _, rec := range records(t, lf.Path) {
+ out = append(out, fmt.Sprintf("%d@%d %s", rec.Ord, rec.Off, rec.ID))
+ }
+ }
+ }
+ return out
+}
+
+// parsedAndVerified parses a session until its chain reaches its files,
+// and checks the files and the chain verify. A second parse writes nothing.
+func parsedAndVerified(t *testing.T, z *storage.Zone, session string) {
+ t.Helper()
+ for {
+ r, err := parse.Session(z, parse.Options{Conversation: session, Session: session})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !r.More {
+ break
+ }
+ }
+ if r, err := parse.Session(z, parse.Options{Conversation: session, Session: session}); err != nil || r.Number != 0 {
+ t.Fatalf("a second parse wrote round %d: %v", r.Number, err)
+ }
+ if sr, err := verify.Session(z, session); err != nil || !sr.OK() {
+ t.Fatalf("the session does not verify: %v %v", err, sr.Details())
+ }
+ if cr, err := verify.Chain(z, session, nil); err != nil || !cr.OK() {
+ t.Fatalf("the chain does not verify: %v %v", err, cr.Details())
+ }
+}
+
+// TestRepackKeepsTheOrderWhenAHeaderComesBack. A header can change and
+// change back within one stream, as an adapter upgraded and then rolled back
+// does. A repack cuts records of two files together only when they are next
+// to each other, so the records keep their order: the transcript's lines and
+// byte offsets run on with no gap, and every provider body still comes after
+// the earlier bodies it copies and cites, so each one rebuilds. Before, the
+// two files with the first header were cut into one, which came after the
+// file between them.
+func TestRepackKeepsTheOrderWhenAHeaderComesBack(t *testing.T) {
+ sc, err := scenario.Load(filepath.Join("..", "scenarios", "provider-bodies-long.yaml"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ out := t.TempDir()
+ var session string
+ for _, checkpoint := range []string{"first", "second", ""} {
+ b, err := scenario.Build(sc, scenario.FormatSD, out, scenario.Options{
+ At: time.Date(2026, 2, 1, 10, 0, 0, 0, time.UTC), Through: checkpoint,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ session = b.Session
+ }
+ src := storage.NewZone(out)
+ files, err := storage.LandedFiles(src, session)
+ if err != nil {
+ t.Fatal(err)
+ }
+ byDir := map[string][]storage.LandedFile{}
+ for _, lf := range files {
+ byDir[filepath.Dir(lf.Path)] = append(byDir[filepath.Dir(lf.Path)], lf)
+ }
+ main, provider := byDir[src.StreamDir(session, "main")], byDir[src.ProviderDir(session)]
+ if len(main) < 3 || len(provider) < 3 {
+ t.Fatalf("the build landed %d transcript and %d provider files; the test needs three of each", len(main), len(provider))
+ }
+ for _, lf := range []storage.LandedFile{main[1], provider[1]} {
+ reheader(t, lf.Path, func(h *sessiondata.Header) { h.Adapter += "-other" })
+ }
+ if sr, err := verify.Session(src, session); err != nil || !sr.OK() {
+ t.Fatalf("the source does not verify: %v %v", err, sr.Details())
+ }
+
+ dst := storage.NewZone(t.TempDir())
+ if _, err := repack.Session(src, dst, session, 64<<20, time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ for _, dir := range []func(*storage.Zone) string{
+ func(z *storage.Zone) string { return z.StreamDir(session, "main") },
+ func(z *storage.Zone) string { return z.ProviderDir(session) },
+ } {
+ if a, b := recordOrder(t, src, session, dir(src)), recordOrder(t, dst, session, dir(dst)); !reflect.DeepEqual(a, b) {
+ t.Fatalf("%s holds its records in another order:\n%v\n%v", filepath.Base(dir(dst)), a, b)
+ }
+ }
+ sr, err := verify.Session(dst, session)
+ if err != nil || !sr.OK() {
+ t.Fatalf("the destination does not verify: %v %v", err, sr.Details())
+ }
+ for {
+ r, err := parse.Session(dst, parse.Options{Conversation: session, Session: session})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !r.More {
+ break
+ }
+ }
+ if cr, err := verify.Chain(dst, session, nil); err != nil || !cr.OK() {
+ t.Fatalf("the destination's chain does not verify: %v %v", err, cr.Details())
+ }
+ lines, _, err := expect.ProviderBodiesRebuild(dst.Root(), session)
+ if err != nil || len(lines) > 0 {
+ t.Fatalf("the destination's provider bodies do not rebuild: %v %v", err, lines)
+ }
+}
+
+// reheader writes a landed file again with its header changed, keeping its
+// record lines byte for byte.
+func reheader(t *testing.T, path string, change func(*sessiondata.Header)) {
+ t.Helper()
+ f, err := sessiondata.Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ r, err := sessiondata.NewReader(f)
+ if err != nil {
+ t.Fatal(err)
+ }
+ hdr := r.Header()
+ var lines [][]byte
+ for {
+ line, err := r.NextRaw()
+ if errors.Is(err, io.EOF) {
+ break
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ lines = append(lines, append([]byte(nil), line...))
+ }
+ f.Close()
+ change(&hdr)
+ if err := os.Chmod(path, 0o644); err != nil {
+ t.Fatal(err)
+ }
+ err = storage.WriteAtomic(path, storage.PermLanded, func(w io.Writer) error {
+ sw, err := sessiondata.NewWriter(w, &hdr)
+ if err != nil {
+ return err
+ }
+ for _, line := range lines {
+ if err := sw.WriteRaw(line); err != nil {
+ return err
+ }
+ }
+ return sw.Close()
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+}
+
+// recordOrder is the line, the byte offset and the id of every record in
+// one directory of a session, in the order the session's files hold them.
+func recordOrder(t *testing.T, z *storage.Zone, session, dir string) []string {
+ t.Helper()
+ files, err := storage.LandedFiles(z, session)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var out []string
+ for _, lf := range files {
+ if filepath.Dir(lf.Path) != dir {
+ continue
+ }
+ for _, rec := range records(t, lf.Path) {
+ out = append(out, fmt.Sprintf("%d@%d %s", rec.Ord, rec.Off, rec.ID))
+ }
+ }
+ return out
+}