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/09/15 10:43:54 UTC
(skywalking-ai-sessionizer) branch provider-bodies created (now 0c83293)
This is an automated email from the ASF dual-hosted git repository.
wu-sheng pushed a change to branch provider-bodies
in repository https://gitbox.apache.org/repos/asf/skywalking-ai-sessionizer.git
at 0c83293 Collect Claude Code's provider bodies, with repeats taken out
This branch includes the following new commits:
new 0c83293 Collect Claude Code's provider bodies, with repeats taken out
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(skywalking-ai-sessionizer) 01/01: Collect Claude Code's provider bodies, with repeats taken out
Posted by wu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wu-sheng pushed a commit to branch provider-bodies
in repository https://gitbox.apache.org/repos/asf/skywalking-ai-sessionizer.git
commit 0c83293cfad9f6e2297c28756f98575f49189f22
Author: Wu Sheng <wu...@foxmail.com>
AuthorDate: Tue Sep 15 14:45:57 2026 +0800
Collect Claude Code's provider bodies, with repeats taken out
Claude Code writes every request it sends its model provider, and every
response, when OTEL_LOG_RAW_API_BODIES=file:<dir> is in its environment.
A request holds what no transcript records: the system prompt, the tool
schemas, the injected reminders and the system messages between the
others. Every call sends its chain's whole message list again, so most
of a body is bytes the session already holds.
- pkg/providerbody cuts a body into pieces, each tool definition and
every string of 1 KiB or more, and copies the front a body shares with
the previous body of its chain. A record keeps only what is new; its
last part, a provider_body/1 manifest in the model's words, rebuilds
the body byte for byte. References are record ids and digests, which
asz repack leaves valid, and they may reach into earlier files. Two
captures of Claude Code 2.1.260 landed at 25.5% and 21.6% of the bytes
written, and all 42 bodies rebuilt exactly.
- The claude-code-provider adapter reads the flat directory, finds each
body's session from the body and the landed transcripts, and lands a
session's new bodies under <session>/provider_body/, cut at
max_delta_bytes. What it knows of Claude Code's names stays in the
adapter. A root table, _provider/seen.state, is derived from the
landed records. Each record is checked through the writer and reader
before it lands.
- The kind stays out of the fold and out of every index lookup.
storage.LandedFiles lists the session's provider_body directory, and
asz verify rebuilds every body. In asz.view an llm.call lists the role
and landed ref of its request and response, joined by the bodies' own
ids; a reader loads the provider_body files up to that ref and
rebuilds. summary.provider_bodies and summary.captured_prompts count
them. On the second capture all 13 calls had both joined.
- A scenario removal deletes the body files with the session's other
files and drops its lines from the provider table.
- Scenarios can write provider bodies in both formats. An expectation's
collect.max_delta_bytes goes into the build's configuration and the
check follows it. provider_bodies_rebuild reads each body the way an
on-demand reader does, on the root and its repacked copy, and
provider_bodies_leave_the_fold compares fold identities.
---
CLAUDE.md | 2 +
asz.yaml | 28 +
cmd/asz/main.go | 8 +-
cmd/asz/provider.go | 171 ++++++
cmd/asz/provider_test.go | 59 ++
cmd/asz/push.go | 3 +-
cmd/asz/refresh.go | 75 ++-
docs/en/adapters/claude-code.md | 47 +-
docs/en/changes/changes.md | 36 ++
docs/en/formats/asz-view-example.yaml | 2 +
docs/en/formats/asz-view.md | 40 +-
docs/en/formats/session-data.md | 67 ++-
docs/en/formats/storage-root.md | 36 +-
docs/en/guides/scenario.md | 34 ++
docs/en/setup/claude-code-provider-bodies.md | 125 ++++
docs/en/setup/command-line.md | 5 +
docs/en/setup/configuration.md | 25 +
docs/en/setup/export-otlp.md | 6 +-
docs/menu.yml | 2 +
internal/adapters/claudecodeprovider/claude.go | 124 ++++
.../adapters/claudecodeprovider/claude_test.go | 48 ++
.../claudecodeprovider/claudecodeprovider.go | 97 ++++
internal/adapters/claudecodeprovider/collect.go | 632 +++++++++++++++++++++
.../adapters/claudecodeprovider/collect_test.go | 516 +++++++++++++++++
internal/adapters/claudecodeprovider/covered.go | 84 +++
.../adapters/claudecodeprovider/glossary_test.go | 36 ++
internal/adapters/claudecodeprovider/seen.go | 276 +++++++++
internal/assemble/streams.go | 6 +-
internal/config/config.go | 21 +
internal/config/config_test.go | 4 +-
internal/index/fromrecord.go | 11 +
internal/index/index.go | 5 +-
internal/index/lookup.go | 17 +-
internal/scenario/build.go | 45 +-
internal/scenario/expect/expect.go | 330 ++++++++++-
internal/scenario/plan.go | 20 +-
internal/scenario/provider.go | 267 +++++++++
internal/scenario/remove/remove.go | 100 +++-
internal/scenario/run/push.go | 13 +-
internal/scenario/run/remove.go | 14 +-
internal/scenario/run/run.go | 95 +++-
internal/scenario/run/sourcebytes.go | 65 ++-
internal/scenario/scenario.go | 12 +-
internal/scenario/write_claudecode.go | 19 +
internal/scenario/write_sd.go | 140 ++++-
internal/storage/landed.go | 6 +
internal/storage/lock.go | 17 +
internal/storage/zone.go | 15 +
internal/verify/provider.go | 100 ++++
internal/verify/session.go | 17 +
internal/view/conversationview.go | 16 +
internal/view/provider.go | 322 +++++++++++
pkg/providerbody/providerbody.go | 159 ++++++
pkg/providerbody/providerbody_test.go | 501 ++++++++++++++++
pkg/providerbody/scan.go | 194 +++++++
pkg/providerbody/session.go | 564 ++++++++++++++++++
pkg/sessionview/sessionview.go | 18 +
tests/scenarios/provider-bodies-errors.expect.yaml | 25 +
tests/scenarios/provider-bodies-errors.yaml | 27 +
.../provider-bodies-late-child.expect.yaml | 30 +
tests/scenarios/provider-bodies-late-child.yaml | 34 ++
tests/scenarios/provider-bodies.expect.yaml | 33 ++
tests/scenarios/provider-bodies.yaml | 50 ++
63 files changed, 5816 insertions(+), 80 deletions(-)
diff --git a/CLAUDE.md b/CLAUDE.md
index 2e4e934..f659e9c 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -55,6 +55,7 @@ pkg/sessiondata/ Session Data: the landed record, `.sd` — publi
pkg/model/ the conversation vocabulary: node kinds, relations, qualification
pkg/sessionflow/ Session Flow: the round chain of conversation structure, `.sf`
pkg/sessionview/ asz.view: one conversation as one document, version 1.0; never a file
+pkg/providerbody/ provider bodies: cuts a body against what its session holds, and rebuilds it byte for byte
internal/index/ derived lookup structure the assembler resolves against
internal/assemble/ the eight-stage pipeline, index in and structure out
internal/parse/ one round: assemble, compare against the chain, publish the delta
@@ -67,6 +68,7 @@ internal/metrics/ derives the runtime's token metric from landed S
internal/repack/ re-cuts landed files into a new root under another budget
internal/adapters/claudecode/ the claude-code-local adapter
internal/adapters/claudecodeotlp/ the claude-code-otlp adapter: a receiver for the runtime's own exporter, metrics into the spool
+internal/adapters/claudecodeprovider/ the claude-code-provider adapter: the request and response bodies Claude Code writes, landed into their sessions
internal/adapters/mock/ the mock dialect: Session Data a scenario writes directly
internal/scenario/ scenarios: the model, the clock, the two writers, and each session's marker (collector side)
internal/scenario/expect/ expectation files evaluated over a root (server side)
diff --git a/asz.yaml b/asz.yaml
index d7a3090..7d0a6df 100644
--- a/asz.yaml
+++ b/asz.yaml
@@ -123,6 +123,34 @@ adapters:
interval: 10m
max_delta_bytes: 2097152
+ # The request and response bodies Claude Code exchanges with its model
+ # provider: the system prompt, the tool schemas and every message as it
+ # was sent. Claude Code writes them when its environment has
+ # OTEL_LOG_RAW_API_BODIES=file:<absolute path of source_root>, for example
+ # in the env block of ~/.claude/settings.json. This adapter finds each
+ # body's session, and lands it with what the session already holds taken
+ # out. On by default because it costs nothing until that variable is set.
+ - name: claude-code-provider
+ enabled: true
+
+ # Where Claude Code writes the bodies. Leave empty for asz/provider-bodies
+ # under CLAUDE_CONFIG_DIR, XDG_CONFIG_HOME/claude, or ~/.claude. The
+ # variable must name the same directory by its absolute path: Claude
+ # Code does not expand ~, and resolves a relative path against each
+ # session's own working directory.
+ source_root: ""
+
+ # The same session filter as claude-code-local, judged by the working
+ # directory the session's main transcript was recorded under.
+ include: []
+ exclude:
+ - /private/tmp/**
+
+ collector:
+ mode: watch
+ interval: 10m
+ max_delta_bytes: 2097152
+
parse:
# The largest round file the parser writes, in bytes. A round travels
# whole as one log record, so it is cut at the same budget as a landed
diff --git a/cmd/asz/main.go b/cmd/asz/main.go
index 5130702..79ae289 100644
--- a/cmd/asz/main.go
+++ b/cmd/asz/main.go
@@ -179,7 +179,7 @@ func main() {
continue
}
switch ad.Name {
- case config.AdapterClaudeCodeLocal, config.AdapterClaudeCodeChanges:
+ case config.AdapterClaudeCodeLocal, config.AdapterClaudeCodeChanges, config.AdapterClaudeCodeProvider:
local = append(local, ad)
case config.AdapterClaudeCodeOTLP:
if cmd != "collect" && cmd != "server" {
@@ -205,7 +205,7 @@ func main() {
switch cmd {
case "sources":
if len(local) == 0 {
- fatal(fmt.Errorf("%s: no enabled %s or %s adapter", cmd, config.AdapterClaudeCodeLocal, config.AdapterClaudeCodeChanges))
+ fatal(fmt.Errorf("%s: no enabled %s, %s or %s adapter", cmd, config.AdapterClaudeCodeLocal, config.AdapterClaudeCodeChanges, config.AdapterClaudeCodeProvider))
}
for _, ad := range local {
if err := run(cfg, ad, *once); err != nil {
@@ -278,6 +278,9 @@ func cmdSources(cfg *config.Config, ad config.Adapter, once bool) error {
if ad.Name == config.AdapterClaudeCodeChanges {
return cmdSourcesChanges(cfg, ad, once)
}
+ if ad.Name == config.AdapterClaudeCodeProvider {
+ return cmdSourcesProvider(cfg, ad, once)
+ }
root, err := claudecode.ResolveSourceRoot(ad.SourceRoot)
if err != nil {
return err
@@ -403,6 +406,7 @@ func printIndexDetail(ix *index.Index, id string) {
index.KindMeta: "agent_meta", index.KindJournal: "journal",
index.KindManifest: "manifest", index.KindScript: "script",
index.KindOther: "other", index.KindUnknown: "unknown", index.KindChanges: "changes",
+ index.KindProviderBody: "provider_body",
}
byKind := map[index.Kind]int{}
msgs := map[uint32]int{}
diff --git a/cmd/asz/provider.go b/cmd/asz/provider.go
new file mode 100644
index 0000000..17f04bc
--- /dev/null
+++ b/cmd/asz/provider.go
@@ -0,0 +1,171 @@
+// 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 main
+
+import (
+ "encoding/json"
+ "fmt"
+ "os"
+ "path/filepath"
+ "sort"
+ "strings"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecode"
+ "github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecodeprovider"
+ "github.com/apache/skywalking-ai-sessionizer/internal/config"
+ "github.com/apache/skywalking-ai-sessionizer/internal/storage"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/providerbody"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+)
+
+// providerFilter builds the provider adapter's session filter for one pass.
+//
+// A body names its session and nothing about where the session ran, so the
+// adapter's include and exclude are judged the way the local adapter judges
+// them: by the directory the session's main transcript sits under. Discovery
+// of Claude Code's projects finds it for a live session. A session whose
+// transcripts were pruned since they landed is judged by the directories its
+// landed transcripts name, the first part of each header's src. A session
+// nothing names a directory for waits.
+func providerFilter(ad config.Adapter, projects string, zone *storage.Zone) claudecodeprovider.Filter {
+ m := claudecode.NewMatcher(ad.Include, ad.Exclude)
+ var found map[string]claudecode.Session
+ judged := map[string]claudecodeprovider.Verdict{}
+ return func(id string) claudecodeprovider.Verdict {
+ if v, ok := judged[id]; ok {
+ return v
+ }
+ if found == nil {
+ found = map[string]claudecode.Session{}
+ sessions, _, err := claudecode.DiscoverWithWarnings(projects)
+ if err == nil {
+ for _, s := range sessions {
+ found[s.ID] = s
+ }
+ }
+ }
+ s, ok := found[id]
+ if !ok {
+ s, ok = landedSession(zone, id)
+ }
+ v := claudecodeprovider.Wait
+ switch {
+ case !ok:
+ case m.Match(s):
+ v = claudecodeprovider.Collect
+ default:
+ v = claudecodeprovider.Exclude
+ }
+ judged[id] = v
+ return v
+ }
+}
+
+// landedSession recovers where a session ran from its landed transcripts: the
+// main transcript's source directory as the primary one, and every
+// transcript's as the directories it spans.
+func landedSession(zone *storage.Zone, id string) (claudecode.Session, bool) {
+ files, err := storage.LandedFiles(zone, id)
+ if err != nil {
+ return claudecode.Session{}, false
+ }
+ s := claudecode.Session{ID: id}
+ seen := map[string]bool{}
+ for _, lf := range files {
+ f, err := os.Open(lf.Path)
+ if err != nil {
+ continue
+ }
+ r, err := sessiondata.NewReader(f)
+ if err != nil {
+ f.Close()
+ continue
+ }
+ hdr := r.Header()
+ f.Close()
+ if hdr.Kind != sessiondata.KindTranscript {
+ continue
+ }
+ dir, _, ok := strings.Cut(hdr.Src, "/")
+ if !ok {
+ continue
+ }
+ if hdr.Stream == storage.StreamMain && s.Primary == "" {
+ s.Primary = dir
+ }
+ if !seen[dir] {
+ seen[dir] = true
+ s.Dirs = append(s.Dirs, dir)
+ }
+ }
+ return s, len(s.Dirs) > 0
+}
+
+// cmdSourcesProvider says what the provider adapter can see: the directory,
+// how many bodies it holds, and which sessions the requests name.
+func cmdSourcesProvider(_ *config.Config, ad config.Adapter, _ bool) error {
+ root, err := claudecodeprovider.ResolveSourceRoot(ad.SourceRoot)
+ if err != nil {
+ return err
+ }
+ fmt.Printf("\nprovider root: %s\n", root)
+ items, err := os.ReadDir(root)
+ if os.IsNotExist(err) {
+ fmt.Printf("no provider bodies; Claude Code writes them when OTEL_LOG_RAW_API_BODIES=file:%s\n", root)
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+ var requests, responses int
+ sessions := map[string]int{}
+ for _, it := range items {
+ name := it.Name()
+ switch claudecodeprovider.RoleOf(name) {
+ case providerbody.RoleResponse:
+ responses++
+ case providerbody.RoleRequest:
+ requests++
+ if s := sessionOf(filepath.Join(root, name)); s != "" {
+ sessions[s]++
+ }
+ }
+ }
+ fmt.Printf("bodies : %d request(s), %d response(s)\n", requests, responses)
+ ids := make([]string, 0, len(sessions))
+ for id := range sessions {
+ ids = append(ids, id)
+ }
+ sort.Strings(ids)
+ for _, id := range ids {
+ fmt.Printf(" %s %d request(s)\n", id, sessions[id])
+ }
+ if n := len(ids); n > 0 {
+ fmt.Printf("\n%d session(s) named by requests; a response is joined to its session when collected\n", n)
+ }
+ return nil
+}
+
+// sessionOf reads the session a request body names, or nothing.
+func sessionOf(path string) string {
+ b, err := os.ReadFile(path)
+ if err != nil || !json.Valid(b) {
+ return ""
+ }
+ return strings.TrimSpace(claudecodeprovider.Lift(filepath.Base(path), b).Session)
+}
diff --git a/cmd/asz/provider_test.go b/cmd/asz/provider_test.go
new file mode 100644
index 0000000..f109ba2
--- /dev/null
+++ b/cmd/asz/provider_test.go
@@ -0,0 +1,59 @@
+// 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 main
+
+import (
+ "io"
+ "path/filepath"
+ "testing"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecodeprovider"
+ "github.com/apache/skywalking-ai-sessionizer/internal/config"
+ "github.com/apache/skywalking-ai-sessionizer/internal/storage"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+)
+
+// A session whose transcripts were pruned is judged by the directory its
+// landed transcripts name, never collected because the root holds it.
+func TestProviderFilterJudgesAPrunedSessionByItsLandedTranscripts(t *testing.T) {
+ const session = "11111111-2222-4333-8444-555555555555"
+ zone := storage.NewZone(t.TempDir())
+ dir := zone.StreamDir(session, storage.StreamMain)
+ err := storage.WriteAtomic(filepath.Join(dir, storage.LandedName("transcript", "20260101T000000.000000000Z", 1)), storage.PermLanded, func(w io.Writer) error {
+ rw, err := sessiondata.NewWriter(w, &sessiondata.Header{Seq: 1, Kind: sessiondata.KindTranscript, Adapter: "claude-code-local/0.1.0",
+ Dialect: "claude-code/1", Src: "-private-tmp-scratch/" + session + ".jsonl", Session: session, Stream: storage.StreamMain})
+ if err != nil {
+ return err
+ }
+ return rw.Close()
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ projects := t.TempDir() // discovery finds nothing: the transcript is pruned
+ ad := config.Adapter{Exclude: []string{"/private/tmp/**"}}
+ if v := providerFilter(ad, projects, zone)(session); v != claudecodeprovider.Exclude {
+ t.Fatalf("a pruned session under an excluded directory: verdict %d, want Exclude", v)
+ }
+ if v := providerFilter(config.Adapter{}, projects, zone)(session); v != claudecodeprovider.Collect {
+ t.Fatalf("a pruned session with no filter: verdict %d, want Collect", v)
+ }
+ if v := providerFilter(config.Adapter{}, projects, zone)("66666666-7777-4888-8999-aaaaaaaaaaaa"); v != claudecodeprovider.Wait {
+ t.Fatalf("a session nothing names a directory for: verdict %d, want Wait", v)
+ }
+}
diff --git a/cmd/asz/push.go b/cmd/asz/push.go
index 877fc43..0e8e0b6 100644
--- a/cmd/asz/push.go
+++ b/cmd/asz/push.go
@@ -26,6 +26,7 @@ import (
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecode"
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecodechanges"
+ "github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecodeprovider"
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/mock"
"github.com/apache/skywalking-ai-sessionizer/internal/config"
"github.com/apache/skywalking-ai-sessionizer/internal/export/otlp"
@@ -134,7 +135,7 @@ func newPusher(cfg *config.Config, zoneRoot string) (*otlp.Pusher, func(), error
Endpoint: otlp.EndpointOf(o.Protocol, o.Endpoint, o.TLS),
Version: version,
ServiceName: o.ServiceName,
- Runtimes: map[string]string{claudecode.Name: claudecode.RuntimeName, claudecodechanges.Name: claudecodechanges.RuntimeName, mock.Name: mock.RuntimeName},
+ Runtimes: map[string]string{claudecode.Name: claudecode.RuntimeName, claudecodechanges.Name: claudecodechanges.RuntimeName, claudecodeprovider.Name: claudecodeprovider.RuntimeName, mock.Name: mock.RuntimeName},
InstanceID: o.InstanceID,
Layer: o.Layer,
BatchBytes: o.BatchBytes,
diff --git a/cmd/asz/refresh.go b/cmd/asz/refresh.go
index 8606378..efffb2e 100644
--- a/cmd/asz/refresh.go
+++ b/cmd/asz/refresh.go
@@ -29,6 +29,7 @@ import (
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecode"
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecodechanges"
+ "github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecodeprovider"
"github.com/apache/skywalking-ai-sessionizer/internal/config"
"github.com/apache/skywalking-ai-sessionizer/internal/export/otlp"
"github.com/apache/skywalking-ai-sessionizer/internal/metrics"
@@ -59,15 +60,23 @@ type refresher struct {
match func(claudecode.Session) bool
changes *claudecodechanges.Collector
cmatch func(claudecodechanges.Session) bool
- // The directories the two collectors read. They are checked on every
- // pass, not once, so a source that appears later is picked up.
- colSource string
- changesSource string
- interval time.Duration
- maxRound int64
+ // provider lands the bodies Claude Code wrote for its model provider,
+ // and providerAd holds its filter settings.
+ provider *claudecodeprovider.Collector
+ providerAd config.Adapter
+ // The directories the collectors read. They are checked on every pass,
+ // not once, so a source that appears later is picked up.
+ colSource string
+ changesSource string
+ providerSource string
+ interval time.Duration
+ maxRound int64
// pushAfter is when sending may resume, set when a receiver asks to be
// left alone for longer than one period. Zero means now.
pushAfter time.Time
+ // waiting and unreadable are the provider body counts the last pass
+ // printed, so a pass that changes either says so.
+ waiting, unreadable int
// base carries the status fields that do not change between passes.
base view.Status
@@ -165,9 +174,18 @@ func newRefresher(srv *view.Server, zone *storage.Zone, ads []config.Adapter, ma
r.cmatch = changesMatch(ad)
r.changesSource = src
names, sources = append(names, ad.Name), append(sources, src)
+ case config.AdapterClaudeCodeProvider:
+ src, err := claudecodeprovider.ResolveSourceRoot(ad.SourceRoot)
+ if err != nil {
+ return nil, err
+ }
+ r.provider = claudecodeprovider.New(src, zone, ad.Collector.MaxDeltaBytes)
+ r.providerAd = ad
+ r.providerSource = src
+ names, sources = append(names, ad.Name), append(sources, src)
}
}
- if r.col == nil && r.changes == nil {
+ if r.col == nil && r.changes == nil && r.provider == nil {
// No local adapter is enabled. A root filled somewhere else, by a
// receiver or copied from another machine, is served as it is.
fmt.Fprintln(os.Stderr, "source : no local adapter enabled; nothing is collected")
@@ -183,6 +201,7 @@ func newRefresher(srv *view.Server, zone *storage.Zone, ads []config.Adapter, ma
r.remover = &remove.Remover{
Zone: zone, Source: r.colSource, Local: r.col, LocalMatch: r.match,
Changes: r.changes, ChangesRoot: r.changesSource, ChangesMatch: r.cmatch,
+ Provider: r.provider, ProviderRoot: r.providerSource,
Derives: r.deriver != nil,
}
}
@@ -333,6 +352,7 @@ func (r *refresher) pass() error {
var errs []error
var landed, records, sessionsSeen, pending, conflicts, busy int
+ var waiting, unreadable int
changed := map[string]bool{}
var cs *claudecode.Stats
@@ -384,6 +404,38 @@ func (r *refresher) pass() error {
}
}
+ // The provider bodies come after the transcripts, so a response the
+ // transcripts are the only evidence for finds its session in this pass.
+ providerHere := false
+ if r.provider != nil {
+ var perr error
+ if providerHere, perr = present(r.providerSource, true); perr != nil {
+ errs = append(errs, perr)
+ }
+ }
+ if providerHere {
+ projects := r.colSource
+ if projects == "" {
+ projects, _ = claudecode.ResolveSourceRoot("")
+ }
+ ps, err := r.provider.CollectAll(providerFilter(r.providerAd, projects, r.zone))
+ if err != nil {
+ errs = append(errs, err)
+ ps = &claudecodeprovider.Stats{}
+ }
+ errs = append(errs, ps.Errors...)
+ landed += ps.SourcesLanded
+ records += ps.Records
+ sessionsSeen += ps.Sessions
+ pending += ps.Pending
+ conflicts += ps.Conflicts
+ busy += ps.Busy
+ waiting, unreadable = ps.Waiting, ps.Unreadable
+ for _, id := range ps.Changed {
+ changed[id] = true
+ }
+ }
+
// What this pass landed, plus whatever an earlier pass could not take
// the lock for.
for id := range r.retry {
@@ -555,7 +607,9 @@ func (r *refresher) pass() error {
if removed > 0 {
pushed += fmt.Sprintf(" removed=%d", removed)
}
- if first || landed > 0 || rounds > 0 || busy > 0 || pushed != "" || len(errs) > 0 {
+ bodies := waiting != r.waiting || unreadable != r.unreadable
+ r.waiting, r.unreadable = waiting, unreadable
+ if first || landed > 0 || rounds > 0 || busy > 0 || pushed != "" || len(errs) > 0 || bodies {
contended := ""
if busy > 0 {
// Two writers on one root is a supported setup, so this is a
@@ -563,6 +617,11 @@ func (r *refresher) pass() error {
// the other one was already doing.
contended = fmt.Sprintf(" busy=%d", busy)
}
+ if waiting > 0 || unreadable > 0 || bodies {
+ // Provider bodies no session claims yet, and files that are not a
+ // body. Neither fails the pass, and both are said.
+ contended += fmt.Sprintf(" bodies_waiting=%d bodies_unreadable=%d", waiting, unreadable)
+ }
fmt.Printf("[%s] refreshed: sessions=%d landed=%d records=%d rounds=%d%s%s (%s)\n",
now.Format("15:04:05"), sessionsSeen, landed, records, rounds, contended, pushed,
now.Sub(start).Round(time.Millisecond))
diff --git a/docs/en/adapters/claude-code.md b/docs/en/adapters/claude-code.md
index 889894e..3e7094c 100644
--- a/docs/en/adapters/claude-code.md
+++ b/docs/en/adapters/claude-code.md
@@ -742,12 +742,57 @@ the same nodes with and without them, and the view joins each record to its step
tool-use id. The scenario `tests/scenarios/workspace-changes.yaml` checks both paths in both
formats, and that the fold is unchanged.
+## Provider bodies
+
+With `OTEL_LOG_RAW_API_BODIES=file:<dir>` in its environment, Claude Code writes the body of every
+request it sends its model provider, and of every response, to that directory.
+`claude-code-provider` is the adapter for them. [Claude Code Provider Bodies](../setup/claude-code-provider-bodies.md)
+says how to turn it on, what the files hold and what lands; this section says what was measured.
+
+Two captures of Claude Code 2.1.260, on `claude-opus-5` with 1M context, each one headless process
+fed one turn at a time:
+
+| | First capture | Second capture |
+| --- | --- | --- |
+| Turns | 3 | 6, including `/compact` |
+| Tools | Read, Grep and Glob only | the full list, write tools refused by permission |
+| Bodies | 6 requests, 6 responses | 15 requests, 15 responses |
+| Bytes written | 674,017 | 1,739,105 |
+| Transcript bytes | 263,111 | 598,360 main, 77,694 subagent |
+| Every body cut and written as one `provider_body` file | 172,120 bytes (25.5%) | 376,154 bytes (21.6%) |
+
+- **Every call sends its chain's whole message list again**, across turns. Earlier messages come
+ back unchanged except the cache marker, which moves to the newest message; a message holding one
+ text block becomes a plain string once the marker leaves it. Each response comes back unchanged as
+ the next assistant message. `role:"system"` messages are placed between the others: the
+ environment, the deferred tool list, and a `<total_tokens>` note after each prompt and tool result.
+- **Bodies are compact JSON**, keys in the order `model`, `messages`, `system`, `tools`, `betas`,
+ `metadata`, `max_tokens`, `thinking`. Encoding each parsed file again compactly gave its exact
+ bytes.
+- **A chain is a stream's run of calls.** Its first request names no `cc_prev_req`; each later one
+ names the previous call of the same chain, across turns. A subagent has its own chain, and its
+ requests carry the parent's session and the parent's current `cc_prompt_id`. No agent id appears in
+ any body.
+- **A compaction request** copies the request of the call before it up to that call's prompt, adds
+ the instruction as a second text block in the same user message, and names the same `cc_prev_req`
+ with no `cc_prompt_id`. The next call names the call before the compaction and starts a new
+ message list with the summary. So `cc_prev_req` alone does not identify a request's call.
+- **Which session a response belongs to.** Of 15 responses, 11 are named by a later request, 2 are
+ the last of their chain and found only through the transcript's message id, and 2, the call that
+ names the session and the compaction call, are found by neither and are not landed.
+- **Names.** The adapter reads Claude Code's fields and writes the model's words: `session_id` in
+ `metadata.user_id` becomes `session`, `cc_prompt_id` becomes `run`, `cc_prev_req` becomes
+ `previous_request`, a response's `id` becomes `call`, and a response file's name becomes `request`.
+ The record id is the file name without `.json`.
+- **The join in the view**, on the second capture: all 13 calls had their request and their response
+ joined exactly, the subagent's included; the two requests with no call were left unjoined.
+
## What this adapter cannot supply
| | Why |
| --- | --- |
| **Reasoning text** | Claude Code asks the provider not to return it, so most thinking blocks carry only a signature. `unavailable` where the runtime wrote none. See below. |
-| **Serialized request** | system prompt, tool schemas and cache annotations are absent from transcripts: 0 files contain `"tools":[`, and `compactMetadata.preservedMessages.allUuids` names 5 ids that exist nowhere on disk. asz produces no input manifest and no `input_of` relation, and reports model-context coverage as `unavailable`, as the model's [adapter contract](../concepts-and-designs/unified-conversation-model.md#adapter-contract) requires. |
+| **Serialized request** | system prompt, tool schemas and cache annotations are absent from transcripts: 0 files contain `"tools":[`, and `compactMetadata.preservedMessages.allUuids` names 5 ids that exist nowhere on disk. asz produces no input manifest and no `input_of` relation, and reports model-context coverage as `unavailable`, as the model's [adapter contract](../concepts-and-designs/unified-conversation-model.md#adapter-contract) requires. Claude Code writes each request and resp [...]
| **Injected preamble** | the instruction block prepended to the first user message has no transcript record. Its *data* survives as `attachment` records; its rendered form does not. |
| **Per-call duration and cost** | not written to transcripts. Available via OTLP. |
| **`tool.execution`** | no local record; the call and result are observable, the execution is not. |
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index 9fc45cf..a9b8b1f 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -2,6 +2,42 @@
> In development, not yet released. `tools/release.sh prepare 0.4.0` removes this note.
+## Provider bodies
+
+- A new adapter, `claude-code-provider`, lands the request and response bodies Claude Code writes
+ for its model provider when `OTEL_LOG_RAW_API_BODIES=file:<dir>` is in its environment: the
+ system prompt, the tool schemas and every message as it was sent, which no transcript holds. A
+ request names its session; a response is claimed by a later request that names its request id,
+ or by the one session whose transcript holds its message id, and waits otherwise. On by default,
+ it does nothing until the variable is set. See
+ [Claude Code Provider Bodies](../setup/claude-code-provider-bodies.md).
+- The kind `provider_body`, reserved until now, lands under `<session>/provider_body/`, one file per
+ session per pass. A record keeps only what its session does not hold yet: each tool definition
+ and every string of 1 KiB or more is a piece named by its SHA-256, and the front a body shares
+ with the previous body of its chain is one copy. Its last part, a `provider_body/1` manifest,
+ rebuilds the body byte for byte, which the adapter checks through the writer and reader before a
+ record lands. On two captures of Claude Code 2.1.260 the landed files were 25.5% and 21.6% of the
+ bytes written, and all 42 bodies rebuilt exactly. `pkg/providerbody` cuts and rebuilds bodies for
+ any reader.
+- Assembly leaves provider bodies out of the fold, and the index keeps them out of every lookup, so
+ a session folds to the same nodes with and without them. In the `asz.view` document, an `llm.call`
+ lists its request and response under `provider_bodies`, each by role and the landed record it
+ rebuilds from, joined by the bodies' own ids; a reader loads the bodies when it wants them.
+ `summary.provider_bodies` counts the bodies and `summary.captured_prompts` the calls whose request is
+ captured. The version stays `1.0`.
+- `asz verify` rebuilds every landed provider body and compares its digest. `storage.LandedFiles`
+ lists the session's `provider_body/` directory, so parse, push, repack and the view see the files.
+ The root keeps a derived table of the body files, `_provider/seen.state`.
+- A scenario with `provider_bodies: true` writes every call's bodies in both formats. The new
+ property `provider_bodies_rebuild` reads each body a call lists from only the provider files up to
+ its own, as an on-demand reader does, on the root and on its repacked copy.
+ `provider_bodies_leave_the_fold` compares the fold without them, and `parts_keep_source_bytes`
+ rebuilds every body and compares it with its file. An expectation's `collect.max_delta_bytes` is
+ written into the build's configuration, and the check collects by it. A scenario removal waits until
+ every body the build wrote is landed, deletes the body files with the session's other files, and
+ drops the session's lines from `_provider/seen.state`. `tests/scenarios/provider-bodies.yaml` is the
+ example, with bodies cut into five files.
+
## Collection
- The collector's default `interval` is 10 minutes, in `asz.yaml` and when no configuration is
diff --git a/docs/en/formats/asz-view-example.yaml b/docs/en/formats/asz-view-example.yaml
index 655f45e..e8e0960 100644
--- a/docs/en/formats/asz-view-example.yaml
+++ b/docs/en/formats/asz-view-example.yaml
@@ -19,6 +19,8 @@ summary:
rounds: 1
unresolved: 0
changes: 1
+ provider_bodies: 0
+ captured_prompts: 0
from: 1767225600000
to: 1767225611100
kinds:
diff --git a/docs/en/formats/asz-view.md b/docs/en/formats/asz-view.md
index 0f949d4..afa56b1 100644
--- a/docs/en/formats/asz-view.md
+++ b/docs/en/formats/asz-view.md
@@ -74,7 +74,7 @@ producer of them.
| `conversation`, `sessions` | the conversation id, and the sessions that contributed to it, from the fold's session nodes; one session, equal to the conversation id, for the Claude Code adapter |
| `head` | `round` and `digest` of the newest round the document was folded to |
| `parser`, `policy` | from the head round's header |
-| `summary` | `title`; `state`, one of `verified`, `incomplete` when a round or a file is missing, `mismatch` when a digest failed; `problems`, one line each, empty when verified; the counts `talks`, `steps`, `streams`, `segments`, `rounds`, `unresolved`, `changes`; `from` and `to`, when the session began and its last activity, from the session node; and `kinds`, `relation_types` and `quality`, the fold sized by node kind, by relation type and by how well each relation is known |
+| `summary` | `title`; `state`, one of `verified`, `incomplete` when a round or a file is missing, `mismatch` when a digest failed; `problems`, one line each, empty when verified; the counts `talks`, `steps`, `streams`, `segments`, `rounds`, `unresolved`, `changes`, `provider_bodies`, the landed provider bodies, and `captured_prompts`, the calls whose request is captured; `from` and `to`, when the session began and its last activity, from the session node; and `kinds`, `relation_types` a [...]
| `rounds` | one per round, in order: `round`, `digest`, `previous` (null on round 1), `from_seq`, `through_seq`, `input_digest`, `from_time`, `through_time` (the record time range of the files the round consumed, null when none carries a time), `verified` |
| `files` | one per `.sd` file, then one per round: `file` (its path on the wire), `format` (`sd` or `sf`), `kind`, `seq` or `round`, `stream` or `run`, `lines`, `bytes`, `digest`, `from_time`, `through_time`. Absent values are null. Together with `rounds`, this is exactly what a rebuild needs. |
| `streams` | one per execution stream: `id`, `name`, `role` (`main` or `child`), `label`, `parent`, `records`, `steps`, `talk`, `named_by`, and `opened_by`, every step the assembler could tie to the start of the stream as `{step, stream, talk, quality}`; several means it did not choose, and neither does a view |
@@ -163,6 +163,44 @@ several windows could have made appears in each of their records, marked `shared
others, and is counted once. A record with `basis: skipped_read_only` carries no changes and
means the call was not observed, never that it changed nothing.
+## Provider bodies
+
+A call step lists the provider bodies joined to it under `provider_bodies`: its request, then its
+response, each only when the join is exact. An entry names where the body is landed. It does not
+carry the body.
+
+```json
+{"id":"call/msg_011C…","kind":"llm.call", …,
+ "provider_bodies":[{"role":"request","ref":{"seq":4,"row":5}},
+ {"role":"response","ref":{"seq":4,"row":6}}]}
+```
+
+| Key | Value |
+| --- | --- |
+| `role` | `request` or `response` |
+| `ref` | the landed `provider_body` record: `seq` names a file under `files`, `row` the record in it |
+
+A reader loads a body when it wants it. A body refers to pieces and bodies that landed before it in
+the same session, sometimes in an earlier file, so a reader loads the session's `provider_body` files
+with a sequence up to the one `ref` names, in order, keeps them, and rebuilds the body as
+[Session Data](session-data.md#provider-bodies) describes. Every such file is listed under `files`.
+A later body needs only the files not loaded yet. The document stays the size of the structure,
+however large the bodies.
+
+A response joins to the call whose message id it carries. A request names no call, only its prompt
+and the request id of the call before it in its chain, so it joins to the call of a stream whose
+previous call's response carries that request id and whose prompt is the one the request names. The
+first call of a stream matches a request that names no previous request. A request joins only when
+exactly one request and exactly one call carry those two ids. A synthetic call, which the runtime wrote
+without calling the provider, takes part in no join. No request joins in a stream whose landed
+transcript lines have a gap, since a call may be missing between two that look consecutive; its
+responses still join. A body that joins to no call, such as
+the request that names the session, a compaction request, or a retried request, is listed on no
+step; its file is still under `files`. Nothing is joined by position or by time.
+
+`summary.provider_bodies` counts the session's landed bodies, and `summary.captured_prompts` the calls
+that list a request.
+
## Rendering the whole conversation
The document is complete: a viewer draws every view of a conversation from it and fetches
diff --git a/docs/en/formats/session-data.md b/docs/en/formats/session-data.md
index 700011f..4c13616 100644
--- a/docs/en/formats/session-data.md
+++ b/docs/en/formats/session-data.md
@@ -41,11 +41,11 @@ stops at the first longer line.
| `schema` | `sd/1` |
| `seq` | the landed sequence number, monotonic per session |
| `at` | when the file was collected |
-| `kind` | what it was collected from: `transcript`, `agent_meta`, `journal`, `workflow_manifest`, `workflow_script` or `changes`. `otlp_log` and `otlp_span` are reserved for a push transport, and `provider_body` is reserved too. No adapter writes any of the three. |
+| `kind` | what it was collected from: `transcript`, `agent_meta`, `journal`, `workflow_manifest`, `workflow_script`, `changes` 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. |
| `src` | the source, relative to the adapter's source root, with forward slashes on every platform |
-| `session`, `stream`, `batch` | the session, the execution stream (`main` or an agent id), and the group of children a workflow run started |
+| `session`, `stream`, `batch` | the session, the execution stream (`main` or an agent id), and the group of children a workflow run started. A `provider_body` file names neither a stream nor a batch |
A reader refuses a header unless `h` is `1`, `schema` is `sd/1`, and `kind`, `session`, `src` and
`dialect` are all present. The writer in `pkg/sessiondata` fills in `h` and `schema`, and refuses
@@ -68,7 +68,8 @@ lets a reader check the record against its source for as long as the source exis
## File kinds
One `.sd` file holds one cut of one source. The header's `kind` says which, and the records of
-each kind carry a characteristic set of fields.
+each kind carry a characteristic set of fields. A `provider_body` file is the exception: its source
+is a directory with one file per body, and each record names its own file.
| `kind` | Collected from | Its records | What refers to it |
| --- | --- | --- | --- |
@@ -78,6 +79,7 @@ each kind carry a characteristic set of fields.
| `workflow_manifest` | a workflow run's manifest | one record with `batch` and `label`, and a `data` part | the run's name |
| `workflow_script` | the program a workflow ran | one record whose part is `unknown`: the source is a program, not data | nothing; kept because it is part of the session |
| `changes` | the files the asz Claude Code plugin saw a tool call change, one line per call | one record per observed call, with `id`, `tool` and `time` lifted from it and the line whole as one `data` part, a `changes/1` record; no `from` and no flags, so assembly emits no node for it | nothing in a round; the view joins each to its step by `tool`. A transcript's `Edit` and `Write` results carry the same record as a second `data` part, from the runtime's own patch. See the [Claude Code [...]
+| `provider_body` | the request and response bodies Claude Code wrote for its model provider, one file per body | one record per body, with `id` unique in the session (for Claude Code, the body's file name without `.json`), `run` on a request, `call` on a response, `model`, and the parts described in [Provider bodies](#provider-bodies); no `from` and no flags, so assembly emits no node for it. The header's `src` is `.` and it names no stream | nothing in a round; the view joins each to i [...]
A source that is one document, not a stream of lines, lands as one record with `ord` 1 and `off` 0.
Its `sha` and `bytes` cover the whole document. The Claude Code adapter's `agent_meta`,
@@ -126,7 +128,9 @@ is a child's synchronous return, which is the parent's copy of the child's outpu
the flag only on a `journal` record. Nothing removes the parent's copy, so it stays as the result
of the call that started the child.
-`asz verify` checks `ord`, `off` and `bytes` without the source. In each stream or run, across its
+`asz verify` checks `ord`, `off` and `bytes` without the source. A provider body is a whole
+document at `ord` 1 and `off` 0, so for it `asz verify` rebuilds the body instead, as
+[Provider bodies](#provider-bodies) says. In each stream or run, across its
files of one kind, `ord` must run 1, 2, 3 with no gap, starting at 1. The first record must start
at byte 0, and each record after it at the byte after the previous record's line ends. Where the
stream has an append cursor, the records must reach the line and the byte the cursor names (see
@@ -154,6 +158,61 @@ the source record, in source order. Content it cannot describe goes in as an `un
same place, never left out. A part the adapter adds beside the content, such as the `changes/1`
record on an edit result, comes after them.
+## Provider bodies
+
+A request Claude Code sends repeats most of the request before it: the whole message list of its
+chain, the system prompt and the tool schemas. A `provider_body` record keeps only what its session
+does not hold yet, and says how to put the rest back.
+
+Its parts come in source order. Each is a `data` part holding a piece of the body, byte for byte: a
+tool definition, or a JSON string of 1 KiB or more, with its quotes. The last part is a `data`
+part that asz builds, the manifest, schema `provider_body/1`:
+
+```json
+{"schema":"provider_body/1","role":"request","src":"fcc261af-….request.json",
+ "sha256":"<SHA-256 of the whole body>","bytes":137054,"depth":1,"chain":"3e1f…",
+ "model":"claude-opus-5","session":"9a91…","run":"75db…","previous_request":"req_011C…",
+ "segments":[{"copy":{"from":"03455f62-….request","sha256":"…","len":26777}},
+ {"lit":"…"},{"part":0},{"piece":"<SHA-256 of a piece>"},{"lit":"…"}]}
+```
+
+| Key | Value |
+| --- | --- |
+| `role`, `src`, `sha256`, `bytes` | request or response, where the body came from relative to the adapter's source root, as a header's `src` is, and the digest and size of the whole body |
+| `segments` | joined in order, they are the body. `lit` is a JSON string whose UTF-8 bytes are literal bytes of the body. `part` is a part of this record. `piece` is the SHA-256 of a piece an earlier record of the session holds. `copy` is the first `len` bytes of the earlier body whose record id is `from` and whose digest is `sha256` |
+| `depth` | how many copies lie between this body and one with none. It is at most 32; a body whose base is that deep copies nothing |
+| `bytes` limit | a body may claim at most 256 MiB, and a rebuild stops as soon as it passes the size its record claims, so a damaged record cannot make a reader hold more than one body |
+| `chain` | the first sixteen hexadecimal characters of the SHA-256 of the request's first message, with the cache marker taken out. It chooses which earlier body a copy comes from. It is storage, and a join must not rely on it |
+| `why` | set when the body is kept whole in one `unknown` part, and why it could not be cut |
+| `model`, `session`, `run`, `call`, `request`, `previous_request` | the values the body carries that join it, as its adapter read them, in the model's words: the session it names; on a request, the run it was sent in; on a response, the call it answers and the provider's id for its request; on a request, the provider request id of the call before it in its chain. Empty when the body carries none. The [adapter's glossary](../adapters/claude-code.md#provider-bodies) says what the runtime [...]
+
+A reference is a digest or a record id, never a file and a row, because `asz repack` renumbers files
+and rows and carries record lines unchanged. Every reference points at a record that landed before
+this one in the same session, so a session rebuilds on its own. A reference may point into an earlier
+file: the tool schemas, for instance, land once, in the session's first provider file. So to rebuild
+a body in the `provider_body` file with sequence N, a reader reads the session's `provider_body` files
+with sequence N or lower, in order, and keeps them: a later body needs only the files it has not read
+yet. Package `pkg/providerbody` does this.
+
+Keeping each file on its own would repeat the tool schemas, the system prompt and one whole request in
+every file. On the second capture, cut into three files of ten bodies, that took 43.9% of the bytes
+Claude Code wrote instead of 21.7%, and a long session with requests of several megabytes would keep
+almost nothing.
+
+A body can have nothing to refer to: the first body of a session, the first of a chain, one past the
+depth limit, or one in a session whose earlier `provider_body` files are gone from the root. It lands
+with all of its pieces, and rebuilds from its own record. A body that refers to a file the reader does
+not have does not rebuild. `asz verify` reports it, and a receiver that keeps a session's files in
+several storage stages has to read the earlier files from whichever stage holds them.
+
+The adapter checks each record before it lands: it writes the record with the Session Data writer,
+reads it back and rebuilds the body. A body that does not come back whole, or is not one JSON object
+in valid UTF-8, lands whole in one `unknown` part with a `why`. `asz verify` rebuilds every landed
+body and compares it with `sha256`, and reports a reference to something the session does not hold.
+
+On two captures of Claude Code 2.1.260, the landed `provider_body` files were 25.5% and 21.6% of the
+bytes Claude Code wrote, and all 42 bodies rebuilt exactly.
+
## Closing line
```json
diff --git a/docs/en/formats/storage-root.md b/docs/en/formats/storage-root.md
index 7b30b7a..0d7bc29 100644
--- a/docs/en/formats/storage-root.md
+++ b/docs/en/formats/storage-root.md
@@ -28,6 +28,8 @@ session.
journal.cursor · journal-…sd
manifest.cursor · manifest-…sd
script.cursor · script-…sd
+ provider_body/ the bodies the runtime exchanged with its model
+ provider_body-<collected-at>-<seq>.sd provider, when claude-code-provider collects them
index/
entries.bin the derived index
index.state how far the index reaches
@@ -38,6 +40,9 @@ session.
rounds/
r000001-<digest>.sf
r000002-<digest>.sf
+ _provider/ when claude-code-provider runs
+ seen.state which body files landed, in which session, and which wait
+ .lock one provider collector per root
_metrics/ the metrics spool, when an adapter produces metrics
spool.state the next number for a received request
metrics.state what the local derivation has counted
@@ -55,7 +60,8 @@ session.
<session-id>.chain/ its conversation's directory, moved here whole
```
-Child streams are flat siblings of `main`, keyed by agent id. The storage path deliberately does
+Child streams are flat siblings of `main`, keyed by agent id. Provider bodies sit under the
+session, not under a stream, because a body names no stream; the view joins each to its call. The storage path deliberately does
not mirror the source tree: a path must not encode a relationship the pipeline is supposed to
derive. [Export over OpenTelemetry](../setup/export-otlp.md) says how the metrics spool fills and
what `push.state` records. `_scenario/` and `_removed/` exist only in a root a scenario build
@@ -273,6 +279,32 @@ A session a scenario build marked is the one case where asz deletes a source its
the session is sent, a pipeline removes its source first and then the whole session directory, as
[Retention](#retention) describes.
+## Provider bodies
+
+Claude Code writes the bodies of every session into one flat directory, one file per body, and never
+appends to one. Neither kind of cursor fits: an append cursor tracks a position in one growing file,
+and a session of thousands of calls would need thousands of snapshot cursors. So the adapter keeps
+one table for the root, `_provider/seen.state`: for each body file, its size, its modification
+time, the start of its digest, its state, its session, and the ids that join it. The states are
+`landed`, `waiting` for a session, `excluded` by the filter, `unreadable`, `changed` after it landed,
+and `conflict` with a body of the same name the session already holds.
+
+The table is derived. Every landed body is a record whose manifest names its file, its size and its
+digest, so a missing table is rebuilt from the landed records, and the first pass after reads each
+file once more to compare its digest. A file lands before the table says so, and the table says so only after the session's index and
+state are saved: a pass that stops between the two reads the file again, finds the session already
+holds it, lands nothing, and closes the index gap. A `landed` line is
+trusted only while its session's `provider_body/` directory exists, so a body whose session was removed
+lands again. A line whose file and session are both gone is dropped. A landed provider file that does not read is
+left out of a rebuilt table and reported, and every other session's bodies are collected. Every pass
+also compares each session's index with its landed provider files, and brings a session whose index
+is behind up to date, so a pass that stopped between landing and saving the index, or met the session
+busy, leaves no lasting gap.
+
+A pass holds `_provider/.lock` for the table and each session's own lock while it lands into that
+session. It lands every new body of a session into one file, in the order the files were written,
+cut at `max_delta_bytes`.
+
## The index
`index/` holds identifiers and roles, never text: which record is in which stream, which call a
@@ -426,7 +458,7 @@ The removal then goes in this order:
3. `<root>/<session-id>/` is renamed to `_removed/<session-id>/`, and deleted there.
4. `_conversations/<session-id>/` is renamed to `_removed/<session-id>.chain/`, and deleted there.
5. The spool files the session owns are deleted.
-6. The session's lines in `push.state` and `metrics.state` are dropped.
+6. The session's lines in `push.state`, `metrics.state` and `_provider/seen.state` are dropped.
7. The marker is deleted.
Each step comes where it does for a reason:
diff --git a/docs/en/guides/scenario.md b/docs/en/guides/scenario.md
index 1439bef..bcf3f4a 100644
--- a/docs/en/guides/scenario.md
+++ b/docs/en/guides/scenario.md
@@ -170,6 +170,10 @@ and the pass says why once, on standard error, as `kept: every marked session: <
names, `XDG_CONFIG_HOME/claude` and `~/.claude`.
- `DIR/_removed` is a symbolic link, or not a directory.
+A session whose build wrote provider bodies is kept too, with a `kept: <session-id>: <reason>` line,
+while the `claude-code-provider` adapter is switched off or reads anything but
+`DIR/_source/provider-bodies`, and waits while any of its bodies is not landed.
+
One session is kept, with a `kept: <session-id>: <reason>` line, when something about it needs a
person. Examples are a receiver that rejected records of it, a file the build wrote that has
changed or is missing, and a file for it that the build did not write. A kept line is not an error.
@@ -202,6 +206,7 @@ whose names produce the same run id, ignoring letter case. `source-bytes.yaml` i
session: mock-build-and-check # optional; default derived from the steps
title: build and check # optional
interval: 1s # the gap between steps, in every stream
+provider_bodies: false # optional; true writes every call's request and response body
steps:
- input: run the build # a person's message: opens a run and a talk
- inject: {type: skill_listing, text: "skills: 1"}
@@ -328,6 +333,8 @@ checkpoints:
unresolved_kinds: {tool_result: none}
session: {from: +0s, to: +11.1s} # the session node's range, as deltas from --at
view: {state: verified, problems: 0, talks: 3, files: 6, first_talk: {label: run the build, runs: 2}}
+ # with provider_bodies: the landed bodies, and the calls whose request is captured
+ # view: {provider_bodies: 14, captured_prompts: 7, provider_files: 5}
verify: {problems: 0} # what asz verify reports over the root
helped:
lose: [{stream: checker, kind: transcript}] # deleted from the root after this checkpoint's parse
@@ -351,6 +358,11 @@ properties: # all on unless set false
push_follows_the_wire: true
removed_after_sent: true
view_covers_the_session: true
+ changes_leave_the_fold: true # when a tool changes a file
+ provider_bodies_leave_the_fold: true # when the scenario writes provider bodies
+ provider_bodies_rebuild: true # when the scenario writes provider bodies
+collect:
+ max_delta_bytes: 0 # the largest landed file; the build writes it into its configuration
parse:
max_round_bytes: 0 # a parse setting, when the scenario needs one
push:
@@ -398,6 +410,28 @@ nothing, and writing every body back to its path must give a root that verifies
same. `push.kinds` names the file kinds a scenario's push must carry; `all-kinds.yaml` names all
six.
+**Provider bodies.** With `provider_bodies: true`, every call writes the request and the response
+Claude Code would write when `OTEL_LOG_RAW_API_BODIES` names a directory: the whole message list of
+its stream again, the cache marker on the newest message, a billing header naming the prompt and
+the previous call's request, and the session in `metadata.user_id`. A compaction starts the list
+again. A `claude-code` build writes them as files under `DIR/_source/provider-bodies`, each stamped
+with when it was written, and its configuration enables `claude-code-provider` there; an `sd` build
+lands the same bodies in the same order. Such a scenario's message and request ids carry the start of
+the session id, because every session shares the one directory and a runtime's ids are unique across
+sessions. `parts_keep_source_bytes` rebuilds every landed body and compares it with its file, and
+`provider_bodies_leave_the_fold` builds the scenario again without bodies, with the same ids, and
+compares the folds node by node, parent by parent and relation by relation. A node whose id is built
+from a landed position is compared by the record it stands on, since files of another kind landing
+between shift positions. `changes_leave_the_fold` compares the same way.
+`provider_bodies_rebuild` reads every body a call lists the way a reader that loads bodies on demand
+does: only the session's `provider_body` files up to the one the body's `ref` names, in order. Each
+body must rebuild to its manifest's digest. It runs again on the root `repack_keeps_structure` re-cuts,
+where every file and row moved. `collect.max_delta_bytes` sets the largest landed file: the build
+writes it into its configuration, the check collects by that configuration, and an `sd` build cuts its
+provider files at it, so a small value spreads bodies over many files and makes them refer across
+files.
+`tests/scenarios/provider-bodies.yaml` is the example.
+
The removal is checked too. `removed_after_sent` makes full copies of the finished root and runs a
pipeline's pass over each: collect, derive, parse, send to a receiver in the test, then remove.
Nothing is removed before a push, after a refused push, after a partial success, when `push.state`
diff --git a/docs/en/setup/claude-code-provider-bodies.md b/docs/en/setup/claude-code-provider-bodies.md
new file mode 100644
index 0000000..afae949
--- /dev/null
+++ b/docs/en/setup/claude-code-provider-bodies.md
@@ -0,0 +1,125 @@
+# Claude Code Provider Bodies
+
+Claude Code can write every request it sends to its model provider, and every response it gets
+back, to a directory. A request holds what no transcript records: the system prompt, the tool
+schemas, the reminder blocks injected into the first message, the `role:"system"` messages placed
+between the others, the cache markers, the betas and the thinking settings. The
+`claude-code-provider` adapter collects those files, finds the session of each, and lands them
+beside the session's transcripts. The view joins each body to the call it belongs to.
+
+Everything on this page about what Claude Code writes was read from the 2.1.260 code and measured
+on two captures of that version, unless it says otherwise.
+
+## Turn it on
+
+Add one line to the `env` block of `~/.claude/settings.json`, with the directory the adapter reads:
+
+```json
+{ "env": { "OTEL_LOG_RAW_API_BODIES": "file:/Users/me/.claude/asz/provider-bodies" } }
+```
+
+- **The path must be absolute.** Claude Code does not expand `~`, and it resolves a relative path
+ against each session's own working directory, so a relative path writes a directory into every
+ project.
+- **The directory** the adapter reads by default is `asz/provider-bodies` under Claude Code's
+ directory: `CLAUDE_CONFIG_DIR`, else `XDG_CONFIG_HOME/claude`, else `~/.claude`. Set
+ `source_root` on the adapter to read another, and name the same directory in the variable.
+- **No telemetry is needed.** The capture wrote the files with no other `OTEL_*` or telemetry
+ variable set.
+- **The adapter is on by default** in `asz.yaml`, and does nothing while the directory does not
+ exist. Setting the variable is the choice to collect.
+
+`asz sources` says what the adapter sees: the directory, how many requests and responses it holds,
+and which sessions the requests name.
+
+## What the files hold
+
+Claude Code writes one file per body, every session into the same directory.
+
+| | Request | Response |
+| --- | --- | --- |
+| Name | `<random UUID>.request.json` | `<provider request id>.response.json`, or a random UUID when there was none |
+| Content | the request as it was built: `model`, `messages`, `system`, `tools`, `betas`, `metadata`, `max_tokens`, `thinking`, … | the last assistant message, with every content block of the streamed message |
+| Session | `metadata.user_id` is a JSON string holding `session_id` | none |
+| Joins | the first system block names `cc_prompt_id`, the prompt, and `cc_prev_req`, the request id of the previous call of its chain | the name is the request id; `id` is the message id a transcript records |
+
+Thinking text is replaced by `<REDACTED>` in both. Claude Code writes a file with one asynchronous
+write, with no temporary file and no rename, and it never deletes one.
+
+**What is in them besides the conversation.** A request holds the device id and the account id in
+`metadata.user_id`, the person's email and the date in the reminder block, the whole instruction
+files and memory index, the schemas of every MCP tool, and the environment block with paths and
+the operating system. asz lands and sends the bodies as they are, and redacts nothing.
+
+**How large.** In a capture with the full tool list, a main request was 121 to 196 KB, of which the
+tool schemas were 74 KB and the system prompt 10 KB, and a subagent's request 33 to 58 KB. Every
+call sends its chain's whole message list again. Earlier messages come back unchanged, apart from
+the cache marker, which moves to the newest message. So most of a body is bytes the session already
+has.
+
+## What the adapter lands
+
+**A session for each body.** A request names its session; a session id that is not in Claude Code's
+own form is not taken, since it names a directory. A response is claimed by a later request that
+names its request id, or by the one session whose landed transcript holds its message id. The
+transcripts searched are those of sessions with a request written within 24 hours of the response,
+so a response whose session has no collected request waits. Anything else waits, and every pass
+tries again. In the capture of 15 calls, 13 responses were claimed. The other two, the call that names the session and the compaction call, have no call in
+any transcript and are never landed. A body that waits is counted in the pass and does not make the
+pass fail. The pass line counts both, as `bodies_waiting` and `bodies_unreadable`.
+
+The adapter runs after `claude-code-local` in a pass, so a response only a transcript names finds
+its session in the same pass. The session filter is the local adapter's: `include` and `exclude`
+are judged by the directory the session's main transcript sits under. A session whose transcripts were
+pruned since they landed is judged by the directories its landed transcripts name. A session nothing
+names a directory for waits.
+
+**Repeats taken out.** A body is cut into pieces: each tool definition, and every JSON string of
+1 KiB or more. A piece the session already holds becomes a reference to it, and the front a body
+shares with an earlier body of its chain becomes one copy of that front. What is left lands as
+literal bytes. On the two captures, the landed files were 25.5% and 21.6% of the bytes Claude Code
+wrote, and all 42 bodies rebuilt to the files' exact bytes.
+[Session Data](../formats/session-data.md#provider-bodies) describes the record.
+
+**One file per pass.** Every new body of a session in one pass goes into landed files under
+`<session>/provider_body/`, a file ending before a body that would take it past `max_delta_bytes`; a
+single body larger than that lands alone. A body names no stream, so the file is filed under
+the session and the view decides which stream and call a body belongs to.
+
+**Checked before it lands.** Each record is written, read back and rebuilt before it lands. A body
+that is not one JSON object in valid UTF-8, or does not come back whole, lands whole in one unknown
+part, and its manifest says why. `asz verify` rebuilds every landed body and compares its digest.
+
+A file changed after it landed is a conflict: nothing more lands from it, and every pass fails while
+it stays, so a single pass of `asz collect` exits non-zero. Putting the landed bytes back ends it. A
+file that is not one JSON value two minutes after it was last written is counted as unreadable.
+
+## In the view
+
+In the [asz.view](../formats/asz-view.md#provider-bodies) document, a call lists its request and its
+response under `provider_bodies`, each by the landed record it rebuilds from. The document does not
+carry the bodies. A reader loads them when asked: the session's `provider_body` files up to the one
+named, read once and kept, since a body can refer to earlier files. A response joins by its message
+id. A request joins to the
+call whose previous call's response carries the request id the request names, and whose prompt is
+the one it names, when exactly one request and one call carry those two ids. On the capture, all 13
+calls had their request and their response joined, the subagent's included. The request that names the session, and the compaction request, which names no
+prompt, were left unjoined.
+
+## Keeping the directory small
+
+Claude Code never deletes these files, and asz never deletes a source file of a real session. The
+capture wrote 1.7 MB for 15 calls. Delete old files yourself once they are collected, for example
+every file older than 30 days:
+
+```sh
+find /Users/me/.claude/asz/provider-bodies -name '*.json' -mtime +30 -delete
+```
+
+A body deleted before a pass collected it is lost. The pass line counts the bodies still waiting.
+
+## Not measured
+
+Subagents or workflow agents running at the same time, server-side clearing of old tool results,
+the `thread` field that sends only a tail of the messages, retries, a long session, `--resume` and
+`--fork-session`, Bedrock, Vertex and gateways, and Windows.
diff --git a/docs/en/setup/command-line.md b/docs/en/setup/command-line.md
index e892d8d..0ba3b20 100644
--- a/docs/en/setup/command-line.md
+++ b/docs/en/setup/command-line.md
@@ -45,6 +45,11 @@ list is absent when no session spans more than one.
With the `claude-code-changes` adapter enabled, a second table lists the sessions the plugin has
written change records for, with their streams and workspace.
+With the `claude-code-provider` adapter enabled, it names the directory Claude Code writes provider
+bodies to, counts the requests and responses there, and lists the sessions the requests name. A
+response names no session, so it is counted only. See
+[Claude Code Provider Bodies](claude-code-provider-bodies.md).
+
## collect
The pipeline. One pass does three things, in this order:
diff --git a/docs/en/setup/configuration.md b/docs/en/setup/configuration.md
index af1d048..3e1218e 100644
--- a/docs/en/setup/configuration.md
+++ b/docs/en/setup/configuration.md
@@ -179,6 +179,31 @@ Set `source_root` to collect from a copy. The session filters are the ones above
workspace each session's records name. It is on by default because it costs nothing when the
plugin is not installed: there is nothing to discover. It takes no `metrics`.
+## The provider adapter
+
+```yaml
+adapters:
+ - name: claude-code-provider
+ enabled: true
+ source_root: ""
+ include: []
+ exclude:
+ - /private/tmp/**
+ collector:
+ mode: watch
+ interval: 10m
+ max_delta_bytes: 2097152
+```
+
+`claude-code-provider` lands the request and response bodies Claude Code writes for its model
+provider when `OTEL_LOG_RAW_API_BODIES=file:<absolute path>` is in its environment. Empty
+`source_root` resolves `asz/provider-bodies` under the same directory `claude-code-local` resolves;
+the variable must name that directory by its absolute path. The session filters are the ones above,
+judged by the directory each session's main transcript sits under. It runs after the other local
+adapters in a pass. It is on by default because it costs nothing until the variable is set: there
+is nothing to list. It takes no `metrics`. See
+[Claude Code Provider Bodies](claude-code-provider-bodies.md).
+
## The receiver adapter
```yaml
diff --git a/docs/en/setup/export-otlp.md b/docs/en/setup/export-otlp.md
index 149dd9e..aeee8d2 100644
--- a/docs/en/setup/export-otlp.md
+++ b/docs/en/setup/export-otlp.md
@@ -93,7 +93,7 @@ The resource, which names the service a record belongs to:
| Attribute | Value |
| --- | --- |
-| `service.name` | `export.otlp.service_name`, or when empty the runtime that produced each session, read off its landed header's adapter: `Claude Code` for `claude-code-local`, `Mock Agent` for `mock`. One service per kind of agent, and a root that holds both is pushed as both |
+| `service.name` | `export.otlp.service_name`, or when empty the runtime that produced each session, read off its landed header's adapter: `Claude Code` for `claude-code-local`, `claude-code-changes` and `claude-code-provider`, `Mock Agent` for `mock`. One service per kind of agent, and a root that holds both is pushed as both |
| `service.instance.id` | `export.otlp.instance_id`: who is pushing, in words the people reading the receiver recognise. A receiver lists it under the service as the instance, so put your mailbox, your name, or the machine there. Empty means `user@host` of the machine running `asz push`, which is stable across restarts. The session a record belongs to is on the record as `asz.session` |
| `service.layer` | `export.otlp.layer`, `AI_AGENT` by default, the layer the receiver places the service in. The OAP selects its rules by layer, and a layer name is upper case with underscores |
| `telemetry.sdk.name` | `asz`, so a receiver can tell these records apart from any other source |
@@ -109,7 +109,7 @@ body:
| `asz.format` | `sd` for a landed file, `sf` for a round |
| `asz.format.version` | the version in the file's first line: `sd/1` or `sf/1` |
| `asz.file` | the file's path relative to the storage root |
-| `asz.file.kind` | for `sd`, the header's kind: `transcript`, `agent_meta`, `journal`, `workflow_manifest`, `workflow_script`; for `sf`, `round` |
+| `asz.file.kind` | for `sd`, the header's kind: `transcript`, `agent_meta`, `journal`, `workflow_manifest`, `workflow_script`, `changes`, `provider_body`; for `sf`, `round` |
| `asz.file.digest` | the file's SHA-256, the digest of the body as received |
| `asz.lines` | how many lines the body has, the header and the closing line included |
| `asz.session` | the session the file belongs to; for `sf`, the session the round was assembled from |
@@ -117,7 +117,7 @@ body:
| `asz.session.from_time`, `asz.session.through_time` | for `sf` only: the session's own range as of that round, when it began and its last activity so far. A landed file never carries it: it can travel before any round exists, and the last activity keeps moving, so the value there would be missing or stale |
| `asz.conversation.title`, `asz.conversation.talks`, `asz.conversation.steps`, `asz.conversation.streams`, `asz.conversation.segments`, `asz.conversation.unresolved` | for `sf` only: what a list of conversations shows, as of that round, copied off the round's header. A receiver lists conversations off its newest round per conversation and never folds |
| `asz.seq` | for `sd`: the landed sequence. With the session it names the file a round's `{seq, row}` reference points at, and the row is a line of the body |
-| `asz.stream`, `asz.run` | for `sd`: the stream or workflow run the file belongs to |
+| `asz.stream`, `asz.run` | for `sd`: the stream or workflow run the file belongs to. A `provider_body` file belongs to the session and carries neither |
| `asz.conversation`, `asz.round` | for `sf`: the conversation and the round number |
The record's time is chosen so a receiver can bound a read by a range it already holds. A landed
diff --git a/docs/menu.yml b/docs/menu.yml
index 8f4c447..c612ee1 100644
--- a/docs/menu.yml
+++ b/docs/menu.yml
@@ -38,6 +38,8 @@ catalog:
path: /en/setup/configuration
- name: Claude Code Plugin
path: /en/setup/claude-code-plugin
+ - name: Claude Code Provider Bodies
+ path: /en/setup/claude-code-provider-bodies
- name: Command Line
path: /en/setup/command-line
- name: Container Image
diff --git a/internal/adapters/claudecodeprovider/claude.go b/internal/adapters/claudecodeprovider/claude.go
new file mode 100644
index 0000000..27bcde8
--- /dev/null
+++ b/internal/adapters/claudecodeprovider/claude.go
@@ -0,0 +1,124 @@
+// 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 claudecodeprovider
+
+import (
+ "encoding/json"
+ "regexp"
+ "strings"
+
+ "github.com/apache/skywalking-ai-sessionizer/pkg/providerbody"
+)
+
+// This file is the whole of what the adapter knows about how Claude Code names
+// and fills its body files. Everything past it reads the model's words.
+
+// ID is the record id of a body file: its name without the .json extension,
+// such as "fcc261af-….request" or "req_011C….response". The role stays in
+// the id, so a request and a response never share one.
+func ID(file string) string { return strings.TrimSuffix(file, ".json") }
+
+// RoleOf reports the role a file name gives a body, or "" for a name Claude
+// Code does not write.
+func RoleOf(file string) string {
+ switch {
+ case strings.HasSuffix(file, ".request.json"):
+ return providerbody.RoleRequest
+ case strings.HasSuffix(file, ".response.json"):
+ return providerbody.RoleResponse
+ }
+ return ""
+}
+
+var (
+ promptIDRe = regexp.MustCompile(`cc_prompt_id=([^;\s]+)`)
+ prevReqRe = regexp.MustCompile(`cc_prev_req=([^;\s]+)`)
+ uuidRe = regexp.MustCompile(`^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$`)
+)
+
+// Lift reads the keys from a body Claude Code wrote under file, in the
+// model's words. A value the body does not carry stays empty; nothing is
+// inferred.
+//
+// For Claude Code 2.1.260: metadata.user_id is a JSON string holding
+// session_id; the first system block is a billing header holding
+// cc_prompt_id, the run, and cc_prev_req, the previous request; a response's
+// id is the call; a response file is named by the provider's request id, or
+// by a random UUID when Claude Code had none.
+func Lift(file string, body []byte) providerbody.Keys {
+ var k providerbody.Keys
+ role := RoleOf(file)
+ if role == providerbody.RoleResponse {
+ // The name holds even when the body is not what a response should be.
+ if stem := strings.TrimSuffix(file, ".response.json"); !uuidRe.MatchString(stem) {
+ k.Request = stem
+ }
+ }
+ var head map[string]json.RawMessage
+ if json.Unmarshal(body, &head) != nil {
+ return k
+ }
+ str := func(raw json.RawMessage) string {
+ var v string
+ _ = json.Unmarshal(raw, &v)
+ return v
+ }
+ k.Model = str(head["model"])
+ switch role {
+ case providerbody.RoleRequest:
+ var meta map[string]json.RawMessage
+ _ = json.Unmarshal(head["metadata"], &meta)
+ var user struct {
+ SessionID string `json:"session_id"`
+ }
+ if json.Unmarshal([]byte(str(meta["user_id"])), &user) == nil {
+ k.Session = user.SessionID
+ }
+ header := firstSystemText(head["system"])
+ if m := promptIDRe.FindStringSubmatch(header); m != nil {
+ k.Run = m[1]
+ }
+ if m := prevReqRe.FindStringSubmatch(header); m != nil {
+ k.PreviousRequest = m[1]
+ }
+ case providerbody.RoleResponse:
+ k.Call = str(head["id"])
+ }
+ return k
+}
+
+// BodyOf is a body file as a session cuts it.
+func BodyOf(file string, body []byte) providerbody.Body {
+ return providerbody.Body{ID: ID(file), Role: RoleOf(file), Src: file, Keys: Lift(file, body), Bytes: body}
+}
+
+// firstSystemText is the text of the first system block, or the system
+// prompt itself when it is one string.
+func firstSystemText(raw json.RawMessage) string {
+ var s string
+ if json.Unmarshal(raw, &s) == nil {
+ return s
+ }
+ var blocks []struct {
+ Text string `json:"text"`
+ }
+ if json.Unmarshal(raw, &blocks) == nil && len(blocks) > 0 {
+ return blocks[0].Text
+ }
+ return ""
+}
diff --git a/internal/adapters/claudecodeprovider/claude_test.go b/internal/adapters/claudecodeprovider/claude_test.go
new file mode 100644
index 0000000..41a90c0
--- /dev/null
+++ b/internal/adapters/claudecodeprovider/claude_test.go
@@ -0,0 +1,48 @@
+// 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 claudecodeprovider_test
+
+import (
+ "testing"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecodeprovider"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/providerbody"
+)
+
+func TestLiftReadsOnlyWhatTheBodyCarries(t *testing.T) {
+ request := []byte(`{"model":"claude-opus-5","messages":[],"system":[{"type":"text","text":"x-anthropic-billing-header: cc_version=2.1.260; cc_prev_req=req_prev; cc_prompt_id=bdee6995;"}],` +
+ `"metadata":{"user_id":"{\"device_id\":\"d\",\"session_id\":\"1a3cfda7-659c-474a-a448-cebc9370530f\"}"}}`)
+ k := claudecodeprovider.Lift("a640339e-9d1c-4b6c-9a1e-2f9b6c1d0e11.request.json", request)
+ want := providerbody.Keys{Model: "claude-opus-5", Session: "1a3cfda7-659c-474a-a448-cebc9370530f", Run: "bdee6995", PreviousRequest: "req_prev"}
+ if k != want {
+ t.Fatalf("request keys %+v, want %+v", k, want)
+ }
+ k = claudecodeprovider.Lift("req_011Cf4.response.json", []byte(`{"id":"msg_1","model":"claude-opus-5"}`))
+ if k != (providerbody.Keys{Model: "claude-opus-5", Request: "req_011Cf4", Call: "msg_1"}) {
+ t.Fatalf("response keys %+v", k)
+ }
+ // A response Claude Code had no request id for is named by a random UUID,
+ // which is no provider request id.
+ if k := claudecodeprovider.Lift("5d0c2d2a-0b8e-4f6a-9a55-3f1c1a0d9e21.response.json", []byte(`{"id":"msg_2"}`)); k.Request != "" {
+ t.Fatalf("a UUID file name was taken for a request id: %+v", k)
+ }
+ // The name holds even when the body is not an object.
+ if k := claudecodeprovider.Lift("req_x.response.json", []byte(`[]`)); k.Request != "req_x" {
+ t.Fatalf("a response that is no object lost its request id: %+v", k)
+ }
+}
diff --git a/internal/adapters/claudecodeprovider/claudecodeprovider.go b/internal/adapters/claudecodeprovider/claudecodeprovider.go
new file mode 100644
index 0000000..19bb99f
--- /dev/null
+++ b/internal/adapters/claudecodeprovider/claudecodeprovider.go
@@ -0,0 +1,97 @@
+// 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 claudecodeprovider implements the claude-code-provider adapter: a
+// pull collector for the request and response bodies Claude Code exchanges
+// with its model provider.
+//
+// Claude Code writes them itself when OTEL_LOG_RAW_API_BODIES is set to
+// file:<dir>, one file per body, every session into the same directory. A
+// request is named by a random UUID and carries its session in
+// metadata.user_id; a response is named by the provider's request id and
+// carries no session. This adapter lists that directory, finds each body's
+// session from what the bodies and the landed transcripts say, and lands the
+// bodies of a session as provider_body records under the session's
+// provider directory, with what the session already holds taken out.
+package claudecodeprovider
+
+import (
+ "os"
+ "path/filepath"
+
+ "github.com/apache/skywalking-ai-sessionizer/pkg/model"
+)
+
+// Name is the adapter identifier used in configuration and in landed headers.
+const Name = "claude-code-provider"
+
+// Version is the adapter's contract version, recorded in landed headers.
+const Version = "0.1.0"
+
+// Dialect names the vocabulary the records are read as: the bodies as Claude
+// Code wrote them, cut by pkg/providerbody.
+const Dialect = "claude-code-provider/1"
+
+// RuntimeName is the service the records are attributed to when pushed.
+const RuntimeName = "Claude Code"
+
+// ResolveSourceRoot determines where Claude Code is told to write the bodies:
+// asz/provider-bodies under the directory Claude Code itself resolves. An
+// explicit override wins. Claude Code resolves a relative path against each
+// session's own working directory, so the variable must name this directory
+// by its absolute path.
+func ResolveSourceRoot(override string) (string, error) {
+ if override != "" {
+ return filepath.Abs(override)
+ }
+ if dir := os.Getenv("CLAUDE_CONFIG_DIR"); dir != "" {
+ return filepath.Join(dir, "asz", "provider-bodies"), nil
+ }
+ if dir := os.Getenv("XDG_CONFIG_HOME"); dir != "" {
+ return filepath.Join(dir, "claude", "asz", "provider-bodies"), nil
+ }
+ home, err := os.UserHomeDir()
+ if err != nil {
+ return "", err
+ }
+ return filepath.Join(home, ".claude", "asz", "provider-bodies"), nil
+}
+
+// Glossary says what the bodies call each name the model uses. A body is
+// evidence beside a call, not a step, so it names a session and, for a
+// response, the call; everything else the model names is structure a body
+// does not record.
+func Glossary() *model.Glossary {
+ native := map[string][2]string{
+ model.KindSession: {"metadata.user_id → session_id", "a request body"},
+ model.RoleRun: {"cc_prompt_id", "the billing header, the first system block of a request body"},
+ model.RoleCall: {"id", "a response body"},
+ model.RoleID: {"the file name without .json", "the body's file"},
+ model.RoleModel: {"model", "a request or response body"},
+ }
+ var terms []model.Term
+ for _, name := range model.Vocabulary() {
+ t := model.Term{Unified: name}
+ if n, ok := native[name]; ok {
+ t.Native, t.Where = n[0], n[1]
+ } else {
+ t.Note = "not recorded; a provider body is evidence beside a call, not a step"
+ }
+ terms = append(terms, t)
+ }
+ return model.NewGlossary(Dialect, terms...)
+}
diff --git a/internal/adapters/claudecodeprovider/collect.go b/internal/adapters/claudecodeprovider/collect.go
new file mode 100644
index 0000000..090376e
--- /dev/null
+++ b/internal/adapters/claudecodeprovider/collect.go
@@ -0,0 +1,632 @@
+// 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 claudecodeprovider
+
+import (
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "os"
+ "path/filepath"
+ "regexp"
+ "sort"
+ "time"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecode"
+ "github.com/apache/skywalking-ai-sessionizer/internal/index"
+ "github.com/apache/skywalking-ai-sessionizer/internal/storage"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/providerbody"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+)
+
+// prefix is the landed file name prefix, which is also the kind.
+const prefix = string(sessiondata.KindProviderBody)
+
+// Settle is how long a file that is not yet one JSON value is left alone
+// before it is called unreadable. Claude Code writes a body with one
+// asynchronous write and no rename, so a pass can meet it half written.
+const Settle = 2 * time.Minute
+
+// Window bounds the sessions a response is looked for in: those with a
+// request whose file was written within this long of the response's.
+const Window = 24 * time.Hour
+
+// Verdict is what a session filter says about a session.
+type Verdict int
+
+const (
+ // Collect lands the session's bodies.
+ Collect Verdict = iota
+ // Exclude leaves them out: include and exclude say so.
+ Exclude
+ // Wait holds them: nothing yet says where the session was run, so the
+ // filter cannot judge it.
+ Wait
+)
+
+// Filter judges a session by its id.
+type Filter func(session string) Verdict
+
+// Collector lands provider bodies into a storage zone.
+type Collector struct {
+ SourceRoot string
+ Zone *storage.Zone
+ MaxDelta int64
+ Now func() time.Time
+}
+
+// Stats summarises one collection pass. The fields the local adapter reports
+// mean the same here, so a caller can add the passes together.
+type Stats struct {
+ Sessions int
+ SourcesSeen int
+ SourcesLanded int
+ Records int
+ Bytes int64
+ Conflicts int
+ Busy int
+ Pending int
+ Indexed int
+ Reindexed int
+ Errors []error
+ Changed []string
+ // Waiting counts bodies no session claims yet, and Unreadable the files
+ // that are not one JSON value. Neither makes a pass incomplete: a body
+ // the runtime wrote for no call a transcript records, such as the one
+ // that names a session, is never claimed.
+ Waiting int
+ Unreadable int
+}
+
+// Complete reports whether the pass collected everything it could.
+func (s *Stats) Complete() bool {
+ return s.Pending == 0 && s.Conflicts == 0 && len(s.Errors) == 0
+}
+
+// New returns a Collector with sensible defaults.
+func New(sourceRoot string, zone *storage.Zone, maxDelta int64) *Collector {
+ if maxDelta <= 0 {
+ maxDelta = 2 << 20
+ }
+ return &Collector{SourceRoot: sourceRoot, Zone: zone, MaxDelta: maxDelta, Now: time.Now}
+}
+
+// nameRe is the file names Claude Code writes: a request under a random UUID,
+// a response under the provider's request id, or a UUID when it had none.
+var nameRe = regexp.MustCompile(`^[A-Za-z0-9_-]+\.(request|response)\.json$`)
+
+// file is one body file found in the source directory.
+type file struct {
+ name string
+ path string
+ size int64
+ mtime int64
+ body []byte
+ keys providerbody.Keys
+ e *entry
+}
+
+// CollectAll lands whatever bodies are new, into the sessions they belong to.
+func (c *Collector) CollectAll(filter Filter) (*Stats, error) {
+ st := &Stats{}
+ items, err := os.ReadDir(c.SourceRoot)
+ if err != nil {
+ if errors.Is(err, os.ErrNotExist) {
+ return st, nil
+ }
+ return nil, err
+ }
+ lock, err := storage.LockProvider(c.Zone.Root())
+ if err != nil {
+ if errors.Is(err, storage.ErrProviderBusy) {
+ st.Busy++
+ return st, nil
+ }
+ return nil, err
+ }
+ defer func() { _ = lock.Unlock() }()
+ sn, err := loadSeen(c.Zone)
+ if err != nil {
+ return nil, err
+ }
+ now := c.Now()
+
+ // A line whose file is gone and whose session is gone describes nothing
+ // any more: a body pruned from the directory after its session was
+ // removed.
+ present := make(map[string]bool, len(items))
+ for _, it := range items {
+ present[it.Name()] = true
+ }
+ for name, e := range sn.entries {
+ if !present[name] && (e.session == "" || !c.sessionHolds(e.session)) {
+ delete(sn.entries, name)
+ sn.dirty = true
+ }
+ }
+
+ var fresh []*file
+ for _, it := range items {
+ if it.IsDir() || !nameRe.MatchString(it.Name()) {
+ continue
+ }
+ info, err := it.Info()
+ if err != nil {
+ if !errors.Is(err, os.ErrNotExist) {
+ // Not gone between the listing and the stat: unreachable, and
+ // a pass that skipped it must not look complete.
+ st.Errors = append(st.Errors, fmt.Errorf("claudecodeprovider: %s: %w", it.Name(), err))
+ }
+ continue
+ }
+ st.SourcesSeen++
+ f := &file{name: it.Name(), path: filepath.Join(c.SourceRoot, it.Name()), size: info.Size(), mtime: info.ModTime().UnixNano()}
+ f.e = sn.entries[f.name]
+ if f.e != nil && f.e.size == f.size && f.e.mtime == f.mtime {
+ switch f.e.state {
+ case stateLanded:
+ if c.sessionHolds(f.e.session) {
+ continue
+ }
+ case stateConflict, stateChanged:
+ // Counted on every pass it stays, so a single pass keeps
+ // failing until a person looks.
+ st.Conflicts++
+ continue
+ case stateUnreadable:
+ st.Unreadable++
+ continue
+ }
+ }
+ if err := c.read(f, sn, now, st); err != nil {
+ st.Errors = append(st.Errors, err)
+ continue
+ }
+ if f.body != nil {
+ fresh = append(fresh, f)
+ }
+ }
+
+ bySession := c.attribute(fresh, sn, filter, st)
+ st.Errors = append(st.Errors, sn.damaged...)
+ // A pass that stopped after landing and before saving an index, or whose
+ // session was busy when the gap was to be closed, left an index behind
+ // what is landed. No new body of that session need ever come to close
+ // it, so every pass looks at each session the table names.
+ checked := map[string]bool{}
+ for _, e := range sn.entries {
+ if e.session == "" || checked[e.session] {
+ continue
+ }
+ checked[e.session] = true
+ if _, fresh := bySession[e.session]; !fresh && c.indexBehind(e.session) {
+ bySession[e.session] = nil
+ }
+ }
+ ids := make([]string, 0, len(bySession))
+ for id := range bySession {
+ ids = append(ids, id)
+ }
+ sort.Strings(ids)
+ for _, id := range ids {
+ st.Sessions++
+ if err := c.collectSession(id, bySession[id], sn, st, now); err != nil {
+ st.Errors = append(st.Errors, fmt.Errorf("session %s: %w", id, err))
+ }
+ }
+ if sn.dirty {
+ if err := sn.save(c.Zone, now); err != nil {
+ return st, err
+ }
+ }
+ return st, nil
+}
+
+// indexBehind reports whether a session's index covers less than its landed
+// provider files.
+func (c *Collector) indexBehind(session string) bool {
+ items, err := os.ReadDir(c.Zone.ProviderDir(session))
+ if err != nil {
+ return false
+ }
+ var top uint64
+ for _, it := range items {
+ if seq, ok := storage.LandedFileSeq(it.Name()); ok && seq > top {
+ top = seq
+ }
+ }
+ if top == 0 {
+ return false
+ }
+ ix, err := storage.LoadIndexState(c.Zone.IndexStatePath(session), session)
+ return err != nil || ix.Schema != index.Schema || ix.IndexedSeq < top
+}
+
+// sessionHolds reports whether a session's provider directory exists. A
+// landed entry whose session was removed lands again.
+func (c *Collector) sessionHolds(session string) bool {
+ _, err := os.Stat(c.Zone.ProviderDir(session))
+ return err == nil
+}
+
+// read reads a file that is new, changed, or not yet landed, and decides what
+// its bytes allow.
+func (c *Collector) read(f *file, sn *seen, now time.Time, st *Stats) error {
+ body, err := os.ReadFile(f.path)
+ if err != nil {
+ if errors.Is(err, os.ErrNotExist) {
+ return nil
+ }
+ return err
+ }
+ sum := providerbody.Digest(body)[:12]
+ if f.e != nil && (f.e.state == stateLanded || f.e.state == stateChanged) && c.sessionHolds(f.e.session) {
+ if f.e.sha == sum {
+ // The bytes that landed: a table rebuilt from landed records, a
+ // file touched, or a changed file put back.
+ f.e.state, f.e.size, f.e.mtime = stateLanded, int64(len(body)), f.mtime
+ sn.dirty = true
+ return nil
+ }
+ first := f.e.state == stateLanded
+ f.e.state, f.e.size, f.e.mtime = stateChanged, f.size, f.mtime
+ sn.dirty = true
+ st.Conflicts++
+ if first {
+ return fmt.Errorf("claudecodeprovider: %s changed after it landed", f.name)
+ }
+ return nil
+ }
+ if f.e != nil && f.e.state == stateConflict {
+ if f.e.sha == sum {
+ f.e.size, f.e.mtime = f.size, f.mtime
+ sn.dirty = true
+ st.Conflicts++
+ return nil
+ }
+ // Other bytes than the ones in conflict: read as new.
+ delete(sn.entries, f.name)
+ f.e = nil
+ sn.dirty = true
+ }
+ if !json.Valid(body) {
+ if now.Sub(time.Unix(0, f.mtime)) < Settle {
+ return nil // still being written
+ }
+ sn.entries[f.name] = &entry{name: f.name, size: f.size, mtime: f.mtime, sha: sum, state: stateUnreadable}
+ sn.dirty = true
+ st.Unreadable++
+ return nil
+ }
+ f.body = body
+ f.keys = Lift(f.name, body)
+ return nil
+}
+
+// attribute finds each body's session and applies the filter. A request
+// names its session. A response is claimed by a later request that names its
+// request id, or by the one session whose transcript holds its message id.
+// Anything else waits.
+func (c *Collector) attribute(fresh []*file, sn *seen, filter Filter, st *Stats) map[string][]*file {
+ // Which sessions a request id is named by, from landed requests and the
+ // ones read this pass.
+ namedBy := map[string]map[string]bool{}
+ addNamed := func(prev, session string) {
+ if prev == "" || session == "" {
+ return
+ }
+ if namedBy[prev] == nil {
+ namedBy[prev] = map[string]bool{}
+ }
+ namedBy[prev][session] = true
+ }
+ type near struct {
+ session string
+ mtime int64
+ }
+ var requests []near
+ for _, e := range sn.entries {
+ if e.state == stateLanded && RoleOf(e.name) == providerbody.RoleRequest {
+ addNamed(e.keys.PreviousRequest, e.session)
+ requests = append(requests, near{e.session, e.mtime})
+ }
+ }
+ for _, f := range fresh {
+ if RoleOf(f.name) == providerbody.RoleRequest && claudecode.IsSessionID(f.keys.Session) {
+ addNamed(f.keys.PreviousRequest, f.keys.Session)
+ requests = append(requests, near{f.keys.Session, f.mtime})
+ }
+ }
+ indexes := map[string]*index.Index{}
+ holdsCall := func(session, call string) bool {
+ ix, ok := indexes[session]
+ if !ok {
+ ix, _, _ = index.Load(c.Zone.IndexDir(session), session)
+ indexes[session] = ix
+ }
+ return ix != nil && len(ix.ProviderCall(call)) > 0
+ }
+
+ out := map[string][]*file{}
+ for _, f := range fresh {
+ session := ""
+ switch RoleOf(f.name) {
+ case providerbody.RoleRequest:
+ // The session names a directory, so only a session id in the
+ // runtime's own form is taken.
+ if claudecode.IsSessionID(f.keys.Session) {
+ session = f.keys.Session
+ }
+ case providerbody.RoleResponse:
+ if names := namedBy[f.keys.Request]; len(names) == 1 {
+ for s := range names {
+ session = s
+ }
+ } else if len(names) == 0 && f.keys.Call != "" {
+ found := map[string]bool{}
+ for _, r := range requests {
+ if r.session == "" || found[r.session] {
+ continue
+ }
+ // A table rebuilt from landed records has no times; its
+ // sessions are all candidates.
+ if r.mtime >= 0 && absDuration(f.mtime-r.mtime) > Window {
+ continue
+ }
+ if holdsCall(r.session, f.keys.Call) {
+ found[r.session] = true
+ }
+ }
+ if len(found) == 1 {
+ for s := range found {
+ session = s
+ }
+ }
+ }
+ }
+ state := stateWaiting
+ if session != "" {
+ switch filter(session) {
+ case Collect:
+ out[session] = append(out[session], f)
+ continue
+ case Exclude:
+ state = stateExcluded
+ }
+ }
+ if state == stateWaiting {
+ st.Waiting++
+ }
+ sn.entries[f.name] = &entry{name: f.name, size: f.size, mtime: f.mtime, sha: providerbody.Digest(f.body)[:12],
+ state: state, session: session, keys: f.keys}
+ sn.dirty = true
+ }
+ return out
+}
+
+func absDuration(ns int64) time.Duration {
+ if ns < 0 {
+ ns = -ns
+ }
+ return time.Duration(ns)
+}
+
+// collectSession lands one session's new bodies under the session's lock,
+// with the session's own sequence, cut into files at MaxDelta.
+func (c *Collector) collectSession(session string, files []*file, sn *seen, st *Stats, now time.Time) error {
+ sessionDir := c.Zone.SessionDir(session)
+ lock, err := storage.LockSession(sessionDir)
+ if err != nil {
+ if errors.Is(err, storage.ErrSessionBusy) {
+ st.Busy++
+ return nil
+ }
+ return err
+ }
+ defer func() { _ = lock.Unlock() }()
+
+ ixState, err := storage.LoadIndexState(c.Zone.IndexStatePath(session), session)
+ if err != nil {
+ return err
+ }
+ statePath := c.Zone.SessionStatePath(session)
+ state, err := storage.LoadSessionState(statePath, session)
+ if err != nil {
+ return err
+ }
+ if err := state.RecoverNextSeq(sessionDir); err != nil {
+ return err
+ }
+ var ix *index.Index
+ if ixState.Schema == index.Schema {
+ if loaded, ok, lerr := index.Load(c.Zone.IndexDir(session), session); lerr == nil && ok {
+ ix = loaded
+ }
+ }
+ if ix == nil {
+ ix = index.New(session)
+ ixState = storage.NewIndexState(session)
+ }
+ // Close any gap between what has landed and what the index covers, as
+ // the other adapters do.
+ reindexed := false
+ if landedTo := state.NextSeq - 1; ixState.IndexedSeq < landedTo {
+ n, rerr := index.Rebuild(c.Zone, session, ix, ixState.IndexedSeq)
+ if rerr != nil {
+ return rerr
+ }
+ st.Reindexed += n
+ ixState.IndexedSeq = landedTo
+ reindexed = true
+ }
+
+ held, err := c.loadSession(session)
+ if err != nil {
+ return err
+ }
+ sort.SliceStable(files, func(i, j int) bool {
+ if files[i].mtime != files[j].mtime {
+ return files[i].mtime < files[j].mtime
+ }
+ return files[i].name < files[j].name
+ })
+
+ var batch []*sessiondata.Record
+ var batchFiles []*file
+ // done are the files landed in this session. The table says so only
+ // once the index and the session state that describe them are saved:
+ // a pass that stops before then reads the files again, finds the
+ // session holds them, and closes the index gap on the way.
+ var done []*file
+ var size int64
+ landed := false
+ flush := func() error {
+ if len(batch) == 0 {
+ return nil
+ }
+ seq := state.Take()
+ name := storage.LandedName(prefix, storage.Stamp(now), seq)
+ hdr := &sessiondata.Header{
+ H: 1, Seq: seq, At: now.UTC().Format(time.RFC3339Nano),
+ Kind: sessiondata.KindProviderBody, Adapter: Name + "/" + Version, Dialect: Dialect,
+ Src: ".", Session: session,
+ }
+ err := storage.WriteAtomic(filepath.Join(c.Zone.ProviderDir(session), name), storage.PermLanded, func(w io.Writer) error {
+ rw, err := sessiondata.NewWriter(w, hdr)
+ if err != nil {
+ return err
+ }
+ for row, rec := range batch {
+ if err := rw.Write(rec); err != nil {
+ return err
+ }
+ e, blocks := index.FromRecord(ix, hdr, rec, uint32(seq), uint32(row+1))
+ ix.Append(e, blocks...)
+ }
+ return rw.Close()
+ })
+ if err != nil {
+ return err
+ }
+ for _, f := range batchFiles {
+ st.Bytes += f.size
+ }
+ done = append(done, batchFiles...)
+ st.Records += len(batch)
+ st.SourcesLanded += len(batchFiles)
+ batch, batchFiles, size = nil, nil, 0
+ landed = true
+ return nil
+ }
+ for _, f := range files {
+ rec, err := held.Encode(providerbody.Body{ID: ID(f.name), Role: RoleOf(f.name), Src: f.name, Keys: f.keys, Bytes: f.body})
+ switch {
+ case errors.Is(err, providerbody.ErrRepeat):
+ done = append(done, f)
+ continue
+ case err != nil:
+ sn.entries[f.name] = &entry{name: f.name, size: f.size, mtime: f.mtime, sha: providerbody.Digest(f.body)[:12],
+ state: stateConflict, session: session, keys: f.keys}
+ sn.dirty = true
+ st.Conflicts++
+ st.Errors = append(st.Errors, fmt.Errorf("claudecodeprovider: %s: %w", f.name, err))
+ continue
+ }
+ // A file ends before a body that would take it past the budget, so
+ // only a body larger than the budget on its own lands in a larger file.
+ n := providerbody.RecordBytes(rec)
+ if len(batch) > 0 && providerbody.FileOverhead+size+n > c.MaxDelta {
+ if err := flush(); err != nil {
+ return err
+ }
+ }
+ batch = append(batch, rec)
+ batchFiles = append(batchFiles, f)
+ size += n
+ }
+ if err := flush(); err != nil {
+ return err
+ }
+
+ if landed || reindexed {
+ st.Changed = append(st.Changed, session)
+ ixState.Schema = index.Schema
+ ixState.IndexedSeq = state.NextSeq - 1
+ ixState.Entries = len(ix.Entries)
+ ixState.Blocks = len(ix.Blocks)
+ ixState.Strings = ix.Strings.Len()
+ if err := ix.Write(c.Zone.IndexDir(session)); err != nil {
+ return err
+ }
+ if err := ixState.Save(c.Zone.IndexStatePath(session), now); err != nil {
+ return err
+ }
+ st.Indexed += ixState.Entries
+ }
+ state.LastScan = now.UTC().Format(time.RFC3339Nano)
+ if err := state.Save(statePath, now); err != nil {
+ return err
+ }
+ for _, f := range done {
+ c.markLanded(sn, f, session)
+ }
+ if sn.dirty {
+ return sn.save(c.Zone, now)
+ }
+ return nil
+}
+
+func (c *Collector) markLanded(sn *seen, f *file, session string) {
+ sn.entries[f.name] = &entry{name: f.name, size: f.size, mtime: f.mtime, sha: providerbody.Digest(f.body)[:12],
+ state: stateLanded, session: session, keys: f.keys}
+ sn.dirty = true
+}
+
+// loadSession reads a session's landed provider bodies, in landing order, so
+// a new body is cut against everything the session already holds.
+//
+// A record that does not hold together is left out: a file it names was
+// lost, cut short, or kept in another storage stage than this root. A new
+// body is then cut only against what does, and lands whole where nothing
+// earlier is left to share, so a damaged or partly restored session still
+// takes new bodies. asz verify reports the damage itself.
+func (c *Collector) loadSession(session string) (*providerbody.Session, error) {
+ held := providerbody.NewSession()
+ files, err := storage.LandedFiles(c.Zone, session)
+ if err != nil {
+ return nil, err
+ }
+ for _, lf := range files {
+ if lf.Stream != "" || lf.RunID != "" {
+ continue
+ }
+ // A file that does not read, whole and with its digest, holds nothing
+ // a new body may refer to.
+ var recs []*sessiondata.Record
+ if err := readLanded(lf.Path, func(_ sessiondata.Header, rec *sessiondata.Record) {
+ recs = append(recs, rec)
+ }); err != nil {
+ continue
+ }
+ for _, rec := range recs {
+ _ = held.Add(rec)
+ }
+ }
+ return held, nil
+}
diff --git a/internal/adapters/claudecodeprovider/collect_test.go b/internal/adapters/claudecodeprovider/collect_test.go
new file mode 100644
index 0000000..ae3d0b5
--- /dev/null
+++ b/internal/adapters/claudecodeprovider/collect_test.go
@@ -0,0 +1,516 @@
+// 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 claudecodeprovider_test
+
+import (
+ "bytes"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecodeprovider"
+ "github.com/apache/skywalking-ai-sessionizer/internal/index"
+ "github.com/apache/skywalking-ai-sessionizer/internal/storage"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/providerbody"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+)
+
+const (
+ sessionA = "11111111-2222-4333-8444-555555555555"
+ sessionB = "66666666-7777-4888-8999-aaaaaaaaaaaa"
+)
+
+var base = time.Date(2026, 9, 15, 9, 0, 0, 0, time.UTC)
+
+type fixture struct {
+ t *testing.T
+ src string
+ zone *storage.Zone
+ col *claudecodeprovider.Collector
+ now time.Time
+ n int
+}
+
+func newFixture(t *testing.T) *fixture {
+ dir := t.TempDir()
+ f := &fixture{t: t, src: filepath.Join(dir, "bodies"), zone: storage.NewZone(filepath.Join(dir, "root")), now: base.Add(time.Hour)}
+ if err := os.MkdirAll(f.src, 0o755); err != nil {
+ t.Fatal(err)
+ }
+ f.col = claudecodeprovider.New(f.src, f.zone, 2<<20)
+ f.col.Now = func() time.Time { return f.now }
+ return f
+}
+
+// request writes a request body of session under a UUID name, naming prev as
+// the previous request, with enough history to be worth cutting.
+func (f *fixture) request(session, prev string, turns int) string {
+ f.n++
+ msgs := []any{map[string]any{"role": "user", "content": []any{
+ map[string]any{"type": "text", "text": "<system-reminder>" + strings.Repeat(" instructions", 600) + "</system-reminder>"},
+ map[string]any{"type": "text", "text": "start"},
+ }}}
+ for i := 0; i < turns; i++ {
+ msgs = append(msgs, map[string]any{"role": "assistant", "content": fmt.Sprintf("answer %d", i)},
+ map[string]any{"role": "user", "content": fmt.Sprintf("question %d %s", i, strings.Repeat("x", 400))})
+ }
+ header := "x-anthropic-billing-header: cc_version=2.1.260;"
+ if prev != "" {
+ header += " cc_prev_req=" + prev + ";"
+ }
+ header += " cc_prompt_id=p1;"
+ user, _ := json.Marshal(map[string]string{"session_id": session})
+ body, _ := json.Marshal(map[string]any{
+ "model": "claude-opus-5", "messages": msgs,
+ "system": []any{map[string]any{"type": "text", "text": header}, map[string]any{"type": "text", "text": strings.Repeat("You are an agent. ", 400)}},
+ "tools": []any{map[string]any{"name": "Read", "description": strings.Repeat("reads a file. ", 300)}},
+ "metadata": map[string]any{"user_id": string(user)},
+ })
+ name := fmt.Sprintf("00000000-0000-4000-8000-%012d.request.json", f.n)
+ f.write(name, body)
+ return name
+}
+
+func (f *fixture) response(requestID, messageID string) string {
+ body, _ := json.Marshal(map[string]any{"id": messageID, "type": "message", "role": "assistant", "model": "claude-opus-5",
+ "content": []any{map[string]any{"type": "text", "text": "ok"}}})
+ name := requestID + ".response.json"
+ f.write(name, body)
+ return name
+}
+
+func (f *fixture) write(name string, body []byte) {
+ f.t.Helper()
+ path := filepath.Join(f.src, name)
+ if err := os.WriteFile(path, body, 0o644); err != nil {
+ f.t.Fatal(err)
+ }
+ at := base.Add(time.Duration(f.n) * time.Second)
+ if err := os.Chtimes(path, at, at); err != nil {
+ f.t.Fatal(err)
+ }
+ f.n++
+}
+
+func all(string) claudecodeprovider.Verdict { return claudecodeprovider.Collect }
+
+func (f *fixture) pass(filter claudecodeprovider.Filter) *claudecodeprovider.Stats {
+ f.t.Helper()
+ st, err := f.col.CollectAll(filter)
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ for _, e := range st.Errors {
+ f.t.Errorf("pass error: %v", e)
+ }
+ return st
+}
+
+// landed rebuilds every body landed for a session, by file name.
+func (f *fixture) landed(session string) map[string][]byte {
+ f.t.Helper()
+ files, err := storage.LandedFiles(f.zone, session)
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ s := providerbody.NewSession()
+ out := map[string][]byte{}
+ for _, lf := range files {
+ if !strings.HasPrefix(filepath.Base(lf.Path), "provider_body-") {
+ continue
+ }
+ fh, err := os.Open(lf.Path)
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ r, err := sessiondata.NewReader(fh)
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ for {
+ rec, err := r.Next()
+ if errors.Is(err, io.EOF) {
+ break
+ }
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ if err := s.Add(rec); err != nil {
+ f.t.Fatalf("%s: %v", rec.ID, err)
+ }
+ m, _ := providerbody.ManifestOf(rec)
+ b, err := s.Body(rec.ID)
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ out[m.Src] = b
+ }
+ fh.Close()
+ }
+ return out
+}
+
+func TestBodiesLandInTheirSessionsAndRebuildExactly(t *testing.T) {
+ f := newFixture(t)
+ r1 := f.request(sessionA, "", 3)
+ p1 := f.response("req_a1", "msg_a1")
+ r2 := f.request(sessionA, "req_a1", 4)
+ b1 := f.request(sessionB, "", 2)
+ // A response no request names and no transcript holds, as the one that
+ // names a session is.
+ f.response("req_title", "msg_title")
+ st := f.pass(all)
+ if st.SourcesLanded != 4 || st.Waiting != 1 {
+ t.Fatalf("landed %d, waiting %d; want 4 and the unclaimed response waiting", st.SourcesLanded, st.Waiting)
+ }
+ for session, names := range map[string][]string{sessionA: {r1, p1, r2}, sessionB: {b1}} {
+ got := f.landed(session)
+ if len(got) != len(names) {
+ t.Fatalf("%s holds %d bodies, want %d", session, len(got), len(names))
+ }
+ for _, n := range names {
+ raw, _ := os.ReadFile(filepath.Join(f.src, n))
+ if !bytes.Equal(got[n], raw) {
+ t.Fatalf("%s rebuilt to other bytes", n)
+ }
+ }
+ }
+ if !st.Complete() {
+ t.Fatal("a pass that left only an unclaimed response was reported incomplete")
+ }
+ // Nothing new: the next pass reads nothing and lands nothing.
+ if st := f.pass(all); st.SourcesLanded != 0 {
+ t.Fatalf("a second pass landed %d", st.SourcesLanded)
+ }
+}
+
+func TestAResponseNoLaterRequestNamesIsClaimedByItsTranscript(t *testing.T) {
+ f := newFixture(t)
+ f.request(sessionA, "", 2)
+ last := f.response("req_last", "msg_last")
+ if st := f.pass(all); st.Waiting != 1 {
+ t.Fatalf("waiting %d, want 1 before the transcript holds the call", st.Waiting)
+ }
+ // The local adapter lands the transcript and indexes the call.
+ ix := index.New(sessionA)
+ hdr := &sessiondata.Header{Kind: sessiondata.KindTranscript, Session: sessionA, Stream: "main"}
+ e, blocks := index.FromRecord(ix, hdr, &sessiondata.Record{ID: "u1", Call: "msg_last", From: sessiondata.FromAgent}, 1, 1)
+ ix.Append(e, blocks...)
+ if err := ix.Write(f.zone.IndexDir(sessionA)); err != nil {
+ t.Fatal(err)
+ }
+ st := f.pass(all)
+ if st.SourcesLanded != 1 || st.Waiting != 0 {
+ t.Fatalf("landed %d, waiting %d; want the response claimed", st.SourcesLanded, st.Waiting)
+ }
+ if _, ok := f.landed(sessionA)[last]; !ok {
+ t.Fatal("the response did not land in the session whose transcript holds its call")
+ }
+}
+
+func TestAHalfWrittenFileWaitsAndIsCalledUnreadableOnlyLater(t *testing.T) {
+ f := newFixture(t)
+ path := filepath.Join(f.src, "00000000-0000-4000-8000-000000000099.request.json")
+ if err := os.WriteFile(path, []byte(`{"model":"claude-opus-5","messages":[`), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ f.now = time.Now()
+ if st := f.pass(all); st.Unreadable != 0 || st.SourcesLanded != 0 {
+ t.Fatalf("a file written a moment ago was judged: %+v", st)
+ }
+ f.now = time.Now().Add(claudecodeprovider.Settle + time.Minute)
+ if st := f.pass(all); st.Unreadable != 1 {
+ t.Fatalf("unreadable %d after the settle time, want 1", st.Unreadable)
+ }
+}
+
+func TestLostStateLandsNothingTwice(t *testing.T) {
+ f := newFixture(t)
+ f.request(sessionA, "", 2)
+ f.response("req_a1", "msg_a1")
+ f.request(sessionA, "req_a1", 3)
+ f.pass(all)
+ before, _ := storage.LandedFiles(f.zone, sessionA)
+ // A crash after landing and before the table was saved, or a table
+ // deleted by hand: it rebuilds from the landed records.
+ if err := os.Remove(f.zone.ProviderStatePath()); err != nil {
+ t.Fatal(err)
+ }
+ if st := f.pass(all); st.SourcesLanded != 0 || st.Conflicts != 0 {
+ t.Fatalf("after the state was lost a pass landed %d with %d conflicts", st.SourcesLanded, st.Conflicts)
+ }
+ after, _ := storage.LandedFiles(f.zone, sessionA)
+ if len(after) != len(before) {
+ t.Fatalf("landed files went from %d to %d", len(before), len(after))
+ }
+}
+
+func TestARemovedSessionLandsAgain(t *testing.T) {
+ f := newFixture(t)
+ r := f.request(sessionA, "", 2)
+ f.pass(all)
+ if err := os.RemoveAll(f.zone.SessionDir(sessionA)); err != nil {
+ t.Fatal(err)
+ }
+ if st := f.pass(all); st.SourcesLanded != 1 {
+ t.Fatalf("landed %d after the session was removed, want 1", st.SourcesLanded)
+ }
+ if _, ok := f.landed(sessionA)[r]; !ok {
+ t.Fatal("the body is not back in the session")
+ }
+}
+
+func TestTheFilterExcludesAndHolds(t *testing.T) {
+ f := newFixture(t)
+ f.request(sessionA, "", 2)
+ f.request(sessionB, "", 2)
+ filter := func(s string) claudecodeprovider.Verdict {
+ if s == sessionA {
+ return claudecodeprovider.Exclude
+ }
+ return claudecodeprovider.Wait
+ }
+ st := f.pass(filter)
+ if st.SourcesLanded != 0 || st.Waiting != 1 {
+ t.Fatalf("landed %d, waiting %d; want nothing landed and the unjudged session waiting", st.SourcesLanded, st.Waiting)
+ }
+ if st := f.pass(all); st.SourcesLanded != 2 {
+ t.Fatalf("landed %d once both sessions are collected, want 2", st.SourcesLanded)
+ }
+}
+
+func TestAFileChangedAfterLandingIsAConflict(t *testing.T) {
+ f := newFixture(t)
+ r := f.request(sessionA, "", 2)
+ f.pass(all)
+ path := filepath.Join(f.src, r)
+ raw, _ := os.ReadFile(path)
+ if err := os.WriteFile(path, bytes.Replace(raw, []byte("start"), []byte("other"), 1), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ for pass := 1; pass <= 2; pass++ {
+ // The conflict fails every pass it stays, not only the one that found it.
+ st, err := f.col.CollectAll(all)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if st.Conflicts != 1 || st.SourcesLanded != 0 || st.Complete() {
+ t.Fatalf("pass %d over a changed body: %+v", pass, st)
+ }
+ }
+ // Putting the bytes back ends it.
+ if err := os.WriteFile(path, raw, 0o644); err != nil {
+ t.Fatal(err)
+ }
+ if st := f.pass(all); st.Conflicts != 0 || !st.Complete() {
+ t.Fatalf("the body put back still conflicts: %+v", st)
+ }
+}
+
+func TestASessionIDThatIsNoSessionIDLandsNothing(t *testing.T) {
+ f := newFixture(t)
+ f.request("../outside", "", 2)
+ if err := os.MkdirAll(filepath.Join(filepath.Dir(f.zone.Root()), "outside"), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ st := f.pass(all)
+ if st.SourcesLanded != 0 || st.Waiting != 1 {
+ t.Fatalf("a request naming ../outside: %+v", st)
+ }
+ if _, err := os.Stat(filepath.Join(filepath.Dir(f.zone.Root()), "outside", "provider_body")); !os.IsNotExist(err) {
+ t.Fatalf("something was written outside the root: %v", err)
+ }
+}
+
+// A session whose earlier provider file is gone, lost or kept in another
+// storage stage, still takes new bodies. They are cut only against what is
+// left, so each rebuilds from the files that remain.
+func TestASessionMissingAnEarlierFileStillTakesNewBodies(t *testing.T) {
+ f := newFixture(t)
+ f.request(sessionA, "", 3)
+ f.pass(all)
+ f.response("req_a1", "msg_a1")
+ f.request(sessionA, "req_a1", 4)
+ f.pass(all)
+ files, _ := storage.LandedFiles(f.zone, sessionA)
+ var provider []string
+ for _, lf := range files {
+ if strings.HasPrefix(filepath.Base(lf.Path), "provider_body-") {
+ provider = append(provider, lf.Path)
+ }
+ }
+ if len(provider) != 2 {
+ t.Fatalf("%d provider files, want one per pass", len(provider))
+ }
+ if err := os.Chmod(filepath.Dir(provider[0]), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Remove(provider[0]); err != nil {
+ t.Fatal(err)
+ }
+ next := f.request(sessionA, "req_a2", 5)
+ f.response("req_a2", "msg_a2")
+ // The new request, and the response it names as its previous one.
+ if st := f.pass(all); st.SourcesLanded != 2 {
+ t.Fatalf("landed %d after an earlier file was lost, want the new request and its named response", st.SourcesLanded)
+ }
+ // Rebuilt from the files that remain, the new body is whole. The body
+ // of the second file refers to the lost one and does not rebuild.
+ s := providerbody.NewSession()
+ var rebuilt []byte
+ files, _ = storage.LandedFiles(f.zone, sessionA)
+ for _, lf := range files {
+ if !strings.HasPrefix(filepath.Base(lf.Path), "provider_body-") {
+ continue
+ }
+ fh, err := os.Open(lf.Path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ _, recs, err := sessiondata.All(fh)
+ fh.Close()
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, rec := range recs {
+ if s.Add(rec) != nil {
+ continue
+ }
+ if m, _ := providerbody.ManifestOf(rec); m.Src == next {
+ if rebuilt, err = s.Body(rec.ID); err != nil {
+ t.Fatal(err)
+ }
+ }
+ }
+ }
+ raw, _ := os.ReadFile(filepath.Join(f.src, next))
+ if !bytes.Equal(rebuilt, raw) {
+ t.Fatal("the new body does not rebuild from the files that remain")
+ }
+}
+
+// A pass that stopped after landing and before saving the index, and whose
+// table is gone too, still leaves the index covering what is landed.
+func TestARebuiltTableClosesTheIndexGap(t *testing.T) {
+ f := newFixture(t)
+ f.request(sessionA, "", 2)
+ f.pass(all)
+ files, _ := storage.LandedFiles(f.zone, sessionA)
+ landedTo := files[len(files)-1].Seq
+ if err := os.RemoveAll(f.zone.IndexDir(sessionA)); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Remove(f.zone.ProviderStatePath()); err != nil {
+ t.Fatal(err)
+ }
+ if st := f.pass(all); st.SourcesLanded != 0 {
+ t.Fatalf("landed %d after the table was rebuilt", st.SourcesLanded)
+ }
+ ix, err := storage.LoadIndexState(f.zone.IndexStatePath(sessionA), sessionA)
+ if err != nil || ix.IndexedSeq != landedTo {
+ t.Fatalf("the index covers %d, and %d is landed: %v", ix.IndexedSeq, landedTo, err)
+ }
+}
+
+// A file ends before a body that would take it past the budget.
+func TestAFileEndsBeforeTheBudget(t *testing.T) {
+ f := newFixture(t)
+ f.col.MaxDelta = 12 << 10
+ f.request(sessionA, "", 2)
+ f.request(sessionB, "", 2)
+ f.request(sessionA, "", 3)
+ f.request(sessionA, "", 4)
+ f.pass(all)
+ files, _ := storage.LandedFiles(f.zone, sessionA)
+ for _, lf := range files {
+ info, err := os.Stat(lf.Path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ fh, _ := os.Open(lf.Path)
+ _, recs, _ := sessiondata.All(fh)
+ fh.Close()
+ if len(recs) > 1 && info.Size() > f.col.MaxDelta {
+ t.Fatalf("%s holds %d bodies in %d bytes, past a %d byte budget", filepath.Base(lf.Path), len(recs), info.Size(), f.col.MaxDelta)
+ }
+ }
+}
+
+// A session that was busy when its index gap was to be closed has it closed
+// by a later pass, with no new body of it arriving.
+func TestABusySessionHasItsIndexGapClosedLater(t *testing.T) {
+ f := newFixture(t)
+ f.request(sessionA, "", 2)
+ f.pass(all)
+ files, _ := storage.LandedFiles(f.zone, sessionA)
+ landedTo := files[len(files)-1].Seq
+ if err := os.RemoveAll(f.zone.IndexDir(sessionA)); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Remove(f.zone.ProviderStatePath()); err != nil {
+ t.Fatal(err)
+ }
+ lock, err := storage.LockSession(f.zone.SessionDir(sessionA))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if st := f.pass(all); st.Busy == 0 {
+ t.Fatal("the pass did not meet the held lock")
+ }
+ _ = lock.Unlock()
+ f.pass(all)
+ ix, err := storage.LoadIndexState(f.zone.IndexStatePath(sessionA), sessionA)
+ if err != nil || ix.IndexedSeq != landedTo {
+ t.Fatalf("the index covers %d, and %d is landed: %v", ix.IndexedSeq, landedTo, err)
+ }
+}
+
+// One damaged provider file does not stop another session's bodies.
+func TestADamagedFileDoesNotStopOtherSessions(t *testing.T) {
+ f := newFixture(t)
+ f.request(sessionA, "", 2)
+ f.pass(all)
+ files, _ := storage.LandedFiles(f.zone, sessionA)
+ damaged := files[len(files)-1].Path
+ if err := os.Chmod(damaged, 0o644); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Truncate(damaged, 100); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Remove(f.zone.ProviderStatePath()); err != nil {
+ t.Fatal(err)
+ }
+ b := f.request(sessionB, "", 2)
+ st, err := f.col.CollectAll(all)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(st.Errors) == 0 {
+ t.Fatal("the damaged file was not reported")
+ }
+ if _, ok := f.landed(sessionB)[b]; !ok {
+ t.Fatal("a damaged file in one session kept another session's body from landing")
+ }
+}
diff --git a/internal/adapters/claudecodeprovider/covered.go b/internal/adapters/claudecodeprovider/covered.go
new file mode 100644
index 0000000..6e545ac
--- /dev/null
+++ b/internal/adapters/claudecodeprovider/covered.go
@@ -0,0 +1,84 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package claudecodeprovider
+
+import (
+ "errors"
+ "os"
+ "path/filepath"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/storage"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/providerbody"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+)
+
+// SessionFiles lists the body files the adapter has attributed to a session,
+// landed or not, that are still in the source directory. Nothing is written.
+func (c *Collector) SessionFiles(session string) (map[string]bool, error) {
+ sn, err := loadSeen(c.Zone)
+ if err != nil {
+ return nil, err
+ }
+ out := map[string]bool{}
+ for name, e := range sn.entries {
+ if e.session != session {
+ continue
+ }
+ if _, err := os.Lstat(filepath.Join(c.SourceRoot, name)); err == nil {
+ out[name] = true
+ }
+ }
+ return out, nil
+}
+
+// Covered reports whether the named body files of a session are landed in
+// it, each with the bytes the file holds now. It reads the session's landed
+// records, not the table, so a landed file lost since is not taken for
+// landed. A removal deletes a body only then. Nothing is written.
+func (c *Collector) Covered(session string, names []string) (bool, string, error) {
+ held := map[string]string{}
+ files, err := storage.LandedFiles(c.Zone, session)
+ if err != nil {
+ return false, "", err
+ }
+ for _, lf := range files {
+ if lf.Stream != "" || lf.RunID != "" {
+ continue
+ }
+ if err := readLanded(lf.Path, func(_ sessiondata.Header, rec *sessiondata.Record) {
+ if m, err := providerbody.ManifestOf(rec); err == nil {
+ held[m.Src] = m.SHA256
+ }
+ }); err != nil {
+ return false, "", err
+ }
+ }
+ for _, name := range names {
+ body, err := os.ReadFile(filepath.Join(c.SourceRoot, name))
+ if errors.Is(err, os.ErrNotExist) {
+ return false, name + " is missing", nil
+ }
+ if err != nil {
+ return false, "", err
+ }
+ if held[name] != providerbody.Digest(body) {
+ return false, "the provider body " + name + " is not landed with the bytes it holds", nil
+ }
+ }
+ return true, "", nil
+}
diff --git a/internal/adapters/claudecodeprovider/glossary_test.go b/internal/adapters/claudecodeprovider/glossary_test.go
new file mode 100644
index 0000000..4b14e13
--- /dev/null
+++ b/internal/adapters/claudecodeprovider/glossary_test.go
@@ -0,0 +1,36 @@
+// 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 claudecodeprovider_test
+
+import (
+ "testing"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecodeprovider"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/model"
+)
+
+// Every name the model can emit has an entry, even where a body has no word
+// for it, so a reader asking for native words is never shown a gap.
+func TestGlossaryCoversTheWholeVocabulary(t *testing.T) {
+ g := claudecodeprovider.Glossary()
+ for _, name := range model.Vocabulary() {
+ if _, ok := g.Lookup(name); !ok {
+ t.Errorf("no glossary entry for %q", name)
+ }
+ }
+}
diff --git a/internal/adapters/claudecodeprovider/seen.go b/internal/adapters/claudecodeprovider/seen.go
new file mode 100644
index 0000000..a65fbd5
--- /dev/null
+++ b/internal/adapters/claudecodeprovider/seen.go
@@ -0,0 +1,276 @@
+// 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 claudecodeprovider
+
+import (
+ "bufio"
+ "errors"
+ "fmt"
+ "io"
+ "os"
+ "sort"
+ "strconv"
+ "strings"
+ "time"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/storage"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/providerbody"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+)
+
+// States of a body file.
+const (
+ // stateLanded: the body is landed in its session.
+ stateLanded = "landed"
+ // stateWaiting: no session claims the body yet, or its session is not
+ // one discovery has found. Every pass tries again.
+ stateWaiting = "waiting"
+ // stateExcluded: the body's session is left out by include and exclude.
+ stateExcluded = "excluded"
+ // stateUnreadable: the file is not one JSON value, long after it was
+ // last written, so it names no session.
+ stateUnreadable = "unreadable"
+ // stateConflict: a body of the same name with other bytes is already
+ // landed in the session. Nothing lands from it.
+ stateConflict = "conflict"
+ // stateChanged: the file changed after it landed. The line keeps the
+ // digest that landed, so putting the bytes back ends it.
+ stateChanged = "changed"
+)
+
+// seenSchema is the version of the state file.
+const seenSchema = "1"
+
+// entry is what the root remembers about one body file.
+type entry struct {
+ name string
+ size int64
+ mtime int64
+ sha string // the first twelve hexadecimal characters of the body's SHA-256
+ state string
+ session string
+ keys providerbody.Keys
+}
+
+// seen is the root's table of body files by name.
+//
+// It is derived. Every landed body is a record whose manifest names its file,
+// its size and its digest, so the table rebuilds from the landed records when
+// it is missing, and a landed entry is trusted only while its session's
+// provider directory exists.
+type seen struct {
+ entries map[string]*entry
+ dirty bool
+ // damaged lists the landed provider files a rebuild could not read.
+ // Their bodies are not in the table, and every other session's are.
+ damaged []error
+}
+
+func loadSeen(z *storage.Zone) (*seen, error) {
+ f, err := os.Open(z.ProviderStatePath())
+ if errors.Is(err, os.ErrNotExist) {
+ return rebuildSeen(z)
+ }
+ if err != nil {
+ return nil, err
+ }
+ defer f.Close()
+ s := &seen{entries: map[string]*entry{}}
+ br := bufio.NewReader(f)
+ first := true
+ for {
+ line, err := br.ReadString('\n')
+ if line = strings.TrimRight(line, "\n"); line != "" {
+ if first {
+ if line != "schema\t"+seenSchema {
+ // Another version's table is discarded, not migrated.
+ return rebuildSeen(z)
+ }
+ first = false
+ } else if e, ok := parseEntry(line); ok {
+ s.entries[e.name] = e
+ }
+ }
+ if errors.Is(err, io.EOF) {
+ break
+ }
+ if err != nil {
+ return nil, err
+ }
+ }
+ return s, nil
+}
+
+// dash writes a value into one tab-separated field. A value that could break
+// the line, which only a body can supply, is written as absent.
+func dash(v string) string {
+ if v == "" || strings.ContainsAny(v, " \t\r\n") {
+ return "-"
+ }
+ return v
+}
+
+func undash(v string) string {
+ if v == "-" {
+ return ""
+ }
+ return v
+}
+
+func parseEntry(line string) (*entry, bool) {
+ f := strings.Split(line, "\t")
+ if len(f) != 10 {
+ return nil, false
+ }
+ size, err1 := strconv.ParseInt(f[1], 10, 64)
+ mtime, err2 := strconv.ParseInt(f[2], 10, 64)
+ if err1 != nil || err2 != nil {
+ return nil, false
+ }
+ return &entry{
+ name: f[0], size: size, mtime: mtime, sha: undash(f[3]), state: f[4], session: undash(f[5]),
+ keys: providerbody.Keys{Request: undash(f[6]), Call: undash(f[7]), PreviousRequest: undash(f[8]), Session: undash(f[9])},
+ }, true
+}
+
+func (s *seen) save(z *storage.Zone, now time.Time) error {
+ names := make([]string, 0, len(s.entries))
+ for n := range s.entries {
+ names = append(names, n)
+ }
+ sort.Strings(names)
+ err := storage.WriteAtomic(z.ProviderStatePath(), storage.PermState, func(w io.Writer) error {
+ bw := bufio.NewWriter(w)
+ fmt.Fprintf(bw, "schema\t%s\n", seenSchema)
+ for _, n := range names {
+ e := s.entries[n]
+ fmt.Fprintf(bw, "%s\t%d\t%d\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n", e.name, e.size, e.mtime, dash(e.sha), e.state,
+ dash(e.session), dash(e.keys.Request), dash(e.keys.Call), dash(e.keys.PreviousRequest), dash(e.keys.Session))
+ }
+ return bw.Flush()
+ })
+ if err == nil {
+ s.dirty = false
+ }
+ _ = now
+ return err
+}
+
+// rebuildSeen reads every landed provider body of every session. A rebuilt
+// entry has no modification time, so the first pass reads each file once
+// more and compares its digest.
+func rebuildSeen(z *storage.Zone) (*seen, error) {
+ s := &seen{entries: map[string]*entry{}, dirty: true}
+ items, err := os.ReadDir(z.Root())
+ if err != nil {
+ if errors.Is(err, os.ErrNotExist) {
+ return s, nil
+ }
+ return nil, err
+ }
+ for _, it := range items {
+ if !it.IsDir() || strings.HasPrefix(it.Name(), "_") {
+ continue
+ }
+ files, err := storage.LandedFiles(z, it.Name())
+ if err != nil {
+ s.damaged = append(s.damaged, fmt.Errorf("claudecodeprovider: session %s: %w", it.Name(), err))
+ continue
+ }
+ for _, lf := range files {
+ if lf.Stream != "" || lf.RunID != "" {
+ continue
+ }
+ if err := readLanded(lf.Path, func(hdr sessiondata.Header, rec *sessiondata.Record) {
+ m, err := providerbody.ManifestOf(rec)
+ if err != nil || len(m.SHA256) < 12 {
+ return
+ }
+ s.entries[m.Src] = &entry{
+ name: m.Src, size: int64(m.Bytes), mtime: -1, sha: m.SHA256[:12], state: stateLanded, session: hdr.Session,
+ keys: providerbody.Keys{Request: m.Request, Call: m.Call, PreviousRequest: m.PreviousRequest, Session: m.Session},
+ }
+ }); err != nil {
+ // One damaged file must not keep every other session's bodies
+ // from being collected. A file cut short or edited holds bodies
+ // the table then does not know, and they are read again.
+ s.damaged = append(s.damaged, err)
+ }
+ }
+ }
+ return s, nil
+}
+
+// readLanded calls fn for every record of a landed provider_body file.
+func readLanded(path string, fn func(sessiondata.Header, *sessiondata.Record)) error {
+ f, err := os.Open(path)
+ if err != nil {
+ return err
+ }
+ defer f.Close()
+ r, err := sessiondata.NewReader(f)
+ if err != nil {
+ return fmt.Errorf("claudecodeprovider: %s: %w", path, err)
+ }
+ if r.Header().Kind != sessiondata.KindProviderBody {
+ return nil
+ }
+ for {
+ rec, err := r.Next()
+ if errors.Is(err, io.EOF) {
+ return nil
+ }
+ if err != nil {
+ return fmt.Errorf("claudecodeprovider: %s: %w", path, err)
+ }
+ fn(r.Header(), rec)
+ }
+}
+
+// Forget drops every line of the given sessions from the root's table. A
+// removal calls it once the sessions' body files and directories are gone, as
+// it drops their lines from the export and metrics state. A table that does
+// not exist has nothing to drop.
+func Forget(z *storage.Zone, sessions []string, now time.Time) error {
+ if _, err := os.Stat(z.ProviderStatePath()); errors.Is(err, os.ErrNotExist) {
+ return nil
+ }
+ lock, err := storage.LockProvider(z.Root())
+ if err != nil {
+ return err
+ }
+ defer func() { _ = lock.Unlock() }()
+ sn, err := loadSeen(z)
+ if err != nil {
+ return err
+ }
+ gone := map[string]bool{}
+ for _, s := range sessions {
+ gone[s] = true
+ }
+ for name, e := range sn.entries {
+ if gone[e.session] || gone[e.keys.Session] {
+ delete(sn.entries, name)
+ sn.dirty = true
+ }
+ }
+ if !sn.dirty {
+ return nil
+ }
+ return sn.save(z, now)
+}
diff --git a/internal/assemble/streams.go b/internal/assemble/streams.go
index b57c080..2526d95 100644
--- a/internal/assemble/streams.go
+++ b/internal/assemble/streams.go
@@ -48,14 +48,16 @@ func (b *builder) stage1Canonical() {
// stream, not a step of it: it names the tool it belongs to and a view
// joins the two. Letting it into a stream would change that stream's
// record count and time range, so a session with such records would
- // fold differently from the same session without them.
+ // fold differently from the same session without them. A provider body
+ // is set aside for the same reason: it is what a call sent and got back,
+ // joined to the call by a view, and no step of any stream.
bounded := make([]int32, 0, len(all))
for _, i := range all {
e := &b.ix.Entries[i]
if b.opt.ThroughSeq > 0 && uint64(e.Seq) > b.opt.ThroughSeq {
continue
}
- if e.Kind == index.KindChanges {
+ if e.Kind == index.KindChanges || e.Kind == index.KindProviderBody {
continue
}
bounded = append(bounded, i)
diff --git a/internal/config/config.go b/internal/config/config.go
index e8c3768..7dedc6d 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -235,6 +235,12 @@ const (
// posture, like the local adapter, and the plugin needs nothing from
// it: it discovers and tails what the plugin has already written.
AdapterClaudeCodeChanges = "claude-code-changes"
+ // AdapterClaudeCodeProvider reads the request and response bodies
+ // Claude Code writes for its model provider when
+ // OTEL_LOG_RAW_API_BODIES names a directory. Pull posture, like the
+ // local adapter: Claude Code writes the files, and this adapter finds
+ // each body's session and lands it.
+ AdapterClaudeCodeProvider = "claude-code-provider"
)
// Default returns the configuration used when none is supplied.
@@ -273,6 +279,18 @@ func Default() *Config {
Interval: DefaultInterval,
MaxDeltaBytes: 2 << 20,
},
+ }, {
+ // The bodies Claude Code writes when it is told to. On by default
+ // because it costs nothing until OTEL_LOG_RAW_API_BODIES names
+ // the directory: there is nothing to list.
+ Name: AdapterClaudeCodeProvider,
+ Enabled: true,
+ Exclude: []string{"/private/tmp/**"},
+ Collector: Collector{
+ Mode: ModeWatch,
+ Interval: DefaultInterval,
+ MaxDeltaBytes: 2 << 20,
+ },
}},
Parse: Parse{MaxRoundBytes: 2 << 20},
Export: Export{OTLP: OTLP{
@@ -400,6 +418,9 @@ func (c *Config) Validate() error {
if a.Name == AdapterClaudeCodeChanges && (a.Metrics || a.MetricsLookback != "" || a.Listen != "") {
return fmt.Errorf("config: adapter %q reads change records only: it takes source_root, include, exclude and collector, not metrics, metrics_lookback or listen", a.Name)
}
+ if a.Name == AdapterClaudeCodeProvider && (a.Metrics || a.MetricsLookback != "" || a.Listen != "") {
+ return fmt.Errorf("config: adapter %q reads provider bodies only: it takes source_root, include, exclude and collector, not metrics, metrics_lookback or listen", a.Name)
+ }
if a.Collector.Mode != ModeWatch && a.Collector.Mode != ModeOnce {
return fmt.Errorf("config: adapter %q: unknown collector mode %q", a.Name, a.Collector.Mode)
}
diff --git a/internal/config/config_test.go b/internal/config/config_test.go
index 0677fd8..fcf7773 100644
--- a/internal/config/config_test.go
+++ b/internal/config/config_test.go
@@ -58,8 +58,8 @@ func TestRepoConfigSpellsOutEveryValue(t *testing.T) {
if err != nil {
t.Fatal(err)
}
- if len(got.Adapters) != 3 {
- t.Fatalf("adapters: got %d, want the local adapter, the receiver and the changes adapter", len(got.Adapters))
+ if len(got.Adapters) != 4 {
+ t.Fatalf("adapters: got %d, want the local adapter, the receiver, the changes adapter and the provider adapter", len(got.Adapters))
}
for _, a := range got.Adapters {
if a.Name == AdapterClaudeCodeOTLP {
diff --git a/internal/index/fromrecord.go b/internal/index/fromrecord.go
index de3dc46..427a4ba 100644
--- a/internal/index/fromrecord.go
+++ b/internal/index/fromrecord.go
@@ -62,6 +62,15 @@ func FromRecord(ix *Index, hdr *sessiondata.Header, rec *sessiondata.Record,
e.TS = t.UnixNano()
}
}
+ if e.Kind == KindProviderBody {
+ // Its parts are pieces of a body, never change records, and a block
+ // per part is all a round needs to count them.
+ blocks := make([]Block, len(rec.Parts))
+ for i := range blocks {
+ blocks[i] = Block{Ord: uint16(i), Kind: BlockOther}
+ }
+ return e, blocks
+ }
return e, blocksOf(in, rec)
}
@@ -95,6 +104,8 @@ func kindOf(hdr *sessiondata.Header, rec *sessiondata.Record) Kind {
return KindScript
case sessiondata.KindChanges:
return KindChanges
+ case sessiondata.KindProviderBody:
+ return KindProviderBody
}
switch rec.From {
case sessiondata.FromAgent:
diff --git a/internal/index/index.go b/internal/index/index.go
index 47bf750..b14be0e 100644
--- a/internal/index/index.go
+++ b/internal/index/index.go
@@ -33,7 +33,7 @@ package index
// Schema is the on-disk index version. Bump it when Entry or Block changes;
// a mismatch discards the index and rebuilds rather than migrating.
-const Schema = 11
+const Schema = 12
// Kind classifies a record without reading it.
type Kind uint8
@@ -50,6 +50,9 @@ const (
KindScript // workflow script
KindOther // an out-of-band record type carrying no uuid
KindChanges // a workspace change record; evidence, never a step
+ // KindProviderBody is a body the runtime exchanged with its model
+ // provider; evidence, never a step, and in no derived lookup.
+ KindProviderBody
)
// Trigger says what caused a prompt cycle.
diff --git a/internal/index/lookup.go b/internal/index/lookup.go
index c9e23f1..a1851c4 100644
--- a/internal/index/lookup.go
+++ b/internal/index/lookup.go
@@ -82,8 +82,11 @@ func (ix *Index) Build() {
dup := make([]bool, len(ix.Entries))
for i := range ix.Entries {
e := &ix.Entries[i]
- if e.Record == 0 {
- continue // no identity to compare on, so every such entry is its own record
+ if e.Record == 0 || e.Kind == KindProviderBody {
+ // No identity to compare on, so every such entry is its own
+ // record. A provider body's id names a body file, never a
+ // record of a stream, so it must not make one a duplicate.
+ continue
}
if _, seen := ix.byRecord[e.Record]; seen {
dup[i] = true
@@ -91,8 +94,12 @@ func (ix *Index) Build() {
}
ix.byRecord[e.Record] = int32(i)
}
+ // A provider body names the call its response came from, but it is no
+ // fragment of that call. In byMsg it would change the call's fragments,
+ // so it stays out of every derived lookup, as a duplicate does.
+ skip := func(i int) bool { return dup[i] || ix.Entries[i].Kind == KindProviderBody }
for i := range ix.Entries {
- if dup[i] {
+ if skip(i) {
continue
}
e := &ix.Entries[i]
@@ -110,7 +117,7 @@ func (ix *Index) Build() {
ix.byToolRef = make(map[uint32][]int32)
ix.byChild = make(map[uint32][]int32)
for i := range ix.Entries {
- if dup[i] {
+ if skip(i) {
continue
}
e := &ix.Entries[i]
@@ -128,7 +135,7 @@ func (ix *Index) Build() {
ix.byToolBlock = make(map[uint32][]int32, len(ix.Blocks))
for i := range ix.Blocks {
b := &ix.Blocks[i]
- if b.ToolID == 0 || dup[b.Entry] {
+ if b.ToolID == 0 || skip(int(b.Entry)) {
continue
}
ix.byToolBlock[b.ToolID] = append(ix.byToolBlock[b.ToolID], int32(i))
diff --git a/internal/scenario/build.go b/internal/scenario/build.go
index db19142..7e51339 100644
--- a/internal/scenario/build.go
+++ b/internal/scenario/build.go
@@ -106,14 +106,14 @@ func Build(sc *Scenario, format Format, out string, opts Options) (*Built, error
if now.IsZero() {
now = time.Now()
}
- res.Files, err = writeSD(p, out, now.UTC())
+ res.Files, err = writeSD(p, out, now.UTC(), opts.MaxDelta)
default:
return nil, fmt.Errorf("scenario: unknown format %q; use %s or %s", format, FormatClaudeCode, FormatSD)
}
if err != nil {
return nil, err
}
- res.Config, err = writeConfig(out, source, opts.Watch)
+ res.Config, err = writeConfig(out, source, opts.Watch, opts.MaxDelta)
return res, err
}
@@ -127,13 +127,13 @@ func Build(sc *Scenario, format Format, out string, opts Options) (*Built, error
// so its configuration says watch, and asz collect or asz server beside it
// picks each one up within feedInterval. The product default is minutes,
// which would leave a feed looking stopped.
-func writeConfig(out, source string, watch bool) (string, error) {
+func writeConfig(out, source string, watch bool, maxDelta int64) (string, error) {
path := filepath.Join(out, "asz.yaml")
mode, other := "once", "watch"
if watch {
mode, other = "watch", "once"
}
- text := configText(out, source, mode, feedInterval)
+ text := configText(out, source, mode, feedInterval, true, maxDelta)
old, err := os.ReadFile(path)
if err != nil {
return path, os.WriteFile(path, []byte(text), 0o644)
@@ -145,15 +145,18 @@ func writeConfig(out, source string, watch bool) (string, error) {
}
// The same build in the other collector mode, which is what adding or
// dropping --every does to a directory that already holds sessions, or
- // a file an earlier build wrote without an interval. Rewrite the block
- // and keep what was appended to it.
- for _, prior := range []string{
- configText(out, source, other, feedInterval),
- configText(out, source, mode, ""),
- configText(out, source, other, ""),
- } {
- if strings.HasPrefix(string(old), prior) {
- return path, os.WriteFile(path, append([]byte(text), old[len(prior):]...), 0o644)
+ // a file an earlier build wrote without an interval or without the
+ // provider adapter. Rewrite the block and keep what was appended to it.
+ for _, m := range []string{mode, other} {
+ for _, interval := range []string{feedInterval, ""} {
+ for _, provider := range []bool{true, false} {
+ for _, delta := range []int64{maxDelta, 0} {
+ prior := configText(out, source, m, interval, provider, delta)
+ if prior != text && strings.HasPrefix(string(old), prior) {
+ return path, os.WriteFile(path, append([]byte(text), old[len(prior):]...), 0o644)
+ }
+ }
+ }
}
}
return "", errors.New("scenario: " + path + " exists and is not this build's; use another --out")
@@ -164,12 +167,20 @@ func writeConfig(out, source string, watch bool) (string, error) {
const feedInterval = "5s"
// configText is the configuration for one collector mode. An empty interval
-// is the form builds wrote before they set one.
-func configText(out, source, mode, interval string) string {
+// is the form builds wrote before they set one, and provider false the form
+// before they wrote provider bodies. A zero maxDelta leaves the default.
+func configText(out, source, mode, interval string, provider bool, maxDelta int64) string {
collector := " collector:\n mode: " + mode + "\n"
if interval != "" {
collector += " interval: " + interval + "\n"
}
+ if maxDelta > 0 {
+ collector += fmt.Sprintf(" max_delta_bytes: %d\n", maxDelta)
+ }
+ providerBlock := ""
+ if provider {
+ providerBlock = " - name: claude-code-provider\n enabled: true\n source_root: " + source + "/" + ProviderBodyDir + "\n" + collector
+ }
return fmt.Sprintf(`# Written by asz scenario build. The storage root is this directory; the
# adapter's source is the source directory beside it, which an sd build
# leaves empty.
@@ -186,12 +197,12 @@ adapters:
%s - name: claude-code-changes
enabled: true
source_root: %s/plugins/data
-%s# To export the session with asz push, name the receiver:
+%s%s# To export the session with asz push, name the receiver:
# export:
# otlp:
# protocol: grpc
# endpoint: 127.0.0.1:11800
-`, out, source, collector, source, collector)
+`, out, source, collector, source, collector, providerBlock)
}
func mkdirAll(dir string) error { return os.MkdirAll(dir, 0o755) }
diff --git a/internal/scenario/expect/expect.go b/internal/scenario/expect/expect.go
index 88c9b9a..bc98744 100644
--- a/internal/scenario/expect/expect.go
+++ b/internal/scenario/expect/expect.go
@@ -23,9 +23,11 @@ package expect
import (
"encoding/json"
+ "errors"
"fmt"
"os"
"path/filepath"
+ "regexp"
"sort"
"strings"
"time"
@@ -39,6 +41,7 @@ import (
"github.com/apache/skywalking-ai-sessionizer/internal/verify"
"github.com/apache/skywalking-ai-sessionizer/internal/view"
"github.com/apache/skywalking-ai-sessionizer/pkg/model"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/providerbody"
"github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
"github.com/apache/skywalking-ai-sessionizer/pkg/sessionflow"
"github.com/apache/skywalking-ai-sessionizer/pkg/sessionview"
@@ -50,6 +53,7 @@ type File struct {
Checkpoints map[string]*Checkpoint `yaml:"checkpoints"`
Properties Properties `yaml:"properties"`
Parse ParseOptions `yaml:"parse"`
+ Collect CollectOptions `yaml:"collect"`
// Push is what the OTLP push of the finished session must carry.
Push *Push `yaml:"push"`
}
@@ -60,6 +64,12 @@ type Push struct {
Kinds []string `yaml:"kinds"`
}
+// CollectOptions are collector settings a scenario runs under. The build
+// writes them into its configuration, and the collection follows it.
+type CollectOptions struct {
+ MaxDeltaBytes int64 `yaml:"max_delta_bytes"`
+}
+
// ParseOptions are parse settings a scenario runs under.
type ParseOptions struct {
MaxRoundBytes int64 `yaml:"max_round_bytes"`
@@ -87,7 +97,16 @@ type Properties struct {
// to the same nodes and relations: a change record is evidence beside a
// step, never a step. Checked when a scenario has changes.
ChangesLeaveTheFold *bool `yaml:"changes_leave_the_fold"`
- MetricsMatchThePlan *bool `yaml:"metrics_match_the_plan"`
+ // ProviderBodiesLeaveTheFold says the same scenario without its provider
+ // bodies folds to the same nodes and relations. Checked when a scenario
+ // writes provider bodies.
+ ProviderBodiesLeaveTheFold *bool `yaml:"provider_bodies_leave_the_fold"`
+ // ProviderBodiesRebuild says every body a call lists rebuilds from the
+ // session's provider files up to the one it is in, as a reader loading
+ // bodies on demand reads them. Checked when a scenario writes provider
+ // bodies, and again on the root repack_keeps_structure re-cuts.
+ ProviderBodiesRebuild *bool `yaml:"provider_bodies_rebuild"`
+ MetricsMatchThePlan *bool `yaml:"metrics_match_the_plan"`
// RemovedAfterSent says what a pipeline's removal does with the session,
// on copies of the finished root. Nothing goes before all of it is sent
// to the one receiver the pipeline sends to. All of it goes once it is,
@@ -164,6 +183,7 @@ type Verify struct {
var landedPrefix = map[string]string{
"transcript": "transcript", "agent_meta": "meta", "journal": "journal",
"workflow_manifest": "manifest", "workflow_script": "script", "changes": "changes",
+ "provider_body": "provider_body",
}
// Resolve finds the landed file a Lose names among a session's files. The
@@ -176,7 +196,7 @@ func (l Lose) Resolve(files []storage.LandedFile, stream string) (*storage.Lande
}
prefix, known := landedPrefix[l.Kind]
if l.Kind != "" && !known {
- return nil, fmt.Errorf("lose: unknown kind %q; the kinds are transcript, agent_meta, journal, workflow_manifest, workflow_script, changes", l.Kind)
+ return nil, fmt.Errorf("lose: unknown kind %q; the kinds are transcript, agent_meta, journal, workflow_manifest, workflow_script, changes, provider_body", l.Kind)
}
seen := 0
for i := range files {
@@ -257,6 +277,12 @@ type View struct {
ChangedFiles *int `yaml:"changed_files"`
CapturedBy map[string]int `yaml:"captured_by"`
ChangesJoined *int `yaml:"changes_joined"`
+ // ProviderBodies counts the landed provider bodies, and CapturedPrompts
+ // the calls that list a request.
+ // ProviderFiles counts the session's provider_body files.
+ ProviderFiles *int `yaml:"provider_files"`
+ ProviderBodies *int `yaml:"provider_bodies"`
+ CapturedPrompts *int `yaml:"captured_prompts"`
}
// Talk is what a talk in the document must say.
@@ -523,6 +549,23 @@ func checkView(root, session string, want *View) ([]string, error) {
if doc.Summary.Changes != len(doc.WorkspaceChanges) {
bad("view.summary.changes is %d, the document lists %d", doc.Summary.Changes, len(doc.WorkspaceChanges))
}
+ if want.ProviderBodies != nil && doc.Summary.ProviderBodies != *want.ProviderBodies {
+ bad("view.provider_bodies is %d, want %d", doc.Summary.ProviderBodies, *want.ProviderBodies)
+ }
+ if want.ProviderFiles != nil {
+ n := 0
+ for _, f := range doc.Files {
+ if f.Kind == string(sessiondata.KindProviderBody) {
+ n++
+ }
+ }
+ if n != *want.ProviderFiles {
+ bad("view.provider_files is %d, want %d", n, *want.ProviderFiles)
+ }
+ }
+ if want.CapturedPrompts != nil && doc.Summary.CapturedPrompts != *want.CapturedPrompts {
+ bad("view.captured_prompts is %d, want %d", doc.Summary.CapturedPrompts, *want.CapturedPrompts)
+ }
if want.ChangedFiles != nil || want.CapturedBy != nil || want.ChangesJoined != nil {
files, joined := 0, 0
bySource := map[string]int{}
@@ -948,7 +991,8 @@ func RecordsWellFormed(root, session string) ([]string, error) {
out = append(out, fmt.Sprintf("records_well_formed: %s: header field %q is empty", filepath.Base(lf.Path), name))
}
}
- if hdr.Session != session || (hdr.Stream == "" && hdr.Batch == "") {
+ // A provider body belongs to the session and no stream or run in it.
+ if hdr.Session != session || (hdr.Stream == "" && hdr.Batch == "" && hdr.Kind != sessiondata.KindProviderBody) {
out = append(out, fmt.Sprintf("records_well_formed: %s: header identity %s stream=%q batch=%q", filepath.Base(lf.Path), hdr.Session, hdr.Stream, hdr.Batch))
}
for i, line := range lines[1 : len(lines)-1] {
@@ -1117,6 +1161,50 @@ func ViewCoversTheSession(root, session string) ([]string, error) {
if doc.Summary.Changes != len(doc.WorkspaceChanges) {
bad("summary.changes is %d, the document lists %d", doc.Summary.Changes, len(doc.WorkspaceChanges))
}
+ // Only a call lists provider bodies, its request before its response,
+ // each naming a landed provider_body record of that role, and the
+ // summary counts what the calls list.
+ bodies := providerRecords(root, session)
+ prompts := 0
+ var checkBodies func(n *sessionview.Node)
+ checkBodies = func(n *sessionview.Node) {
+ if len(n.ProviderBodies) > 0 && n.Kind != model.KindLLMCall {
+ bad("%s (%s) lists provider bodies and is no call", n.ID, n.Kind)
+ }
+ // A synthetic call was never sent to a provider: it has no body.
+ for _, c := range n.Children {
+ if c.Kind == model.KindMessageSynthetic && len(n.ProviderBodies) > 0 {
+ bad("%s is a synthetic call and lists provider bodies", n.ID)
+ }
+ }
+ for i, pb := range n.ProviderBodies {
+ if pb.Role == "request" {
+ prompts++
+ }
+ role, ok := bodies[[2]uint64{pb.Ref.Seq, pb.Ref.Row}]
+ switch {
+ case !ok:
+ bad("call %s lists a provider body at %d/%d that is no landed provider body", n.ID, pb.Ref.Seq, pb.Ref.Row)
+ case role != pb.Role:
+ bad("call %s lists a %s at %d/%d, and the record there is a %s", n.ID, pb.Role, pb.Ref.Seq, pb.Ref.Row, role)
+ case i > 0 && pb.Role == "request":
+ bad("call %s lists its request after another body", n.ID)
+ }
+ }
+ for i := range n.Children {
+ checkBodies(&n.Children[i])
+ }
+ }
+ for i := range doc.Talks {
+ checkBodies(&doc.Talks[i])
+ }
+ for i := range doc.Loose {
+ checkBodies(&doc.Loose[i])
+ }
+ if doc.Summary.CapturedPrompts != prompts || doc.Summary.ProviderBodies != len(bodies) {
+ bad("summary.provider_bodies is %d and captured_prompts %d, the root holds %d bodies and %d calls list a request",
+ doc.Summary.ProviderBodies, doc.Summary.CapturedPrompts, len(bodies), prompts)
+ }
if len(doc.Talks) != len(v.NodesByKind(model.KindTalk)) || doc.Summary.Talks != len(doc.Talks) {
bad("%d talks in the document, %d in the fold, summary says %d", len(doc.Talks), len(v.NodesByKind(model.KindTalk)), doc.Summary.Talks)
}
@@ -1154,3 +1242,239 @@ func firstN(s string, n int) string {
}
return s[:n]
}
+
+// providerRecords is the role of every landed provider body of a session, by
+// file sequence and row, a body landed twice counted once.
+func providerRecords(root, session string) map[[2]uint64]string {
+ out := map[[2]uint64]string{}
+ files, err := storage.LandedFiles(storage.NewZone(root), session)
+ if err != nil {
+ return out
+ }
+ seen := map[string]bool{}
+ for _, lf := range files {
+ if !strings.HasPrefix(filepath.Base(lf.Path), string(sessiondata.KindProviderBody)+"-") {
+ continue
+ }
+ f, err := os.Open(lf.Path)
+ if err != nil {
+ continue
+ }
+ _, recs, err := sessiondata.All(f)
+ f.Close()
+ if err != nil {
+ continue
+ }
+ for row, rec := range recs {
+ m, err := providerbody.ManifestOf(rec)
+ if err != nil || seen[rec.ID] {
+ continue
+ }
+ seen[rec.ID] = true
+ out[[2]uint64{lf.Seq, uint64(row + 1)}] = m.Role
+ }
+ }
+ return out
+}
+
+// ProviderBodiesRebuild reads every provider body a call lists the way a
+// reader that loads bodies on demand reads it: the session's provider_body
+// files with a sequence up to the one the body's ref names, in order, and
+// nothing else. Each body must rebuild to the digest and size its manifest
+// states. It also says how many provider files the session has, so a check
+// can tell that bodies referred across files.
+func ProviderBodiesRebuild(root, session string) ([]string, int, error) {
+ c, err := view.New(storage.NewZone(root), nil).Load(session)
+ if err != nil {
+ return nil, 0, err
+ }
+ doc, err := c.Build()
+ if err != nil {
+ return nil, 0, err
+ }
+ var out []string
+ bad := func(format string, a ...any) {
+ out = append(out, "provider_bodies_rebuild: "+fmt.Sprintf(format, a...))
+ }
+ var refs []sessionview.ProviderBody
+ var walk func(n *sessionview.Node)
+ walk = func(n *sessionview.Node) {
+ refs = append(refs, n.ProviderBodies...)
+ for i := range n.Children {
+ walk(&n.Children[i])
+ }
+ }
+ for i := range doc.Talks {
+ walk(&doc.Talks[i])
+ }
+ for i := range doc.Loose {
+ walk(&doc.Loose[i])
+ }
+ sort.Slice(refs, func(i, j int) bool {
+ if refs[i].Ref.Seq != refs[j].Ref.Seq {
+ return refs[i].Ref.Seq < refs[j].Ref.Seq
+ }
+ return refs[i].Ref.Row < refs[j].Ref.Row
+ })
+
+ files, err := storage.LandedFiles(storage.NewZone(root), session)
+ if err != nil {
+ return nil, 0, err
+ }
+ type landed struct {
+ seq uint64
+ recs []*sessiondata.Record
+ }
+ var provider []landed
+ for _, lf := range files {
+ if !strings.HasPrefix(filepath.Base(lf.Path), string(sessiondata.KindProviderBody)+"-") {
+ continue
+ }
+ f, err := os.Open(lf.Path)
+ if err != nil {
+ return nil, 0, err
+ }
+ _, recs, err := sessiondata.All(f)
+ f.Close()
+ if err != nil {
+ return nil, 0, err
+ }
+ provider = append(provider, landed{lf.Seq, recs})
+ }
+
+ // One session, fed file by file as the refs reach further, never past
+ // the file a ref names.
+ held := providerbody.NewSession()
+ next := 0
+ for _, r := range refs {
+ for next < len(provider) && provider[next].seq <= r.Ref.Seq {
+ for _, rec := range provider[next].recs {
+ if err := held.Add(rec); err != nil && !errors.Is(err, providerbody.ErrRepeat) {
+ bad("%s in file %d does not hold together: %v", rec.ID, provider[next].seq, err)
+ }
+ }
+ next++
+ }
+ var rec *sessiondata.Record
+ for _, p := range provider {
+ if p.seq == r.Ref.Seq && r.Ref.Row >= 1 && int(r.Ref.Row) <= len(p.recs) {
+ rec = p.recs[r.Ref.Row-1]
+ }
+ }
+ if rec == nil {
+ bad("a call lists %d/%d, which is no landed provider body", r.Ref.Seq, r.Ref.Row)
+ continue
+ }
+ m, err := providerbody.ManifestOf(rec)
+ if err != nil {
+ bad("%v", err)
+ continue
+ }
+ body, err := held.Body(rec.ID)
+ switch {
+ case err != nil:
+ bad("%s does not rebuild from the files up to %d: %v", rec.ID, r.Ref.Seq, err)
+ case len(body) != m.Bytes || providerbody.Digest(body) != m.SHA256:
+ bad("%s rebuilds to other bytes than its manifest states", rec.ID)
+ case m.Role != r.Role:
+ bad("%s is a %s, listed as a %s", rec.ID, m.Role, r.Role)
+ }
+ }
+ return out, len(provider), nil
+}
+
+// SameIdentity compares two folds of one scenario node by node, parent by
+// parent and relation by relation, not only by count. A node id built from a
+// landed position names the file's sequence, which files of another kind
+// landing between them shift, so a node with a reference is named by the
+// record it stands on: its kind, its stream, its file's source and the
+// record's line in that source, and the part.
+func SameIdentity(rootA, sessionA, rootB, sessionB string) ([]string, error) {
+ a, err := foldIdentity(rootA, sessionA)
+ if err != nil {
+ return nil, err
+ }
+ b, err := foldIdentity(rootB, sessionB)
+ if err != nil {
+ return nil, err
+ }
+ var out []string
+ for _, side := range []struct {
+ label string
+ has, not map[string]bool
+ }{{"only in the first", a, b}, {"only in the second", b, a}} {
+ var missing []string
+ for k := range side.has {
+ if !side.not[k] {
+ missing = append(missing, k)
+ }
+ }
+ sort.Strings(missing)
+ for i, k := range missing {
+ if i == 5 {
+ out = append(out, fmt.Sprintf("%s: … and %d more", side.label, len(missing)-5))
+ break
+ }
+ out = append(out, side.label+": "+k)
+ }
+ }
+ return out, nil
+}
+
+// positionID matches an id sessionflow.RefID builds: a kind, a landed
+// sequence and row, and a part.
+var positionID = regexp.MustCompile(`^[^/]+/[0-9]+/[0-9]+(:[0-9]+)?$`)
+
+func foldIdentity(root, session string) (map[string]bool, error) {
+ v, err := parse.View(root, session)
+ if err != nil {
+ return nil, err
+ }
+ files, err := storage.LandedFiles(storage.NewZone(root), session)
+ if err != nil {
+ return nil, err
+ }
+ type at struct{ seq, row uint64 }
+ source := map[at]string{}
+ for _, lf := range files {
+ f, err := os.Open(lf.Path)
+ if err != nil {
+ return nil, err
+ }
+ hdr, recs, err := sessiondata.All(f)
+ f.Close()
+ if err != nil {
+ return nil, err
+ }
+ for row, rec := range recs {
+ source[at{lf.Seq, uint64(row + 1)}] = fmt.Sprintf("%s#%d", hdr.Src, rec.Ord)
+ }
+ }
+ name := map[string]string{}
+ nameOf := func(id string) string {
+ if s, ok := name[id]; ok {
+ return s
+ }
+ n := v.Nodes[id]
+ // Only an id built from a landed position is renamed; any other id
+ // is compared as it is, since nothing but the evidence moves it.
+ if n == nil || n.Ref == nil || !positionID.MatchString(id) {
+ name[id] = id
+ return id
+ }
+ s := fmt.Sprintf("%s|%s|%s", n.Kind, n.Stream, source[at{n.Ref.Seq, n.Ref.Row}])
+ if n.Ref.Block != nil {
+ s += fmt.Sprintf(":%d", *n.Ref.Block)
+ }
+ name[id] = s
+ return s
+ }
+ out := map[string]bool{}
+ for id, n := range v.Nodes {
+ out["node "+nameOf(id)+" in "+nameOf(n.Parent)] = true
+ }
+ for _, r := range v.Relations {
+ out["relation "+r.Type+" "+nameOf(r.From)+" -> "+nameOf(r.To)] = true
+ }
+ return out, nil
+}
diff --git a/internal/scenario/plan.go b/internal/scenario/plan.go
index 9a6e187..8c81312 100644
--- a/internal/scenario/plan.go
+++ b/internal/scenario/plan.go
@@ -46,6 +46,10 @@ type Options struct {
// wrote, once all of it is sent. The build records it in the session's
// marker. It has no effect on the plan.
Remove Removal
+ // MaxDelta is the largest landed file, in bytes. The build writes it
+ // into the configuration as every local adapter's max_delta_bytes, and
+ // an sd build cuts its provider files at it. Zero leaves the default.
+ MaxDelta int64
}
// EventKind is one shape of record.
@@ -213,6 +217,10 @@ type Plan struct {
Events []Event
Streams []Stream
Runs []Run
+ // Provider says the plan's ids are those of a scenario with provider
+ // bodies, and bodies says it writes them. See ProviderBodies.
+ Provider bool
+ bodies bool
// at is the base time the plan was resolved against, which Span
// measures from.
at time.Time
@@ -273,7 +281,7 @@ func (sc *Scenario) Plan(opts Options) (*Plan, error) {
if scale == 0 {
scale = 1
}
- p := &Plan{Session: sc.Session, Project: sc.Project, Title: sc.Title, at: at, interval: interval, scale: scale}
+ p := &Plan{Session: sc.Session, Project: sc.Project, Title: sc.Title, Provider: sc.ProviderBodies, bodies: sc.ProviderBodies && !sc.omitBodies, at: at, interval: interval, scale: scale}
if p.Project == "" {
p.Project = "-Users-dev-scenario"
}
@@ -439,6 +447,16 @@ func (b *planner) call(l *lane, s *Step, id string) error {
}
}
call, req := id+"-call", id+"-req"
+ if b.p.Provider {
+ // A runtime's message and request ids are unique across sessions.
+ // Positions are not, and provider bodies of every session share one
+ // directory, named by the request id and joined by the message id,
+ // so they carry the session too. Sessions a repeated build makes
+ // differ only near the end of their ids, so all of it is used.
+ sum := sha256.Sum256([]byte(b.p.Session))
+ tag := hex.EncodeToString(sum[:6])
+ call, req = call+"-"+tag, req+"-"+tag
+ }
var frags []Event
if c.Thinking != "" {
text := c.Thinking
diff --git a/internal/scenario/provider.go b/internal/scenario/provider.go
new file mode 100644
index 0000000..af9df5c
--- /dev/null
+++ b/internal/scenario/provider.go
@@ -0,0 +1,267 @@
+// 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 scenario
+
+import (
+ "bytes"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "fmt"
+ "sort"
+ "strings"
+ "time"
+)
+
+// ProviderBodyDir is where, under the source root, a Claude Code build writes
+// the provider bodies: the directory OTEL_LOG_RAW_API_BODIES would name.
+const ProviderBodyDir = "provider-bodies"
+
+// ProviderBody is one body a claude-code build writes as a file, and an sd
+// build lands directly. Both formats take the same bytes in the same order.
+type ProviderBody struct {
+ Name string
+ At time.Time
+ Bytes []byte
+}
+
+// ProviderBodies renders the request and response of every provider call the
+// plan holds, in the order Claude Code writes them: a request when the call
+// is built, and its response when the last fragment has arrived.
+//
+// They follow what Claude Code 2.1.260 was measured to write. Every call
+// sends its stream's whole message list again. The newest message carries
+// the cache marker, and a one-block text message becomes a plain string once
+// the marker has left it. The first system block is a billing header naming
+// the prompt and the previous call's request. metadata.user_id is a JSON
+// string holding the session. A compaction starts the list again with the
+// summary. Thinking text is written as <REDACTED>, as Claude Code writes it.
+// A response is named by the request id; a request by a UUID drawn from the
+// call, so the same plan writes the same files.
+func (p *Plan) ProviderBodies() []ProviderBody {
+ if !p.bodies {
+ return nil
+ }
+ lost := p.lostStreams()
+ chains := map[string]*bodyChain{}
+ chainOf := func(stream string) *bodyChain {
+ c := chains[stream]
+ if c == nil {
+ c = &bodyChain{main: stream == "main"}
+ chains[stream] = c
+ }
+ return c
+ }
+ var out []ProviderBody
+ done := map[string]bool{}
+ for i := range p.Events {
+ e := &p.Events[i]
+ if e.Lost || lost[e.Stream] || e.Replayed {
+ continue
+ }
+ c := chainOf(e.Stream)
+ switch e.Kind {
+ case EvInput, EvNotice:
+ c.prompt = e.Run
+ c.user(e.Text)
+ case EvSummary:
+ c.prompt = e.Run
+ c.msgs = nil
+ c.user(e.Text)
+ case EvResult:
+ c.push(map[string]any{"role": "user", "content": []any{
+ map[string]any{"type": "tool_result", "tool_use_id": e.Of, "content": e.Text},
+ }})
+ case EvFragment:
+ if done[e.Call] {
+ continue
+ }
+ if c.call != e.Call {
+ c.call, c.blocks = e.Call, nil
+ out = append(out, ProviderBody{
+ Name: requestName(p.Session, e.Call),
+ At: e.At.Add(-time.Millisecond),
+ Bytes: c.request(p.Session),
+ })
+ }
+ c.blocks = append(c.blocks, fragmentBlock(e))
+ if e.Last {
+ done[e.Call] = true
+ out = append(out, ProviderBody{
+ Name: e.Req + ".response.json",
+ At: e.At.Add(time.Millisecond),
+ Bytes: encodeBody(map[string]any{
+ "id": e.Call, "type": "message", "role": "assistant", "model": ccModel,
+ "content": c.blocks, "stop_reason": e.Stop,
+ }),
+ })
+ c.push(map[string]any{"role": "assistant", "content": c.blocks})
+ c.prev, c.call, c.blocks = e.Req, "", nil
+ }
+ }
+ }
+ // The order the adapter lands them in: by when the file was written,
+ // then by name. An sd build lands them in the same order, so a body
+ // holds the same parts in both formats.
+ sort.SliceStable(out, func(i, j int) bool {
+ if !out[i].At.Equal(out[j].At) {
+ return out[i].At.Before(out[j].At)
+ }
+ return out[i].Name < out[j].Name
+ })
+ return out
+}
+
+// WithoutProviderBodies is the scenario with no provider bodies written, and
+// every id as it is with them.
+func (sc *Scenario) WithoutProviderBodies() *Scenario {
+ out := *sc
+ out.omitBodies = true
+ return &out
+}
+
+type bodyChain struct {
+ main bool
+ msgs []map[string]any
+ prompt string
+ prev string
+ call string
+ blocks []any
+}
+
+// user adds a prompt. The first message of a chain carries the injected
+// reminder beside it, as Claude Code's does.
+func (c *bodyChain) user(text string) {
+ if len(c.msgs) == 0 {
+ reminder := "<system-reminder>\nAs you answer the user's questions, you can use the following context.\n</system-reminder>"
+ if c.main {
+ reminder = "<system-reminder>\nCodebase and user instructions are shown below.\n" + strings.Repeat("Follow the project's rules. ", 60) + "\n</system-reminder>"
+ }
+ c.push(map[string]any{"role": "user", "content": []any{
+ map[string]any{"type": "text", "text": reminder},
+ map[string]any{"type": "text", "text": text},
+ }})
+ return
+ }
+ c.push(map[string]any{"role": "user", "content": []any{map[string]any{"type": "text", "text": text}}})
+}
+
+// push adds a message, moving the cache marker to it.
+func (c *bodyChain) push(m map[string]any) {
+ for _, old := range c.msgs {
+ blocks, ok := old["content"].([]any)
+ if !ok {
+ continue
+ }
+ for _, b := range blocks {
+ if block, ok := b.(map[string]any); ok {
+ delete(block, "cache_control")
+ }
+ }
+ if len(blocks) == 1 {
+ if block, ok := blocks[0].(map[string]any); ok && block["type"] == "text" {
+ old["content"] = block["text"]
+ }
+ }
+ }
+ if blocks, ok := m["content"].([]any); ok && len(blocks) > 0 {
+ last := map[string]any{}
+ for k, v := range blocks[len(blocks)-1].(map[string]any) {
+ last[k] = v
+ }
+ last["cache_control"] = map[string]any{"type": "ephemeral"}
+ copied := append([]any(nil), blocks...)
+ copied[len(copied)-1] = last
+ m = map[string]any{"role": m["role"], "content": copied}
+ }
+ c.msgs = append(c.msgs, m)
+}
+
+// request renders the body of the call about to be made.
+func (c *bodyChain) request(session string) []byte {
+ header := "x-anthropic-billing-header: cc_version=2.1.260; cc_entrypoint=cli; cch=00000;"
+ if c.prev != "" {
+ header += " cc_prev_req=" + c.prev + ";"
+ }
+ if c.prompt != "" {
+ header += " cc_prompt_id=" + c.prompt + ";"
+ }
+ system := "You are Claude Code, working in a scenario. " + strings.Repeat("Use the tools to answer. ", 80)
+ tools := []any{
+ map[string]any{"name": "Read", "description": strings.Repeat("Reads a file from the local filesystem. ", 40),
+ "input_schema": map[string]any{"type": "object", "properties": map[string]any{"file_path": map[string]any{"type": "string"}}}},
+ map[string]any{"name": "Bash", "description": strings.Repeat("Executes a shell command and returns its output. ", 40),
+ "input_schema": map[string]any{"type": "object", "properties": map[string]any{"command": map[string]any{"type": "string"}}}},
+ }
+ if !c.main {
+ system = "You are an agent that searches a repository. " + strings.Repeat("Report what you find. ", 60)
+ tools = tools[:1]
+ }
+ user, _ := json.Marshal(map[string]string{"device_id": "scenario", "account_uuid": "scenario", "session_id": session})
+ type body struct {
+ Model string `json:"model"`
+ Messages []map[string]any `json:"messages"`
+ System []any `json:"system"`
+ Tools []any `json:"tools"`
+ Metadata map[string]any `json:"metadata"`
+ MaxTokens int `json:"max_tokens"`
+ }
+ return encodeBody(body{
+ Model: ccModel, Messages: c.msgs,
+ System: []any{
+ map[string]any{"type": "text", "text": header},
+ map[string]any{"type": "text", "text": system, "cache_control": map[string]any{"type": "ephemeral"}},
+ },
+ Tools: tools,
+ Metadata: map[string]any{"user_id": string(user)},
+ MaxTokens: 64000,
+ })
+}
+
+func fragmentBlock(e *Event) any {
+ switch e.Frag {
+ case FragThinking:
+ return map[string]any{"type": "thinking", "thinking": "<REDACTED>", "signature": "sig"}
+ case FragToolUse:
+ input := e.Tool.Input
+ if input == nil {
+ input = map[string]any{}
+ }
+ return map[string]any{"type": "tool_use", "id": e.Tool.ID, "name": e.Tool.Name, "input": input}
+ default:
+ return map[string]any{"type": "text", "text": e.Text}
+ }
+}
+
+// encodeBody writes compact JSON with no HTML escapes, the way JSON.stringify
+// does.
+func encodeBody(v any) []byte {
+ var buf bytes.Buffer
+ enc := json.NewEncoder(&buf)
+ enc.SetEscapeHTML(false)
+ _ = enc.Encode(v)
+ return bytes.TrimSuffix(buf.Bytes(), []byte{'\n'})
+}
+
+// requestName is the file name of a call's request: a UUID drawn from the
+// session and the call, in the form Claude Code's random ones take.
+func requestName(session, call string) string {
+ sum := sha256.Sum256([]byte("request:" + session + ":" + call))
+ h := hex.EncodeToString(sum[:16])
+ return fmt.Sprintf("%s-%s-4%s-8%s-%s.request.json", h[0:8], h[8:12], h[13:16], h[17:20], h[20:32])
+}
diff --git a/internal/scenario/remove/remove.go b/internal/scenario/remove/remove.go
index 017cf30..642ecc2 100644
--- a/internal/scenario/remove/remove.go
+++ b/internal/scenario/remove/remove.go
@@ -46,10 +46,12 @@ import (
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecode"
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecodechanges"
+ "github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecodeprovider"
"github.com/apache/skywalking-ai-sessionizer/internal/export/otlp"
"github.com/apache/skywalking-ai-sessionizer/internal/metrics"
"github.com/apache/skywalking-ai-sessionizer/internal/scenario"
"github.com/apache/skywalking-ai-sessionizer/internal/storage"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/providerbody"
"github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
)
@@ -68,6 +70,10 @@ const conversationsDir = "_conversations"
// there, and a removal checks that the changes adapter reads it there.
const changesPrefix = "plugins/data/"
+// providerPrefix is where, under the local adapter's source directory, the
+// provider adapter's source directory is. The build writes the bodies there.
+const providerPrefix = scenario.ProviderBodyDir + "/"
+
// Remover removes the sessions a scenario build marked, once all of each is
// sent. The pipeline calls Pass at the end of every pass, after the push,
// while it holds the scenario lock.
@@ -82,6 +88,10 @@ type Remover struct {
Changes *claudecodechanges.Collector
ChangesRoot string
ChangesMatch func(claudecodechanges.Session) bool
+ // Provider is nil when the provider adapter is not enabled. A session
+ // whose build wrote provider bodies waits for it.
+ Provider *claudecodeprovider.Collector
+ ProviderRoot string
// Derives says claude-code-local derives metrics, so every landed file
// must be derived before its session can go.
Derives bool
@@ -349,6 +359,12 @@ func (r *Remover) Pass(in Input) *Result {
res.Errors = append(res.Errors, err)
return res
}
+ // The provider adapter's table names the sessions' body files, which are
+ // gone with them.
+ if err := claudecodeprovider.Forget(r.Zone, ids, p.now); err != nil {
+ res.Errors = append(res.Errors, err)
+ return res
+ }
if err := r.hook(Point{Step: 6, Where: WhereDone}); err != nil {
res.Errors = append(res.Errors, err)
return res
@@ -472,7 +488,11 @@ func (r *Remover) Blocked(in Input) string {
// not only the one this process's environment selects. The pipeline can
// run with CLAUDE_CONFIG_DIR or XDG_CONFIG_HOME set differently from the
// Claude Code a person runs on the same machine.
- for _, src := range []string{r.Source, r.ChangesRoot} {
+ sources := []string{r.Source, r.ChangesRoot}
+ if r.Provider != nil {
+ sources = append(sources, r.ProviderRoot)
+ }
+ for _, src := range sources {
for _, own := range claudeCodeDirs() {
if overlaps(src, own) {
return fmt.Sprintf("%s is, holds, or lies inside %s, where Claude Code keeps its own files; asz never removes there", src, own)
@@ -545,6 +565,8 @@ type session struct {
hasLocal bool
changes claudecodechanges.Session
hasChanges bool
+ // providerFiles are the body files the marker lists, by name.
+ providerFiles []string
}
func (s *session) unlock() {
@@ -693,6 +715,34 @@ func (p *pass) whole(s *session) verdict {
for _, src := range s.changes.Sources {
found[changesPrefix+src.Rel] = true
}
+ // The provider bodies the build wrote are found where the adapter reads
+ // them, and every body the adapter gave this session must be one of them.
+ for _, f := range s.m.Files {
+ if strings.HasPrefix(f.Path, providerPrefix) {
+ s.providerFiles = append(s.providerFiles, strings.TrimPrefix(f.Path, providerPrefix))
+ }
+ }
+ if len(s.providerFiles) > 0 {
+ want := resolved(filepath.Join(p.r.Source, scenario.ProviderBodyDir))
+ if p.r.Provider == nil {
+ return blocked("the provider adapter is off; the bodies the build wrote under %s would never be landed", want)
+ }
+ if got := resolved(p.r.ProviderRoot); got != want {
+ return blocked("the provider adapter reads %s, not %s; the bodies the build wrote would never be landed", p.r.ProviderRoot, want)
+ }
+ for _, name := range s.providerFiles {
+ found[providerPrefix+name] = true
+ }
+ }
+ if p.r.Provider != nil {
+ attributed, err := p.r.Provider.SessionFiles(s.id)
+ if err != nil {
+ return blocked("%v", err)
+ }
+ for name := range attributed {
+ found[providerPrefix+name] = true
+ }
+ }
for _, rel := range sortedKeys(found) {
if !want[rel] {
return blocked("%s is not a file the build wrote", p.source(rel))
@@ -751,6 +801,15 @@ func (p *pass) covered(s *session) verdict {
return waiting("%s", reason)
}
}
+ if len(s.providerFiles) > 0 {
+ ok, reason, err := p.r.Provider.Covered(s.id, s.providerFiles)
+ if err != nil {
+ return blocked("%v", err)
+ }
+ if !ok {
+ return waiting("%s", reason)
+ }
+ }
return verdict{}
}
@@ -796,6 +855,19 @@ func (p *pass) landed(s *session) (*landedSet, verdict) {
case claudecode.Name:
case claudecodechanges.Name:
src = changesPrefix + src
+ case claudecodeprovider.Name:
+ // A provider file holds many bodies, and each record names the
+ // body file it came from.
+ names, err := providerBodyFiles(lf.Path)
+ if err != nil {
+ return nil, blocked("landed file %s cannot be read: %v", rel, err)
+ }
+ for _, name := range names {
+ if !want[providerPrefix+name] {
+ return nil, blocked("landed file %s holds a body the build did not write", rel)
+ }
+ }
+ src = providerPrefix + names[0]
default:
return nil, blocked("landed file %s came through the adapter %q, which reads no file a build writes", rel, info.Adapter)
}
@@ -1280,3 +1352,29 @@ func sortedKeys(m map[string]bool) []string {
sort.Strings(out)
return out
}
+
+// providerBodyFiles lists the body file each record of a landed provider file
+// came from. A file with no record names none, and is refused.
+func providerBodyFiles(path string) ([]string, error) {
+ f, err := os.Open(path)
+ if err != nil {
+ return nil, err
+ }
+ defer f.Close()
+ _, recs, err := sessiondata.All(f)
+ if err != nil {
+ return nil, err
+ }
+ var out []string
+ for _, rec := range recs {
+ m, err := providerbody.ManifestOf(rec)
+ if err != nil {
+ return nil, err
+ }
+ out = append(out, m.Src)
+ }
+ if len(out) == 0 {
+ return nil, errors.New("it holds no body")
+ }
+ return out, nil
+}
diff --git a/internal/scenario/run/push.go b/internal/scenario/run/push.go
index 91f8e84..4b510ba 100644
--- a/internal/scenario/run/push.go
+++ b/internal/scenario/run/push.go
@@ -34,6 +34,7 @@ import (
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecode"
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecodechanges"
+ "github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecodeprovider"
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/mock"
"github.com/apache/skywalking-ai-sessionizer/internal/export/otlp"
"github.com/apache/skywalking-ai-sessionizer/internal/export/otlp/otlptest"
@@ -48,7 +49,7 @@ import (
// The file kinds the export page names.
var wireKinds = map[string]bool{
- "changes": true,
+ "changes": true, "provider_body": true,
"transcript": true, "agent_meta": true, "journal": true, "workflow_manifest": true, "workflow_script": true, "round": true,
}
@@ -105,7 +106,7 @@ func pushOver(protocol, out, session string, f scenario.Format, want *expect.Pus
// for a runtime format, the mock's for sd.
named := newPusher()
named.ServiceName = ""
- named.Runtimes = map[string]string{claudecode.Name: claudecode.RuntimeName, claudecodechanges.Name: claudecodechanges.RuntimeName, mock.Name: mock.RuntimeName}
+ named.Runtimes = map[string]string{claudecode.Name: claudecode.RuntimeName, claudecodechanges.Name: claudecodechanges.RuntimeName, claudecodeprovider.Name: claudecodeprovider.RuntimeName, mock.Name: mock.RuntimeName}
if _, err := named.Pass(); err != nil {
return nil, err
}
@@ -219,7 +220,13 @@ func pushOver(protocol, out, session string, f scenario.Format, want *expect.Pus
if a["asz.seq"] != fmt.Sprintf("int:%d", f.seq) {
bad("%s: seq %s, the header says %d", f.rel, a["asz.seq"], f.seq)
}
- if (a["asz.stream"] != "") == (a["asz.run"] != "") || a["asz.stream"] != f.stream || a["asz.run"] != f.batch {
+ // A file names a stream or a run, except a provider file, which
+ // belongs to the session and names neither.
+ names := (a["asz.stream"] != "") != (a["asz.run"] != "")
+ if f.kind == string(sessiondata.KindProviderBody) {
+ names = a["asz.stream"] == "" && a["asz.run"] == ""
+ }
+ if !names || a["asz.stream"] != f.stream || a["asz.run"] != f.batch {
bad("%s: stream %q run %q, the header says stream %q batch %q", f.rel, a["asz.stream"], a["asz.run"], f.stream, f.batch)
}
for _, k := range []string{"asz.conversation", "asz.round", "asz.session.from_time", "asz.session.through_time", "asz.conversation.title", "asz.conversation.talks"} {
diff --git a/internal/scenario/run/remove.go b/internal/scenario/run/remove.go
index f08accc..6ae020f 100644
--- a/internal/scenario/run/remove.go
+++ b/internal/scenario/run/remove.go
@@ -31,6 +31,7 @@ import (
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecode"
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecodechanges"
+ "github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecodeprovider"
"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/metrics"
@@ -146,8 +147,11 @@ func (c *removalCase) bad(format string, a ...any) {
}
func (c *removalCase) changesRoot() string { return filepath.Join(c.source, "plugins", "data") }
-func (c *removalCase) markerPath() string { return scenario.MarkerPath(c.source, c.session) }
-func (c *removalCase) chainDir() string { return filepath.Join(c.dir, "_conversations", c.session) }
+func (c *removalCase) providerRoot() string {
+ return filepath.Join(c.source, scenario.ProviderBodyDir)
+}
+func (c *removalCase) markerPath() string { return scenario.MarkerPath(c.source, c.session) }
+func (c *removalCase) chainDir() string { return filepath.Join(c.dir, "_conversations", c.session) }
func (c *removalCase) sourcePath(rel string) string {
return filepath.Join(c.source, filepath.FromSlash(rel))
@@ -169,7 +173,9 @@ func (c *removalCase) remover() *remove.Remover {
return &remove.Remover{Zone: c.zone, Source: c.source,
Local: claudecode.New(c.source, c.zone, 0),
Changes: claudecodechanges.New(c.changesRoot(), c.zone, 0), ChangesRoot: c.changesRoot(),
- Derives: true, Now: checkNow}
+ Provider: claudecodeprovider.New(c.providerRoot(), c.zone, 0),
+ ProviderRoot: c.providerRoot(),
+ Derives: true, Now: checkNow}
}
// push sends what the copy has not sent yet, and records the check's
@@ -528,7 +534,7 @@ func (c *removalCase) goneWhole(m *scenario.Marker, when string) error {
bad("the spool file %s is still there", it.Name())
}
}
- for _, state := range []string{filepath.Join(c.dir, otlp.StateFile), filepath.Join(c.dir, storage.SpoolDir, metrics.StateFile)} {
+ for _, state := range []string{filepath.Join(c.dir, otlp.StateFile), filepath.Join(c.dir, storage.SpoolDir, metrics.StateFile), c.zone.ProviderStatePath()} {
data, err := os.ReadFile(state)
if err != nil && !errors.Is(err, fs.ErrNotExist) {
return err
diff --git a/internal/scenario/run/run.go b/internal/scenario/run/run.go
index 31765cf..21e184a 100644
--- a/internal/scenario/run/run.go
+++ b/internal/scenario/run/run.go
@@ -31,6 +31,8 @@ import (
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecode"
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecodechanges"
+ "github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecodeprovider"
+ "github.com/apache/skywalking-ai-sessionizer/internal/config"
"github.com/apache/skywalking-ai-sessionizer/internal/metrics"
"github.com/apache/skywalking-ai-sessionizer/internal/parse"
"github.com/apache/skywalking-ai-sessionizer/internal/repack"
@@ -156,7 +158,7 @@ func checkFormat(sc *scenario.Scenario, ex *expect.File, f scenario.Format, out
name = "final"
}
var err error
- built, err = scenario.Build(sc, f, out, scenario.Options{At: opts.At, Scale: opts.Scale, Interval: opts.Interval, Through: cp})
+ built, err = scenario.Build(sc, f, out, scenario.Options{At: opts.At, Scale: opts.Scale, Interval: opts.Interval, Through: cp, MaxDelta: ex.Collect.MaxDeltaBytes})
if err != nil {
return nil, "", err
}
@@ -255,6 +257,19 @@ func checkFormat(sc *scenario.Scenario, ex *expect.File, f scenario.Format, out
}
return changesLeaveTheFold(sc, f, out, session, opts, ex.Parse.MaxRoundBytes)
}},
+ {"provider_bodies_leave_the_fold", ex.Properties.ProviderBodiesLeaveTheFold, func() ([]string, error) {
+ if !sc.ProviderBodies {
+ return nil, nil
+ }
+ return providerBodiesLeaveTheFold(sc, f, out, session, opts, ex.Parse.MaxRoundBytes)
+ }},
+ {"provider_bodies_rebuild", ex.Properties.ProviderBodiesRebuild, func() ([]string, error) {
+ if !sc.ProviderBodies {
+ return nil, nil
+ }
+ lines, _, err := expect.ProviderBodiesRebuild(out, session)
+ return lines, err
+ }},
{"reproducible", ex.Properties.Reproducible, func() ([]string, error) {
// Two parses of identical landed evidence must produce identical
// rounds. The chain in out was cut at checkpoints, so it is not
@@ -302,6 +317,13 @@ func checkFormat(sc *scenario.Scenario, ex *expect.File, f scenario.Format, out
if cs.Records != 0 || cs.SourcesLanded != 0 {
return []string{fmt.Sprintf("recollect_idempotent: a second collect of the changes landed %d records from %d sources", cs.Records, cs.SourcesLanded)}, nil
}
+ ps, err := claudecodeprovider.New(filepath.Join(out, "_source", scenario.ProviderBodyDir), zone, 0).CollectAll(collectEvery)
+ if err != nil {
+ return nil, err
+ }
+ if ps.Records != 0 || ps.SourcesLanded != 0 {
+ return []string{fmt.Sprintf("recollect_idempotent: a second collect of the provider bodies landed %d records from %d sources", ps.Records, ps.SourcesLanded)}, nil
+ }
return nil, nil
}},
struct {
@@ -315,8 +337,9 @@ func checkFormat(sc *scenario.Scenario, ex *expect.File, f scenario.Format, out
fn func() ([]string, error)
}{"parts_keep_source_bytes", ex.Properties.PartsKeepSourceBytes, func() ([]string, error) {
return partsKeepSourceBytes(out, session, map[string]string{
- claudecode.Name: filepath.Join(out, "_source"),
- claudecodechanges.Name: filepath.Join(out, "_source", "plugins", "data"),
+ claudecode.Name: filepath.Join(out, "_source"),
+ claudecodechanges.Name: filepath.Join(out, "_source", "plugins", "data"),
+ claudecodeprovider.Name: filepath.Join(out, "_source", scenario.ProviderBodyDir),
})
}},
struct {
@@ -397,7 +420,8 @@ func collect(f scenario.Format, out string) (int, error) {
if f != scenario.FormatClaudeCode {
return 0, nil
}
- st, err := claudecode.New(filepath.Join(out, "_source"), storage.NewZone(out), 0).CollectAll(nil)
+ budget := budgets(out)
+ st, err := claudecode.New(filepath.Join(out, "_source"), storage.NewZone(out), budget[config.AdapterClaudeCodeLocal]).CollectAll(nil)
if err != nil {
return 0, err
}
@@ -405,24 +429,63 @@ func collect(f scenario.Format, out string) (int, error) {
return 0, fmt.Errorf("collect: %v", st.Errors)
}
// The plugin's lines, through their own adapter, as asz view runs both.
- cs, err := claudecodechanges.New(filepath.Join(out, "_source", "plugins", "data"), storage.NewZone(out), 0).CollectAll(nil)
+ cs, err := claudecodechanges.New(filepath.Join(out, "_source", "plugins", "data"), storage.NewZone(out), budget[config.AdapterClaudeCodeChanges]).CollectAll(nil)
if err != nil {
return 0, err
}
if len(cs.Errors) != 0 {
return 0, fmt.Errorf("collect changes: %v", cs.Errors)
}
- return st.SourcesLanded + cs.SourcesLanded, nil
+ // The provider bodies last, as the pipeline collects them, so a
+ // response only a transcript names finds its session.
+ ps, err := claudecodeprovider.New(filepath.Join(out, "_source", scenario.ProviderBodyDir), storage.NewZone(out), budget[config.AdapterClaudeCodeProvider]).CollectAll(collectEvery)
+ if err != nil {
+ return 0, err
+ }
+ if len(ps.Errors) != 0 {
+ return 0, fmt.Errorf("collect provider bodies: %v", ps.Errors)
+ }
+ return st.SourcesLanded + cs.SourcesLanded + ps.SourcesLanded, nil
+}
+
+// budgets reads each local adapter's max_delta_bytes from the configuration
+// the build wrote, so a check lands files cut the way the pipeline over that
+// configuration cuts them. A root with no configuration, such as a copy the
+// removal check makes, lands at the defaults.
+func budgets(out string) map[string]int64 {
+ out2 := map[string]int64{}
+ cfg, err := config.Load(filepath.Join(out, "asz.yaml"))
+ if err != nil {
+ return out2
+ }
+ for _, ad := range cfg.Adapters {
+ out2[ad.Name] = ad.Collector.MaxDeltaBytes
+ }
+ return out2
}
+// collectEvery is the provider adapter's filter in a scenario root, where
+// every session is the build's own.
+func collectEvery(string) claudecodeprovider.Verdict { return claudecodeprovider.Collect }
+
// changesLeaveTheFold builds the scenario again without its changes and
// compares the folds: the nodes, the relations and the talks must be the
// same, because a change record is evidence beside a step and never a
// step. The chain's bytes differ, since a round binds the files it read.
func changesLeaveTheFold(sc *scenario.Scenario, f scenario.Format, out, session string, opts Options, maxRound int64) ([]string, error) {
+ return leavesTheFold("changes_leave_the_fold", "the changes", sc.WithoutChanges(), f, out, session, opts, maxRound)
+}
+
+// providerBodiesLeaveTheFold does the same for provider bodies: a body is
+// what a call sent and got back, joined to the call by a view, and no step.
+func providerBodiesLeaveTheFold(sc *scenario.Scenario, f scenario.Format, out, session string, opts Options, maxRound int64) ([]string, error) {
+ return leavesTheFold("provider_bodies_leave_the_fold", "the provider bodies", sc.WithoutProviderBodies(), f, out, session, opts, maxRound)
+}
+
+func leavesTheFold(property, what string, plainScenario *scenario.Scenario, f scenario.Format, out, session string, opts Options, maxRound int64) ([]string, error) {
plain := out + "-plain"
defer os.RemoveAll(plain)
- built, err := scenario.Build(sc.WithoutChanges(), f, plain, scenario.Options{At: opts.At, Scale: opts.Scale, Interval: opts.Interval})
+ built, err := scenario.Build(plainScenario, f, plain, scenario.Options{At: opts.At, Scale: opts.Scale, Interval: opts.Interval})
if err != nil {
return nil, err
}
@@ -442,7 +505,14 @@ func changesLeaveTheFold(sc *scenario.Scenario, f scenario.Format, out, session
}
var lines []string
for _, d := range expect.Compare(with, without) {
- lines = append(lines, "changes_leave_the_fold: the fold differs with and without the changes: "+d)
+ lines = append(lines, property+": the fold differs with and without "+what+": "+d)
+ }
+ same, err := expect.SameIdentity(out, session, plain, built.Session)
+ if err != nil {
+ return nil, err
+ }
+ for _, d := range same {
+ lines = append(lines, property+": a node or relation differs with and without "+what+": "+d)
}
return lines, nil
}
@@ -610,6 +680,15 @@ func repackKeepsStructure(out, session string, maxRound int64) ([]string, error)
for _, d := range expect.Compare(a, b) {
lines = append(lines, "repack_keeps_structure: "+d)
}
+ // Provider bodies refer to each other by record id and digest, never by
+ // file and row, so they rebuild on the re-cut root too.
+ rebuilt, _, err := expect.ProviderBodiesRebuild(twin, session)
+ if err != nil {
+ return nil, err
+ }
+ for _, l := range rebuilt {
+ lines = append(lines, "repack_keeps_structure: "+l)
+ }
return lines, nil
}
diff --git a/internal/scenario/run/sourcebytes.go b/internal/scenario/run/sourcebytes.go
index 9c59643..75aa3ad 100644
--- a/internal/scenario/run/sourcebytes.go
+++ b/internal/scenario/run/sourcebytes.go
@@ -28,6 +28,7 @@ import (
"github.com/apache/skywalking-ai-sessionizer/internal/storage"
"github.com/apache/skywalking-ai-sessionizer/pkg/changes"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/providerbody"
"github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
)
@@ -67,7 +68,9 @@ func partsKeepSourceBytes(root, session string, sources map[string]string) ([]st
if err != nil {
return nil, fmt.Errorf("%s: %w", lf.Path, err)
}
- if len(recs) > 0 && recs[0].Ord == 1 {
+ // Every provider record starts at ord 1, being a whole body, so none
+ // of them is an older version of another.
+ if len(recs) > 0 && recs[0].Ord == 1 && hdr.Kind != sessiondata.KindProviderBody {
newest[sourceKey{hdr.Adapter, hdr.Src}] = len(all)
}
all = append(all, landed{filepath.Base(lf.Path), hdr, recs})
@@ -76,12 +79,21 @@ func partsKeepSourceBytes(root, session string, sources map[string]string) ([]st
fail := func(format string, a ...any) {
out = append(out, "parts_keep_source_bytes: "+fmt.Sprintf(format, a...))
}
+ held := providerbody.NewSession()
for i, l := range all {
+ adapter, _, _ := strings.Cut(l.hdr.Adapter, "/")
+ dir, ok := sources[adapter]
+ if l.hdr.Kind == sessiondata.KindProviderBody {
+ if !ok {
+ fail("%s: no source directory for adapter %s", l.name, adapter)
+ continue
+ }
+ providerKeepsSourceBytes(l.name, dir, l.recs, held, fail)
+ continue
+ }
if i < newest[sourceKey{l.hdr.Adapter, l.hdr.Src}] {
continue
}
- adapter, _, _ := strings.Cut(l.hdr.Adapter, "/")
- dir, ok := sources[adapter]
if !ok {
fail("%s: no source directory for adapter %s", l.name, adapter)
continue
@@ -133,3 +145,50 @@ func partsKeepSourceBytes(root, session string, sources map[string]string) ([]st
}
return out, nil
}
+
+// providerKeepsSourceBytes checks provider bodies against the files the
+// runtime wrote: each record's digest and size are its file's, each piece it
+// keeps is a run of that file's bytes, and the body it rebuilds to, against
+// what the session held before it, is the file exactly.
+func providerKeepsSourceBytes(name, dir string, recs []*sessiondata.Record, held *providerbody.Session, fail func(string, ...any)) {
+ for row, rec := range recs {
+ m, err := providerbody.ManifestOf(rec)
+ if err != nil {
+ fail("%s row %d: %v", name, row+1, err)
+ continue
+ }
+ src, err := os.ReadFile(filepath.Join(dir, m.Src))
+ if err != nil {
+ fail("%s row %d: %v", name, row+1, err)
+ continue
+ }
+ sum := sha256.Sum256(src)
+ if got := hex.EncodeToString(sum[:])[:12]; got != rec.Sha || rec.Bytes != len(src) {
+ fail("%s row %d: sha %s and %d bytes, and its file %s gives %s and %d", name, row+1, rec.Sha, rec.Bytes, m.Src, got, len(src))
+ }
+ for b, p := range rec.Parts[:len(rec.Parts)-1] {
+ kept := []byte(p.Data)
+ if p.Kind == sessiondata.PartUnknown {
+ if kept, err = p.Raw(); err != nil {
+ fail("%s row %d block %d: %v", name, row+1, b, err)
+ continue
+ }
+ }
+ if !bytes.Contains(src, kept) {
+ fail("%s row %d block %d: the piece is not its file's bytes", name, row+1, b)
+ }
+ }
+ if err := held.Add(rec); err != nil {
+ fail("%s row %d: %v", name, row+1, err)
+ continue
+ }
+ body, err := held.Body(rec.ID)
+ if err != nil {
+ fail("%s row %d: %v", name, row+1, err)
+ continue
+ }
+ if !bytes.Equal(body, src) {
+ fail("%s row %d: the body rebuilds to other bytes than %s", name, row+1, m.Src)
+ }
+ }
+}
diff --git a/internal/scenario/scenario.go b/internal/scenario/scenario.go
index a022fcd..bd12ebe 100644
--- a/internal/scenario/scenario.go
+++ b/internal/scenario/scenario.go
@@ -53,7 +53,17 @@ type Scenario struct {
// Interval is the session's clock: the gap between consecutive steps in
// every stream, unless a step says after. Zero means one second.
Interval time.Duration `yaml:"interval"`
- Steps []Step `yaml:"steps"`
+ // ProviderBodies writes the request and response body of every provider
+ // call, as Claude Code does when OTEL_LOG_RAW_API_BODIES names a
+ // directory. A claude-code build writes them as files; an sd build lands
+ // them.
+ ProviderBodies bool `yaml:"provider_bodies"`
+ Steps []Step `yaml:"steps"`
+
+ // omitBodies writes no provider bodies while keeping everything else a
+ // scenario with provider bodies has, its ids included, so the two builds
+ // can be compared.
+ omitBodies bool
}
// Step is one thing that happened. Exactly one of the kind fields is set.
diff --git a/internal/scenario/write_claudecode.go b/internal/scenario/write_claudecode.go
index 8ddcc03..911a34d 100644
--- a/internal/scenario/write_claudecode.go
+++ b/internal/scenario/write_claudecode.go
@@ -160,6 +160,25 @@ func writeClaudeCode(p *Plan, root string) ([]string, []MarkerFile, error) {
return nil, nil, err
}
}
+ // The provider bodies, where Claude Code writes them: one flat directory
+ // for every session, each file stamped with when it was written, which
+ // is the order the adapter lands them in.
+ for _, b := range p.ProviderBodies() {
+ rel := ProviderBodyDir + "/" + b.Name
+ path := filepath.Join(root, filepath.FromSlash(rel))
+ if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
+ return nil, nil, err
+ }
+ if err := os.WriteFile(path, b.Bytes, 0o644); err != nil {
+ return nil, nil, err
+ }
+ if err := os.Chtimes(path, b.At, b.At); err != nil {
+ return nil, nil, err
+ }
+ written = append(written, path)
+ sum := sha256.Sum256(b.Bytes)
+ marked[rel] = MarkerFile{Path: rel, Size: int64(len(b.Bytes)), SHA256: hex.EncodeToString(sum[:])}
+ }
sort.Strings(written)
files := make([]MarkerFile, 0, len(marked))
for _, f := range marked {
diff --git a/internal/scenario/write_sd.go b/internal/scenario/write_sd.go
index 2364e24..aad7375 100644
--- a/internal/scenario/write_sd.go
+++ b/internal/scenario/write_sd.go
@@ -20,17 +20,21 @@ package scenario
import (
"crypto/sha256"
"encoding/hex"
+ "errors"
"fmt"
"io"
+ "os"
"path/filepath"
"sort"
"time"
+ "github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecodeprovider"
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/mock"
"github.com/apache/skywalking-ai-sessionizer/internal/index"
"github.com/apache/skywalking-ai-sessionizer/internal/storage"
"github.com/apache/skywalking-ai-sessionizer/pkg/changes"
"github.com/apache/skywalking-ai-sessionizer/pkg/model"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/providerbody"
"github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
)
@@ -40,7 +44,7 @@ import (
// checkpoint lands only what is new, sequences taken from the session's
// state, and every file write-once. asz verify passes on the result and the
// parser reads it like any other root.
-func writeSD(p *Plan, root string, now time.Time) ([]string, error) {
+func writeSD(p *Plan, root string, now time.Time, maxDelta int64) ([]string, error) {
z := storage.NewZone(root)
sessionDir := z.SessionDir(p.Session)
if err := mkdirAll(sessionDir); err != nil {
@@ -59,7 +63,10 @@ func writeSD(p *Plan, root string, now time.Time) ([]string, error) {
if err := state.RecoverNextSeq(sessionDir); err != nil {
return nil, err
}
- w := &sdWriter{p: p, z: z, state: state, now: now}
+ if maxDelta <= 0 {
+ maxDelta = 2 << 20
+ }
+ w := &sdWriter{p: p, z: z, state: state, now: now, maxDelta: maxDelta}
// Streams first, main before its children, then the runs' files, so the
// sequences a round refers to come out in landed order. A lost record
@@ -98,6 +105,9 @@ func writeSD(p *Plan, root string, now time.Time) ([]string, error) {
return nil, err
}
}
+ if err := w.provider(); err != nil {
+ return nil, err
+ }
if err := state.Save(statePath, now); err != nil {
return nil, err
}
@@ -142,11 +152,12 @@ func (w *sdWriter) reindex() error {
}
type sdWriter struct {
- p *Plan
- z *storage.Zone
- state *storage.SessionState
- now time.Time
- written []string
+ p *Plan
+ z *storage.Zone
+ state *storage.SessionState
+ now time.Time
+ maxDelta int64
+ written []string
}
// source names a stream's or a run's file as the header's src: the mock has
@@ -426,3 +437,118 @@ func finish(r *sessiondata.Record, ord uint64, off *uint64) {
r.Sha, r.Bytes = hex.EncodeToString(sum[:6]), len(b)
*off += uint64(len(b)) + 1
}
+
+// provider lands the plan's provider bodies that the cursor has not seen, cut
+// against what the session already holds, into files of at most maxDelta
+// bytes, as the adapter cuts them. The adapter lands the same bodies in the
+// same order, so each record holds the same parts.
+func (w *sdWriter) provider() error {
+ bodies := w.p.ProviderBodies()
+ if len(bodies) == 0 {
+ return nil
+ }
+ dir := w.z.ProviderDir(w.p.Session)
+ if err := mkdirAll(dir); err != nil {
+ return err
+ }
+ cursorPath := filepath.Join(dir, string(sessiondata.KindProviderBody)+".cursor")
+ cur, err := storage.LoadCursor(cursorPath, storage.CursorAppend, ".")
+ if err != nil {
+ return err
+ }
+ held := providerbody.NewSession()
+ files, err := storage.LandedFiles(w.z, w.p.Session)
+ if err != nil {
+ return err
+ }
+ for _, lf := range files {
+ if lf.Stream != "" || lf.RunID != "" {
+ continue
+ }
+ _, recs, err := readAll(lf.Path)
+ if err != nil {
+ return err
+ }
+ for _, r := range recs {
+ if err := held.Add(r); err != nil && !errors.Is(err, providerbody.ErrRepeat) {
+ return err
+ }
+ }
+ }
+ // A body is new when the session does not hold it, not when it comes
+ // after the ones a build landed before: a later checkpoint can plan a
+ // body that sorts before a child's bodies an earlier one landed.
+ var recs []*sessiondata.Record
+ for _, b := range bodies {
+ if _, ok := held.Holds(claudecodeprovider.ID(b.Name)); ok {
+ continue
+ }
+ rec, err := held.Encode(claudecodeprovider.BodyOf(b.Name, b.Bytes))
+ if err != nil {
+ return fmt.Errorf("scenario: provider body %s: %w", b.Name, err)
+ }
+ recs = append(recs, rec)
+ }
+ if len(recs) == 0 {
+ return nil
+ }
+ var last uint64
+ flush := func(batch []*sessiondata.Record) error {
+ seq := w.state.Take()
+ hdr := sessiondata.Header{
+ H: 1, Seq: seq, At: w.now.UTC().Format(time.RFC3339Nano), Kind: sessiondata.KindProviderBody,
+ Adapter: mock.Name + "/" + mock.Version, Dialect: mock.Dialect, Src: ".", Session: w.p.Session,
+ }
+ path := filepath.Join(dir, storage.LandedName(string(sessiondata.KindProviderBody), storage.Stamp(w.now), seq))
+ err := storage.WriteAtomic(path, storage.PermLanded, func(out io.Writer) error {
+ rw, err := sessiondata.NewWriter(out, &hdr)
+ if err != nil {
+ return err
+ }
+ for _, r := range batch {
+ if err := rw.Write(r); err != nil {
+ return err
+ }
+ }
+ return rw.Close()
+ })
+ if err != nil {
+ return err
+ }
+ w.written = append(w.written, path)
+ last = seq
+ return nil
+ }
+ var batch []*sessiondata.Record
+ var size int64
+ for _, r := range recs {
+ // As the adapter cuts: a file ends before a body that would take it
+ // past the budget.
+ n := providerbody.RecordBytes(r)
+ if len(batch) > 0 && providerbody.FileOverhead+size+n > w.maxDelta {
+ if err := flush(batch); err != nil {
+ return err
+ }
+ batch, size = nil, 0
+ }
+ batch = append(batch, r)
+ size += n
+ }
+ if len(batch) > 0 {
+ if err := flush(batch); err != nil {
+ return err
+ }
+ }
+ seq := last
+ cur.Ord, cur.LastSeq, cur.State = uint64(len(bodies)), seq, storage.CursorActive
+ return cur.Save(cursorPath, w.now)
+}
+
+func readAll(path string) (sessiondata.Header, []*sessiondata.Record, error) {
+ f, err := os.Open(path)
+ if err != nil {
+ return sessiondata.Header{}, nil, err
+ }
+ defer f.Close()
+ return sessiondata.All(f)
+}
diff --git a/internal/storage/landed.go b/internal/storage/landed.go
index ef6d42d..7f971cb 100644
--- a/internal/storage/landed.go
+++ b/internal/storage/landed.go
@@ -35,6 +35,8 @@ type LandedFile struct {
Seq uint64
Stream string // set for a stream's records
RunID string // set for a workflow run's records
+ // Neither is set for a file at the session level: the provider bodies,
+ // which belong to no one stream or run.
}
// LandedFiles lists every landed record file for a session, in sequence order.
@@ -99,6 +101,10 @@ func LandedFiles(z *Zone, session string) ([]LandedFile, error) {
}
}
+ if err := add(z.ProviderDir(session), "", false); err != nil {
+ return nil, err
+ }
+
sort.Slice(out, func(i, j int) bool { return out[i].Seq < out[j].Seq })
return out, nil
}
diff --git a/internal/storage/lock.go b/internal/storage/lock.go
index f8e984f..22032ca 100644
--- a/internal/storage/lock.go
+++ b/internal/storage/lock.go
@@ -81,6 +81,23 @@ func LockExport(root string) (*SessionLock, error) {
return l, err
}
+// ErrProviderBusy means another collector holds this root's provider state.
+var ErrProviderBusy = errors.New("storage: the provider body state is held by another collector")
+
+// LockProvider takes an exclusive lock over one storage root's provider body
+// state.
+//
+// The runtime writes every session's bodies into one flat directory, so which
+// of them have landed is known for the root, not for a session. Two collectors
+// deciding that at once would both land a new body into its session.
+func LockProvider(root string) (*SessionLock, error) {
+ l, err := lockDir(filepath.Join(root, "_provider"))
+ if errors.Is(err, ErrSessionBusy) {
+ return nil, ErrProviderBusy
+ }
+ return l, err
+}
+
// LockExportWait is LockExport, waiting up to timeout for the lock rather
// than giving up at once.
//
diff --git a/internal/storage/zone.go b/internal/storage/zone.go
index d9c6a26..fcc737b 100644
--- a/internal/storage/zone.go
+++ b/internal/storage/zone.go
@@ -76,6 +76,21 @@ func (z *Zone) RunDir(session, runID string) string {
return filepath.Join(z.SessionDir(session), "runs", runID)
}
+// ProviderDir holds the session's provider bodies: what the runtime sent its
+// model provider and what came back. It sits at the session level, not under
+// a stream, because a body names no stream. Which stream and which call it
+// belongs to is joined later, from evidence, and a path must not say it.
+func (z *Zone) ProviderDir(session string) string {
+ return filepath.Join(z.SessionDir(session), "provider_body")
+}
+
+// ProviderStatePath is where the root keeps which provider body files have
+// landed, and which wait for a session. It is derived from the landed records
+// and rebuilt when it is missing.
+func (z *Zone) ProviderStatePath() string {
+ return filepath.Join(z.root, "_provider", "seen.state")
+}
+
// IndexDir holds the session's derived lookup index.
//
// The index is disposable: deleting it loses nothing, because it rebuilds from
diff --git a/internal/verify/provider.go b/internal/verify/provider.go
new file mode 100644
index 0000000..40a2ed2
--- /dev/null
+++ b/internal/verify/provider.go
@@ -0,0 +1,100 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package verify
+
+import (
+ "errors"
+ "fmt"
+ "io"
+ "os"
+ "path/filepath"
+
+ "github.com/apache/skywalking-ai-sessionizer/pkg/providerbody"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+)
+
+// BodyProblem is a provider body that does not come back whole.
+type BodyProblem struct {
+ File string
+ Row uint32
+ Why string
+}
+
+func (p BodyProblem) String() string {
+ return fmt.Sprintf("%s row %d: %s", filepath.Base(p.File), p.Row, p.Why)
+}
+
+// Provider checks a session's provider bodies.
+//
+// Each body is a whole document, one record at ord 1 and offset 0, so the
+// line and byte checks of Stream say nothing about them. What can go wrong is
+// a body that no longer rebuilds: a piece or an earlier body it names is not
+// there, or the bytes it rebuilds to do not match its digest. So every body is
+// rebuilt, in landing order, and compared. A body landed twice with the same
+// digest is a repeat, as in a stream.
+func Provider(dir string) (*StreamReport, []BodyProblem, error) {
+ files, err := landedFiles(dir, string(sessiondata.KindProviderBody))
+ if err != nil {
+ return nil, nil, err
+ }
+ rep := &StreamReport{Dir: dir, Files: len(files)}
+ var problems []BodyProblem
+ s := providerbody.NewSession()
+ for _, path := range files {
+ f, err := os.Open(path)
+ if err != nil {
+ return nil, nil, err
+ }
+ r, err := sessiondata.NewReader(f)
+ if err != nil {
+ f.Close()
+ return nil, nil, fmt.Errorf("verify: %s: %w", path, err)
+ }
+ for row := uint32(1); ; row++ {
+ rec, rerr := r.Next()
+ if errors.Is(rerr, io.EOF) {
+ break
+ }
+ if rerr != nil {
+ f.Close()
+ return nil, nil, fmt.Errorf("verify: %s row %d: %w", path, row, rerr)
+ }
+ rep.Records++
+ rep.FirstOrd, rep.LastOrd = 1, 1
+ if err := s.Add(rec); err != nil {
+ if errors.Is(err, providerbody.ErrRepeat) {
+ rep.Relanded++
+ continue
+ }
+ problems = append(problems, BodyProblem{File: path, Row: row, Why: err.Error()})
+ continue
+ }
+ m, _ := providerbody.ManifestOf(rec)
+ body, err := s.Body(rec.ID)
+ switch {
+ case err != nil:
+ problems = append(problems, BodyProblem{File: path, Row: row, Why: err.Error()})
+ case rec.Bytes != len(body) || len(m.SHA256) < 12 || rec.Sha != m.SHA256[:12]:
+ problems = append(problems, BodyProblem{File: path, Row: row,
+ Why: "the record's size or digest is not the body's"})
+ }
+ }
+ f.Close()
+ }
+ return rep, problems, nil
+}
diff --git a/internal/verify/session.go b/internal/verify/session.go
index 4503b4f..d9dbcd7 100644
--- a/internal/verify/session.go
+++ b/internal/verify/session.go
@@ -40,6 +40,9 @@ type SessionReport struct {
// Relanded counts records an interrupted pass caused to be landed twice.
// It is reported so it is visible, not because it is wrong.
Relanded int
+ // Bodies lists the provider bodies that do not rebuild. They are counted
+ // in Problems too.
+ Bodies []BodyProblem
}
// OK reports whether every stream in the session is contiguous and intact.
@@ -96,6 +99,17 @@ func Session(z *storage.Zone, session string) (*SessionReport, error) {
}
}
}
+ pr, bodies, err := Provider(z.ProviderDir(session))
+ if err != nil {
+ return nil, err
+ }
+ if pr.Files > 0 {
+ rep.Streams = append(rep.Streams, pr)
+ rep.Records += pr.Records
+ rep.Relanded += pr.Relanded
+ rep.Bodies = bodies
+ rep.Problems += len(bodies)
+ }
return rep, nil
}
@@ -116,5 +130,8 @@ func (r *SessionReport) Details() []string {
out = append(out, fmt.Sprintf("end gap %s", *s.End))
}
}
+ for _, b := range r.Bodies {
+ out = append(out, fmt.Sprintf("body %s", b))
+ }
return out
}
diff --git a/internal/view/conversationview.go b/internal/view/conversationview.go
index 9993130..83b3107 100644
--- a/internal/view/conversationview.go
+++ b/internal/view/conversationview.go
@@ -33,6 +33,7 @@ import (
"github.com/apache/skywalking-ai-sessionizer/internal/storage"
"github.com/apache/skywalking-ai-sessionizer/internal/verify"
"github.com/apache/skywalking-ai-sessionizer/pkg/model"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/providerbody"
"github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
"github.com/apache/skywalking-ai-sessionizer/pkg/sessionflow"
"github.com/apache/skywalking-ai-sessionizer/pkg/sessionview"
@@ -174,6 +175,21 @@ func (c *Conversation) Build() (*sessionview.Conversation, error) {
annotateChanges(v.Talks, byStep)
annotateChanges(v.Loose, byStep)
v.Summary.Changes = len(v.WorkspaceChanges)
+
+ // The provider bodies, joined to their calls. A call carries where its
+ // request and response are landed; a reader loads them when it wants
+ // them, and the document stays the size of the structure.
+ total, byCall := c.providerBodies(landed)
+ annotateProvider(v.Talks, byCall)
+ annotateProvider(v.Loose, byCall)
+ v.Summary.ProviderBodies = total
+ for _, bodies := range byCall {
+ for _, b := range bodies {
+ if b.Role == providerbody.RoleRequest {
+ v.Summary.CapturedPrompts++
+ }
+ }
+ }
c.built = v
return v, nil
}
diff --git a/internal/view/provider.go b/internal/view/provider.go
new file mode 100644
index 0000000..82f2534
--- /dev/null
+++ b/internal/view/provider.go
@@ -0,0 +1,322 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package view
+
+import (
+ "bytes"
+ "encoding/json"
+ "errors"
+ "io"
+ "os"
+ "path/filepath"
+ "slices"
+ "sort"
+ "strings"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/storage"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/model"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/providerbody"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessionflow"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessionview"
+)
+
+// providerBodies joins the session's provider bodies to their calls. It
+// returns how many bodies the session holds, and each call step's joined
+// bodies.
+//
+// A response joins by its message id, which is the call's own. A request
+// names no call, only the request before it and its prompt, so a request
+// joins to the call of its stream whose previous call's response carries
+// that request id, and whose prompt is the one it names, when exactly one
+// request and exactly one call carry those two ids. A compaction request
+// names no prompt, a retried request carries the same two ids twice, and the
+// first calls of two streams under one prompt carry the same two ids; none of
+// them is joined, and the entry says why. Nothing is joined by position or by
+// time.
+func (c *Conversation) providerBodies(landed []storage.LandedFile) (int, map[string][]sessionview.ProviderBody) {
+ type body struct {
+ id, step, join, role string
+ ref sessionflow.Ref
+ promptID, prevRequest, requestID, msgID string
+ }
+ const (
+ joinExact = "exact"
+ joinAmbiguous = "ambiguous"
+ joinUnresolved = "unresolved"
+ )
+ out := []body{}
+ seen := map[string]bool{}
+ for _, lf := range landed {
+ if !strings.HasPrefix(filepath.Base(lf.Path), string(sessiondata.KindProviderBody)+"-") {
+ continue
+ }
+ f, err := os.Open(lf.Path)
+ if err != nil {
+ continue
+ }
+ rd, err := sessiondata.NewReader(f)
+ if err != nil {
+ f.Close()
+ continue
+ }
+ for row := uint64(1); ; row++ {
+ rec, err := rd.Next()
+ if errors.Is(err, io.EOF) || err != nil {
+ break
+ }
+ m, err := providerbody.ManifestOf(rec)
+ if err != nil || seen[rec.ID] {
+ // A body landed twice by an interrupted pass is one body.
+ continue
+ }
+ seen[rec.ID] = true
+ out = append(out, body{
+ id: rec.ID, join: joinUnresolved, role: m.Role, ref: sessionflow.Ref{Seq: lf.Seq, Row: row},
+ promptID: m.Run, prevRequest: m.PreviousRequest, requestID: m.Request, msgID: m.Call,
+ })
+ }
+ f.Close()
+ }
+ byStep := map[string][]sessionview.ProviderBody{}
+ if len(out) == 0 {
+ return 0, byStep
+ }
+
+ // The calls, with their message id, their prompt and their stream, in
+ // the order their records landed, which is line order in a stream.
+ type call struct {
+ id, stream, msg, prompt string
+ at sessionflow.Ref
+ }
+ var calls []*call
+ var refs []*sessionflow.Ref
+ for _, n := range c.View.Nodes {
+ if n.Kind != model.KindLLMCall || n.Ref == nil {
+ continue
+ }
+ refs = append(refs, n.Ref)
+ if p := c.View.Nodes[n.Parent]; p != nil && p.Kind == model.KindRun && p.Ref != nil {
+ refs = append(refs, p.Ref)
+ }
+ }
+ recs := c.records(refs)
+ for _, n := range c.View.Nodes {
+ if n.Kind != model.KindLLMCall || n.Ref == nil {
+ continue
+ }
+ // A call whose record is gone stays in its stream with no message id,
+ // so the call after it has no previous response to name and is left
+ // unjoined rather than taken for the first of its stream.
+ k := &call{id: n.ID, stream: n.Stream, at: *n.Ref}
+ if rec := recs[[2]uint64{n.Ref.Seq, n.Ref.Row}]; rec != nil {
+ // A synthetic record sits in the stream like a response, but no
+ // provider was called: it has no request and no response, and
+ // it is not the call before the next one.
+ if slices.Contains(rec.Flags, "synthetic") {
+ continue
+ }
+ k.msg = rec.Call
+ }
+ if p := c.View.Nodes[n.Parent]; p != nil && p.Kind == model.KindRun && p.Ref != nil {
+ if run := recs[[2]uint64{p.Ref.Seq, p.Ref.Row}]; run != nil {
+ k.prompt = run.Run
+ }
+ }
+ calls = append(calls, k)
+ }
+ sort.Slice(calls, func(i, j int) bool {
+ if calls[i].stream != calls[j].stream {
+ return calls[i].stream < calls[j].stream
+ }
+ if calls[i].at.Seq != calls[j].at.Seq {
+ return calls[i].at.Seq < calls[j].at.Seq
+ }
+ return calls[i].at.Row < calls[j].at.Row
+ })
+
+ join := func(i int, step, quality string) {
+ out[i].step, out[i].join = step, quality
+ }
+ // Responses, by message id.
+ byMsg := map[string][]int{}
+ for i, b := range out {
+ if b.role == providerbody.RoleResponse && b.msgID != "" {
+ byMsg[b.msgID] = append(byMsg[b.msgID], i)
+ }
+ }
+ requestOf := map[string]string{} // call id -> the request id its response carries
+ responseOf := map[string]int{}
+ for _, k := range calls {
+ if k.msg == "" {
+ continue
+ }
+ switch hits := byMsg[k.msg]; len(hits) {
+ case 0:
+ case 1:
+ join(hits[0], k.id, joinExact)
+ requestOf[k.id], responseOf[k.id] = out[hits[0]].requestID, hits[0]
+ default:
+ for _, i := range hits {
+ out[i].join = joinAmbiguous
+ }
+ }
+ }
+ // Requests, by the request before them and their prompt.
+ type key struct{ prev, prompt string }
+ byKey := map[key][]int{}
+ for i, b := range out {
+ if b.role == providerbody.RoleRequest && b.promptID != "" {
+ k := key{b.prevRequest, b.promptID}
+ byKey[k] = append(byKey[k], i)
+ }
+ }
+ // The key each call's request would carry. A call whose previous call
+ // has no response carrying its request id has none, and neither has a
+ // call in a stream whose landed lines have a gap: a call may be missing
+ // between two that look consecutive.
+ gapped := map[string]bool{}
+ for _, k := range calls {
+ if _, done := gapped[k.stream]; !done {
+ gapped[k.stream] = streamHasGap(landed, k.stream)
+ }
+ }
+ callKey := map[string]key{}
+ callsByKey := map[key]int{}
+ for i, k := range calls {
+ if gapped[k.stream] {
+ continue
+ }
+ prev := ""
+ if i > 0 && calls[i-1].stream == k.stream {
+ var ok bool
+ if prev, ok = requestOf[calls[i-1].id]; !ok || prev == "" {
+ continue
+ }
+ }
+ if k.prompt == "" {
+ continue
+ }
+ callKey[k.id] = key{prev, k.prompt}
+ callsByKey[key{prev, k.prompt}]++
+ }
+ requestFor := map[string]int{}
+ for _, k := range calls {
+ ck, ok := callKey[k.id]
+ if !ok {
+ continue
+ }
+ hits := byKey[ck]
+ switch {
+ case len(hits) == 0:
+ case len(hits) == 1 && callsByKey[ck] == 1:
+ // One request and one call carry the key: nothing else could be
+ // this call's request, and this request no other call's.
+ if out[hits[0]].join == joinUnresolved {
+ join(hits[0], k.id, joinExact)
+ requestFor[k.id] = hits[0]
+ }
+ default:
+ for _, i := range hits {
+ if out[i].join == joinUnresolved {
+ out[i].join = joinAmbiguous
+ }
+ }
+ }
+ }
+ for _, k := range calls {
+ if i, ok := requestFor[k.id]; ok {
+ byStep[k.id] = append(byStep[k.id], sessionview.ProviderBody{Role: out[i].role, Ref: out[i].ref})
+ }
+ if i, ok := responseOf[k.id]; ok {
+ byStep[k.id] = append(byStep[k.id], sessionview.ProviderBody{Role: out[i].role, Ref: out[i].ref})
+ }
+ }
+ return len(out), byStep
+}
+
+// streamHasGap reports whether a stream's landed transcript lines skip a
+// line: the first is not line 1, or a line after a later one is missing.
+// A line landed twice is a repeat, not a gap. Only each record's ord is read.
+func streamHasGap(landed []storage.LandedFile, stream string) bool {
+ var prev uint64
+ for _, lf := range landed {
+ if lf.Stream != stream || !strings.HasPrefix(filepath.Base(lf.Path), string(sessiondata.KindTranscript)+"-") {
+ continue
+ }
+ f, err := os.Open(lf.Path)
+ if err != nil {
+ return true
+ }
+ rd, err := sessiondata.NewReader(f)
+ if err != nil {
+ f.Close()
+ return true
+ }
+ for {
+ line, err := rd.NextRaw()
+ if err != nil {
+ break
+ }
+ ord, ok := leadingOrd(line)
+ if !ok {
+ f.Close()
+ return true
+ }
+ if ord > prev+1 {
+ f.Close()
+ return true
+ }
+ if ord > prev {
+ prev = ord
+ }
+ }
+ f.Close()
+ }
+ return false
+}
+
+// leadingOrd reads a record line's ord, which the writer puts first.
+func leadingOrd(line []byte) (uint64, bool) {
+ const prefix = `{"ord":`
+ if !bytes.HasPrefix(line, []byte(prefix)) {
+ var r struct {
+ Ord uint64 `json:"ord"`
+ }
+ if json.Unmarshal(line, &r) != nil {
+ return 0, false
+ }
+ return r.Ord, true
+ }
+ var n uint64
+ i := len(prefix)
+ for ; i < len(line) && line[i] >= '0' && line[i] <= '9'; i++ {
+ n = n*10 + uint64(line[i]-'0')
+ }
+ return n, i > len(prefix)
+}
+
+// annotateProvider writes each call step's joined bodies onto the step.
+func annotateProvider(nodes []sessionview.Node, byStep map[string][]sessionview.ProviderBody) {
+ for i := range nodes {
+ if bodies := byStep[nodes[i].ID]; len(bodies) > 0 {
+ nodes[i].ProviderBodies = bodies
+ }
+ annotateProvider(nodes[i].Children, byStep)
+ }
+}
diff --git a/pkg/providerbody/providerbody.go b/pkg/providerbody/providerbody.go
new file mode 100644
index 0000000..0c33ed4
--- /dev/null
+++ b/pkg/providerbody/providerbody.go
@@ -0,0 +1,159 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package providerbody stores the bodies a runtime exchanged with its model
+// provider, one record per body, with what a session already holds taken out
+// and every body rebuildable byte for byte.
+//
+// Every call sends the whole message list of its chain again, so a body is
+// mostly bytes an earlier body of the same session already had. A record keeps
+// only what is new. Its last part is a manifest: the segments that, joined in
+// order, give the body back. A segment is literal bytes, a part of the record,
+// a piece an earlier record holds, or the front of an earlier body.
+//
+// A piece is a JSON value cut out of the body whole: each tool definition, and
+// every string of MinPiece bytes or more anywhere else. A body is compact JSON,
+// so every value is one contiguous run of its bytes. Pieces are named by their
+// SHA-256, and an earlier body by its record id and SHA-256, never by file and
+// row, because asz repack renumbers files and rows and carries records whole.
+package providerbody
+
+import (
+ "crypto/sha256"
+ "encoding/hex"
+)
+
+// Schema is the manifest's schema.
+const Schema = "provider_body/1"
+
+// Roles of a body.
+const (
+ RoleRequest = "request"
+ RoleResponse = "response"
+)
+
+const (
+ // MinPiece is the smallest string cut out as a piece. Measured on two
+ // captures of Claude Code 2.1.260, 1 KiB kept 25.9% and 23.2% of the raw
+ // bytes; 256 bytes kept 1 to 4 points less with three to four times as
+ // many parts.
+ MinPiece = 1 << 10
+ // MinCopy is the shortest front of an earlier body worth a copy segment.
+ // The first body of a chain shares only a few hundred bytes with anything
+ // earlier, and a copy of that saves nothing.
+ MinCopy = 4 << 10
+ // MaxDepth bounds a chain of copies. Rebuilding a body rebuilds its base
+ // first, so a body whose base is this deep lands without a copy and
+ // starts a new chain. Not measured on a long session yet.
+ MaxDepth = 32
+ // MaxBytes bounds the size a body may claim, and a rebuild stops as soon
+ // as it passes the size its record claims. A record is small, but its
+ // references can name large pieces many times over, so without a bound
+ // a damaged record could make a rebuild hold far more than any body.
+ MaxBytes = 256 << 20
+)
+
+// Manifest is the last part of a provider_body record. Its keys are the
+// model's words; a runtime's own names for these values stop at its adapter.
+type Manifest struct {
+ Schema string `json:"schema"`
+ Role string `json:"role"`
+ // Src is where the body came from, relative to the adapter's source
+ // root, as a header's src is.
+ Src string `json:"src"`
+ // SHA256 and Bytes describe the whole body, which the segments give back.
+ SHA256 string `json:"sha256"`
+ Bytes int `json:"bytes"`
+ // Depth counts the copies between this body and one with none.
+ Depth int `json:"depth"`
+ // Chain names the chain a request belongs to: the first sixteen
+ // hexadecimal characters of the SHA-256 of its first message, with the
+ // moving cache marker taken out. It is storage, not structure: it picks
+ // the body a copy comes from, and says nothing a join may rely on.
+ Chain string `json:"chain,omitempty"`
+ // Why says the body is kept whole, in part 0, and why it was not cut.
+ Why string `json:"why,omitempty"`
+
+ // The values the body carries that join it, as its adapter read them,
+ // so a reader can join a body without rebuilding it. Empty when the
+ // body does not carry one.
+ Model string `json:"model,omitempty"`
+ Session string `json:"session,omitempty"`
+ Run string `json:"run,omitempty"`
+ Call string `json:"call,omitempty"`
+ Request string `json:"request,omitempty"`
+ PreviousRequest string `json:"previous_request,omitempty"`
+
+ Segments []Segment `json:"segments"`
+}
+
+// Segment is one run of a body's bytes. Exactly one field is set.
+type Segment struct {
+ // Lit holds bytes that are no piece, as a string: the keys, the
+ // punctuation, the short values.
+ Lit string `json:"lit,omitempty"`
+ // Part is a part of this record: its data, or an unknown part's bytes.
+ Part *int `json:"part,omitempty"`
+ // Piece is the SHA-256 of a piece an earlier record of the session holds.
+ Piece string `json:"piece,omitempty"`
+ // Copy is the front of an earlier body of the session.
+ Copy *Copy `json:"copy,omitempty"`
+}
+
+// Copy names the first Len bytes of an earlier body.
+type Copy struct {
+ From string `json:"from"`
+ SHA256 string `json:"sha256"`
+ Len int `json:"len"`
+}
+
+// Keys are the values a body carries that join it to a session and a call,
+// in the model's words. An adapter reads them from its runtime's bodies.
+type Keys struct {
+ Model string
+ // Session is the session the body names.
+ Session string
+ // Run is the run, one prompt cycle, a request was sent in.
+ Run string
+ // Call is the provider call a response answers.
+ Call string
+ // Request is the provider's id for the request a response answers.
+ Request string
+ // PreviousRequest is the provider request id of the call before a
+ // request in its chain.
+ PreviousRequest string
+}
+
+// Body is one body for a session to cut: its record id, which is unique in the
+// session and stable however often it is read, its role, where it came from,
+// the keys its adapter read, and its bytes.
+type Body struct {
+ ID string
+ Role string
+ Src string
+ Keys Keys
+ Bytes []byte
+}
+
+// Digest is the SHA-256 of b in hexadecimal, as a manifest and a copy name
+// a body and a piece.
+func Digest(b []byte) string { return digest(b) }
+
+func digest(b []byte) string {
+ sum := sha256.Sum256(b)
+ return hex.EncodeToString(sum[:])
+}
diff --git a/pkg/providerbody/providerbody_test.go b/pkg/providerbody/providerbody_test.go
new file mode 100644
index 0000000..51c3a2b
--- /dev/null
+++ b/pkg/providerbody/providerbody_test.go
@@ -0,0 +1,501 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package providerbody
+
+import (
+ "bytes"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "strings"
+ "testing"
+
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+)
+
+// chain builds request bodies the way Claude Code 2.1.260 was measured to:
+// the whole message list again on every call, the cache marker on the
+// newest message only, a billing header naming the previous request and the
+// prompt, and the session in metadata.user_id.
+type chain struct {
+ session, prompt string
+ system []any
+ tools []any
+ msgs []map[string]any
+ prev string
+ reminder string
+}
+
+func newChain(session, prompt, systemText string, tools int) *chain {
+ c := &chain{session: session, prompt: prompt, reminder: "<system-reminder>" + strings.Repeat(" instructions", 400) + "</system-reminder>"}
+ c.system = []any{
+ map[string]any{"type": "text", "text": ""},
+ map[string]any{"type": "text", "text": systemText, "cache_control": map[string]any{"type": "ephemeral"}},
+ }
+ for i := 0; i < tools; i++ {
+ c.tools = append(c.tools, map[string]any{
+ "name": fmt.Sprintf("Tool%d", i), "description": strings.Repeat(fmt.Sprintf("tool %d does a thing. ", i), 80),
+ "input_schema": map[string]any{"type": "object"},
+ })
+ }
+ return c
+}
+
+func (c *chain) add(role string, text string) {
+ if len(c.msgs) == 0 {
+ // The first message carries the injected reminders beside the
+ // prompt, several blocks, so it keeps its shape on every call.
+ c.msgs = append(c.msgs, map[string]any{"role": role, "content": []any{
+ map[string]any{"type": "text", "text": c.reminder},
+ map[string]any{"type": "text", "text": text},
+ }})
+ return
+ }
+ for _, m := range c.msgs {
+ if blocks, ok := m["content"].([]any); ok && len(blocks) == 1 {
+ if b, ok := blocks[0].(map[string]any); ok && b["type"] == "text" {
+ m["content"] = b["text"]
+ }
+ }
+ }
+ c.msgs = append(c.msgs, map[string]any{"role": role, "content": []any{
+ map[string]any{"type": "text", "text": text, "cache_control": map[string]any{"type": "ephemeral"}},
+ }})
+}
+
+func (c *chain) request() []byte {
+ header := "x-anthropic-billing-header: cc_version=2.1.260; cch=00000;"
+ if c.prev != "" {
+ header += " cc_prev_req=" + c.prev + ";"
+ }
+ header += " cc_prompt_id=" + c.prompt + ";"
+ c.system[0].(map[string]any)["text"] = header
+ user, _ := json.Marshal(map[string]string{"device_id": "d", "session_id": c.session})
+ body := map[string]any{
+ "model": "claude-opus-5", "messages": c.msgs, "system": c.system, "tools": c.tools,
+ "metadata": map[string]any{"user_id": string(user)}, "max_tokens": 64000,
+ }
+ return compact(body)
+}
+
+// compact encodes the way JSON.stringify does: no spaces, no HTML escapes.
+// Go orders map keys alphabetically, which is still one fixed order.
+func compact(v any) []byte {
+ var buf bytes.Buffer
+ enc := json.NewEncoder(&buf)
+ enc.SetEscapeHTML(false)
+ _ = enc.Encode(v)
+ return bytes.TrimSuffix(buf.Bytes(), []byte{'\n'})
+}
+
+func response(id, text string) []byte {
+ return compact(map[string]any{"id": id, "type": "message", "role": "assistant", "model": "claude-opus-5",
+ "content": []any{map[string]any{"type": "text", "text": text}}, "stop_reason": "end_turn"})
+}
+
+// encode names a body the way these tests write them: an id and a role from
+// a file name, and no keys, since cutting and rebuilding read none.
+func encode(s *Session, file string, b []byte) (*sessiondata.Record, error) {
+ role := RoleRequest
+ if strings.HasSuffix(file, ".response.json") {
+ role = RoleResponse
+ }
+ return s.Encode(Body{ID: strings.TrimSuffix(file, ".json"), Role: role, Src: file, Bytes: b})
+}
+
+type body struct {
+ file string
+ b []byte
+}
+
+// interleaved is a main chain with a subagent whose calls land between two
+// main calls, then a compaction that starts the main chain again. With
+// sharedReminder the child's first message starts with the main chain's
+// reminder; measured subagents diverge within the first few hundred bytes.
+func interleaved(sharedReminder bool) []body {
+ var out []body
+ main := newChain("s1", "p1", strings.Repeat("You are an agent. ", 300), 12)
+ child := newChain("s1", "p1", strings.Repeat("You search a repository. ", 120), 6)
+ if !sharedReminder {
+ child.reminder = "<system-reminder>" + strings.Repeat(" as you answer", 300) + "</system-reminder>"
+ }
+ call := func(c *chain, req, resp, text string) {
+ c.add("user", text+strings.Repeat(" context", 200))
+ out = append(out, body{req + ".request.json", c.request()})
+ reply := "answer to " + text
+ out = append(out, body{resp + ".response.json", response("msg_"+resp, reply)})
+ c.add("assistant", reply)
+ c.prev = resp
+ }
+ call(main, "00000000-0000-4000-8000-000000000001", "req_1", "read the file")
+ call(main, "00000000-0000-4000-8000-000000000002", "req_2", "now the other file")
+ call(child, "00000000-0000-4000-8000-000000000003", "req_3", "find input_digest")
+ call(child, "00000000-0000-4000-8000-000000000004", "req_4", "grep further")
+ call(main, "00000000-0000-4000-8000-000000000005", "req_5", "what did it find")
+ call(child, "00000000-0000-4000-8000-000000000006", "req_6", "one more search")
+ call(main, "00000000-0000-4000-8000-000000000007", "req_7", "summarise")
+ main.msgs = nil
+ call(main, "00000000-0000-4000-8000-000000000008", "req_8", "This session is being continued")
+ call(main, "00000000-0000-4000-8000-000000000009", "req_9", "and then")
+ return out
+}
+
+func TestEncodedBodiesRebuildExactlyAfterLanding(t *testing.T) {
+ bodies := interleaved(false)
+ s := NewSession()
+ var recs []*sessiondata.Record
+ raw, landed := 0, 0
+ for _, b := range bodies {
+ rec, err := encode(s, b.file, b.b)
+ if err != nil {
+ t.Fatalf("%s: %v", b.file, err)
+ }
+ recs = append(recs, rec)
+ raw += len(b.b)
+ }
+
+ // Land every record through the writer, read the file back into a new
+ // session, and rebuild every body from what was read.
+ var buf bytes.Buffer
+ w, err := sessiondata.NewWriter(&buf, &sessiondata.Header{Kind: sessiondata.KindProviderBody, Session: "s1", Src: ".", Dialect: "test"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, rec := range recs {
+ if err := w.Write(rec); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if err := w.Close(); err != nil {
+ t.Fatal(err)
+ }
+ landed = buf.Len()
+ r, err := sessiondata.NewReader(&buf)
+ if err != nil {
+ t.Fatal(err)
+ }
+ fresh := NewSession()
+ for i := 0; ; i++ {
+ rec, err := r.Next()
+ if errors.Is(err, io.EOF) {
+ break
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := fresh.Add(rec); err != nil {
+ t.Fatalf("add %s: %v", rec.ID, err)
+ }
+ got, err := fresh.Body(rec.ID)
+ if err != nil {
+ t.Fatalf("rebuild %s: %v", rec.ID, err)
+ }
+ if !bytes.Equal(got, bodies[i].b) {
+ t.Fatalf("%s rebuilt to different bytes", rec.ID)
+ }
+ }
+ if landed*2 > raw {
+ t.Fatalf("landed %d bytes for %d raw; the repeats were not removed", landed, raw)
+ }
+}
+
+func TestACopyFollowsItsOwnChainAcrossInterleavedCalls(t *testing.T) {
+ // Call 5 is the main chain's, after two child calls; it copies call 2.
+ // Call 6 is the child's, after a main call; it copies call 4. The first
+ // call after compaction starts a new chain.
+ want := map[string]string{
+ "00000000-0000-4000-8000-000000000002.request": "00000000-0000-4000-8000-000000000001.request",
+ "00000000-0000-4000-8000-000000000004.request": "00000000-0000-4000-8000-000000000003.request",
+ "00000000-0000-4000-8000-000000000005.request": "00000000-0000-4000-8000-000000000002.request",
+ "00000000-0000-4000-8000-000000000006.request": "00000000-0000-4000-8000-000000000004.request",
+ "00000000-0000-4000-8000-000000000007.request": "00000000-0000-4000-8000-000000000005.request",
+ "00000000-0000-4000-8000-000000000009.request": "00000000-0000-4000-8000-000000000008.request",
+ }
+ for _, shared := range []bool{false, true} {
+ s := NewSession()
+ base := map[string]string{}
+ for _, b := range interleaved(shared) {
+ rec, err := encode(s, b.file, b.b)
+ if err != nil {
+ t.Fatal(err)
+ }
+ m, _ := ManifestOf(rec)
+ for _, seg := range m.Segments {
+ if seg.Copy != nil {
+ base[rec.ID] = seg.Copy.From
+ }
+ }
+ }
+ for id, from := range want {
+ if base[id] != from {
+ t.Errorf("shared reminder %v: %s copies from %q, want %q", shared, id, base[id], from)
+ }
+ }
+ // With its own reminder, a chain's first call shares too little with
+ // anything to copy. With the main chain's reminder, it may copy that
+ // front from the main chain, and the main chain keeps its own link.
+ if !shared {
+ for _, id := range []string{"00000000-0000-4000-8000-000000000001.request", "00000000-0000-4000-8000-000000000003.request"} {
+ if from, ok := base[id]; ok {
+ t.Errorf("%s is the first call of its chain and copies from %s", id, from)
+ }
+ }
+ }
+ }
+}
+
+func TestAChainOfCopiesStopsAtMaxDepth(t *testing.T) {
+ s := NewSession()
+ c := newChain("s1", "p1", strings.Repeat("system ", 2000), 4)
+ deepest := 0
+ for i := 0; i < MaxDepth+5; i++ {
+ c.add("user", fmt.Sprintf("turn %d %s", i, strings.Repeat("x", 3000)))
+ rec, err := encode(s, fmt.Sprintf("00000000-0000-4000-8000-%012d.request.json", i), c.request())
+ if err != nil {
+ t.Fatal(err)
+ }
+ m, _ := ManifestOf(rec)
+ if m.Depth > MaxDepth {
+ t.Fatalf("depth %d past %d", m.Depth, MaxDepth)
+ }
+ deepest = max(deepest, m.Depth)
+ c.add("assistant", "ok")
+ }
+ if deepest != MaxDepth {
+ t.Fatalf("the chain reached depth %d, want %d", deepest, MaxDepth)
+ }
+}
+
+func TestABodyThatCannotBeCutLandsWholeAndSaysWhy(t *testing.T) {
+ for name, raw := range map[string][]byte{
+ "not JSON": []byte(`{"model":"claude-opus-5","messages":[`),
+ "not UTF-8": []byte("{\"model\":\"\xff\"}"),
+ "not an object": []byte(`["a"]`),
+ "not compact": []byte("{\"tools\": [ {\"name\": \"a\"} ]}"),
+ } {
+ s := NewSession()
+ rec, err := encode(s, "00000000-0000-4000-8000-000000000001.request.json", raw)
+ if err != nil {
+ t.Fatalf("%s: %v", name, err)
+ }
+ m, _ := ManifestOf(rec)
+ got, err := s.Body(rec.ID)
+ if err != nil || !bytes.Equal(got, raw) {
+ t.Fatalf("%s: rebuilt %q, %v", name, got, err)
+ }
+ if name == "not compact" {
+ // Whitespace between tokens is no problem: literals keep it.
+ continue
+ }
+ if m.Why == "" || rec.Parts[0].Kind != sessiondata.PartUnknown {
+ t.Fatalf("%s: landed cut, want whole with a reason", name)
+ }
+ }
+}
+
+func TestAddRefusesAReferenceToSomethingNotYetHeld(t *testing.T) {
+ bodies := interleaved(false)
+ s := NewSession()
+ var recs []*sessiondata.Record
+ for _, b := range bodies[:6] {
+ rec, err := encode(s, b.file, b.b)
+ if err != nil {
+ t.Fatal(err)
+ }
+ recs = append(recs, rec)
+ }
+ fresh := NewSession()
+ // The third request copies an earlier one; added first, it names
+ // something the session does not hold.
+ if err := fresh.Add(recs[2]); err == nil {
+ t.Fatal("a record whose base was never added was accepted")
+ }
+ if err := fresh.Add(recs[0]); err != nil {
+ t.Fatal(err)
+ }
+ if err := fresh.Add(recs[0]); !errors.Is(err, ErrRepeat) {
+ t.Fatalf("the same record added twice: %v, want ErrRepeat", err)
+ }
+ if _, err := encode(s, bodies[0].file, append([]byte(nil), bodies[1].b...)); err == nil {
+ t.Fatal("a body read again under the same name with other bytes was accepted")
+ }
+}
+
+func TestPiecesAreToolsAndLongStrings(t *testing.T) {
+ long := strings.Repeat("a", MinPiece)
+ b := compact(map[string]any{
+ "messages": []any{map[string]any{"content": long, "role": "user"}, map[string]any{"content": "short", "role": "user"}},
+ "tools": []any{map[string]any{"description": long, "name": "A"}, map[string]any{"name": "B"}},
+ })
+ spans, _, err := pieces(b)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var got []string
+ for _, sp := range spans {
+ p := string(b[sp.a:sp.z])
+ if len(p) > 20 {
+ p = p[:20]
+ }
+ got = append(got, p)
+ }
+ want := []string{`"aaaaaaaaaaaaaaaaaaa`, `{"description":"aaaa`, `{"name":"B"}`}
+ if fmt.Sprint(got) != fmt.Sprint(want) {
+ t.Fatalf("pieces %q, want %q", got, want)
+ }
+}
+
+func TestAddRefusesNegativeSizesAndABadRepeat(t *testing.T) {
+ bodies := interleaved(false)
+ s := NewSession()
+ var recs []*sessiondata.Record
+ for _, b := range bodies[:3] {
+ rec, err := encode(s, b.file, b.b)
+ if err != nil {
+ t.Fatal(err)
+ }
+ recs = append(recs, rec)
+ }
+ withManifest := func(rec *sessiondata.Record, edit func(*Manifest)) *sessiondata.Record {
+ m, _ := ManifestOf(rec)
+ edit(m)
+ cp := *rec
+ cp.Parts = append(append([]sessiondata.Part(nil), rec.Parts[:len(rec.Parts)-1]...), manifestPart(m))
+ return &cp
+ }
+ fresh := NewSession()
+ if err := fresh.Add(withManifest(recs[0], func(m *Manifest) { m.Bytes = -1 })); err == nil {
+ t.Fatal("a record claiming -1 bytes was accepted")
+ }
+ if err := fresh.Add(recs[0]); err != nil {
+ t.Fatal(err)
+ }
+ if err := fresh.Add(recs[1]); err != nil {
+ t.Fatal(err)
+ }
+ negative := withManifest(recs[2], func(m *Manifest) {
+ for i := range m.Segments {
+ if m.Segments[i].Copy != nil {
+ m.Segments[i].Copy.Len = -1
+ }
+ }
+ })
+ if err := fresh.Add(negative); err == nil {
+ t.Fatal("a copy of -1 bytes was accepted")
+ }
+ // A repeat that claims the body's digest but does not rebuild to it.
+ lie := withManifest(recs[0], func(m *Manifest) { m.Segments = []Segment{{Lit: "[]"}} })
+ if err := fresh.Add(lie); err == nil || errors.Is(err, ErrRepeat) {
+ t.Fatalf("a repeat that does not rebuild was taken for a repeat: %v", err)
+ }
+}
+
+func TestABodyHandedOutCannotChangeWhatLaterBodiesCopy(t *testing.T) {
+ bodies := interleaved(false)
+ s := NewSession()
+ first, err := encode(s, bodies[0].file, bodies[0].b)
+ if err != nil {
+ t.Fatal(err)
+ }
+ got, err := s.Body(first.ID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for i := range got {
+ got[i] = 'x'
+ }
+ again, err := s.Body(first.ID)
+ if err != nil || !bytes.Equal(again, bodies[0].b) {
+ t.Fatal("changing a body a caller was given changed the session's copy")
+ }
+}
+
+func TestAddAndRebuildWithstandAnyClaim(t *testing.T) {
+ bodies := interleaved(false)
+ s := NewSession()
+ first, err := encode(s, bodies[0].file, bodies[0].b)
+ if err != nil {
+ t.Fatal(err)
+ }
+ withManifest := func(rec *sessiondata.Record, edit func(*Manifest)) *sessiondata.Record {
+ m, _ := ManifestOf(rec)
+ edit(m)
+ cp := *rec
+ cp.Parts = append(append([]sessiondata.Part(nil), rec.Parts[:len(rec.Parts)-1]...), manifestPart(m))
+ return &cp
+ }
+ fresh := NewSession()
+ // A size past MaxBytes is refused before anything is allocated.
+ huge := withManifest(first, func(m *Manifest) { m.Bytes = 1 << 62; m.Segments = []Segment{{Lit: "{}"}} })
+ huge.ID = "huge.request"
+ if err := fresh.Add(huge); err == nil {
+ t.Fatal("a body claiming 2^62 bytes was accepted")
+ }
+ // A repeat with a bogus copy is refused, not taken for a repeat.
+ if err := fresh.Add(first); err != nil {
+ t.Fatal(err)
+ }
+ bogus := withManifest(first, func(m *Manifest) {
+ m.Depth = 1
+ m.Segments = []Segment{{Copy: &Copy{From: first.ID, SHA256: "bogus", Len: 2}}}
+ })
+ if err := fresh.Add(bogus); err == nil || errors.Is(err, ErrRepeat) {
+ t.Fatalf("a repeat with a copy of the wrong digest: %v", err)
+ }
+ deep := withManifest(first, func(m *Manifest) { m.Depth = 3 })
+ if err := fresh.Add(deep); err == nil || errors.Is(err, ErrRepeat) {
+ t.Fatalf("a record that copies nothing and claims depth 3: %v", err)
+ }
+}
+
+func TestARebuildStopsAtItsClaim(t *testing.T) {
+ bodies := interleaved(false)
+ s := NewSession()
+ first, err := encode(s, bodies[0].file, bodies[0].b)
+ if err != nil {
+ t.Fatal(err)
+ }
+ m, _ := ManifestOf(first)
+ var piece string
+ for _, seg := range m.Segments {
+ if seg.Part != nil {
+ piece = digest(first.Parts[*seg.Part].Data)
+ break
+ }
+ }
+ segs := make([]Segment, 10000)
+ for i := range segs {
+ segs[i] = Segment{Piece: piece}
+ }
+ bomb := &Manifest{Schema: Schema, Role: RoleRequest, Src: "x", SHA256: digest([]byte("{}")), Bytes: 2, Segments: segs}
+ rec := &sessiondata.Record{Ord: 1, ID: "bomb.request", Parts: []sessiondata.Part{manifestPart(bomb)}}
+ fresh := NewSession()
+ if err := fresh.Add(first); err != nil {
+ t.Fatal(err)
+ }
+ if err := fresh.Add(rec); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := fresh.Body("bomb.request"); err == nil {
+ t.Fatal("a record naming a piece 10,000 times rebuilt")
+ }
+ if n := RecordBytes(first); n <= 0 {
+ t.Fatalf("RecordBytes is %d", n)
+ }
+}
diff --git a/pkg/providerbody/scan.go b/pkg/providerbody/scan.go
new file mode 100644
index 0000000..7111527
--- /dev/null
+++ b/pkg/providerbody/scan.go
@@ -0,0 +1,194 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package providerbody
+
+import (
+ "errors"
+)
+
+// span is a half-open run [a, z) of a body's bytes.
+type span struct{ a, z int }
+
+// pieces finds the pieces of a body that is one valid JSON object: each
+// element of the top-level "tools" array, and every string token of MinPiece
+// bytes or more outside those elements. Spans come back in order and never
+// overlap. It also returns the span of the first element of the top-level
+// "messages" array, empty when there is none. The caller has checked that
+// the body is valid JSON.
+func pieces(b []byte) ([]span, span, error) {
+ s := &scanner{b: b}
+ i := s.ws(0)
+ if i >= len(b) || b[i] != '{' {
+ return nil, span{}, errors.New("the body is not a JSON object")
+ }
+ if _, err := s.value(i, true, false); err != nil {
+ return nil, span{}, err
+ }
+ return s.spans, s.first, nil
+}
+
+type scanner struct {
+ b []byte
+ spans []span
+ first span
+}
+
+func (s *scanner) ws(i int) int {
+ for i < len(s.b) {
+ switch s.b[i] {
+ case ' ', '\t', '\n', '\r':
+ i++
+ default:
+ return i
+ }
+ }
+ return i
+}
+
+var errShape = errors.New("the body ended inside a value")
+
+// value reads the value at i and returns the index after it. top is true for
+// the body's own object; whole is true when the value is a piece in itself
+// and nothing inside it is cut.
+func (s *scanner) value(i int, top, whole bool) (int, error) {
+ if i >= len(s.b) {
+ return 0, errShape
+ }
+ switch s.b[i] {
+ case '{':
+ return s.object(i, top, whole)
+ case '[':
+ return s.array(i, false, whole)
+ case '"':
+ z, err := s.str(i)
+ if err != nil {
+ return 0, err
+ }
+ if !whole && z-i >= MinPiece {
+ s.spans = append(s.spans, span{i, z})
+ }
+ return z, nil
+ default:
+ // A number, true, false or null: runs to the next delimiter.
+ z := i
+ for z < len(s.b) {
+ c := s.b[z]
+ if c == ',' || c == '}' || c == ']' || c == ' ' || c == '\t' || c == '\n' || c == '\r' {
+ break
+ }
+ z++
+ }
+ return z, nil
+ }
+}
+
+func (s *scanner) object(i int, top, whole bool) (int, error) {
+ i = s.ws(i + 1)
+ if i < len(s.b) && s.b[i] == '}' {
+ return i + 1, nil
+ }
+ for {
+ if i >= len(s.b) || s.b[i] != '"' {
+ return 0, errShape
+ }
+ kz, err := s.str(i)
+ if err != nil {
+ return 0, err
+ }
+ tools := top && string(s.b[i:kz]) == `"tools"`
+ messages := top && string(s.b[i:kz]) == `"messages"`
+ i = s.ws(kz)
+ if i >= len(s.b) || s.b[i] != ':' {
+ return 0, errShape
+ }
+ i = s.ws(i + 1)
+ switch {
+ case tools && i < len(s.b) && s.b[i] == '[':
+ i, err = s.array(i, true, whole)
+ case messages && i < len(s.b) && s.b[i] == '[':
+ if j := s.ws(i + 1); j < len(s.b) && s.b[j] != ']' {
+ z, ferr := s.value(j, false, true)
+ if ferr != nil {
+ return 0, ferr
+ }
+ s.first = span{j, z}
+ }
+ i, err = s.array(i, false, whole)
+ default:
+ i, err = s.value(i, false, whole)
+ }
+ if err != nil {
+ return 0, err
+ }
+ i = s.ws(i)
+ if i >= len(s.b) {
+ return 0, errShape
+ }
+ switch s.b[i] {
+ case ',':
+ i = s.ws(i + 1)
+ case '}':
+ return i + 1, nil
+ default:
+ return 0, errShape
+ }
+ }
+}
+
+// array reads an array. When each is true, every element is a piece whole.
+func (s *scanner) array(i int, each, whole bool) (int, error) {
+ i = s.ws(i + 1)
+ if i < len(s.b) && s.b[i] == ']' {
+ return i + 1, nil
+ }
+ for {
+ start := i
+ z, err := s.value(i, false, whole || each)
+ if err != nil {
+ return 0, err
+ }
+ if each && !whole {
+ s.spans = append(s.spans, span{start, z})
+ }
+ i = s.ws(z)
+ if i >= len(s.b) {
+ return 0, errShape
+ }
+ switch s.b[i] {
+ case ',':
+ i = s.ws(i + 1)
+ case ']':
+ return i + 1, nil
+ default:
+ return 0, errShape
+ }
+ }
+}
+
+// str returns the index after the string token that starts at i.
+func (s *scanner) str(i int) (int, error) {
+ for j := i + 1; j < len(s.b); j++ {
+ switch s.b[j] {
+ case '\\':
+ j++
+ case '"':
+ return j + 1, nil
+ }
+ }
+ return 0, errShape
+}
diff --git a/pkg/providerbody/session.go b/pkg/providerbody/session.go
new file mode 100644
index 0000000..927c663
--- /dev/null
+++ b/pkg/providerbody/session.go
@@ -0,0 +1,564 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package providerbody
+
+import (
+ "bytes"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "unicode/utf8"
+
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+)
+
+// ErrRepeat reports a body the session already holds with the same bytes: a
+// file landed again after an interrupted pass.
+var ErrRepeat = errors.New("providerbody: the session already holds this body")
+
+// maxTips bounds how many chain tips a new body is compared against. A tip is
+// one per chain that is still growing: the main stream and each child running
+// at the time.
+const maxTips = 64
+
+// otherTips is how many tips of other chains a body is also compared with,
+// for a chain whose first message changed shape between two calls.
+const otherTips = 8
+
+// cacheLimit bounds the rebuilt bodies kept in memory.
+const cacheLimit = 64 << 20
+
+// Session holds the provider_body records of one session, in landing order,
+// and rebuilds any of them. Records must be added in the order they landed,
+// because a record may refer only to what came before it.
+type Session struct {
+ records map[string]*held
+ pieces map[string]pieceAt
+ // tips holds the latest request body of each chain, by chain, and
+ // tipOrder the chains from the one updated longest ago.
+ tips map[string]string
+ tipOrder []string
+
+ cache map[string][]byte
+ cacheOrder []string
+ cacheBytes int
+}
+
+type held struct {
+ rec *sessiondata.Record
+ m *Manifest
+}
+
+type pieceAt struct {
+ id string
+ part int
+}
+
+// NewSession returns an empty session.
+func NewSession() *Session {
+ return &Session{records: map[string]*held{}, pieces: map[string]pieceAt{}, tips: map[string]string{}, cache: map[string][]byte{}}
+}
+
+// ManifestOf returns the manifest a provider_body record carries as its last part.
+func ManifestOf(rec *sessiondata.Record) (*Manifest, error) {
+ if len(rec.Parts) == 0 {
+ return nil, errors.New("providerbody: the record has no parts")
+ }
+ last := rec.Parts[len(rec.Parts)-1]
+ var m Manifest
+ if last.Kind != sessiondata.PartData || json.Unmarshal(last.Data, &m) != nil || m.Schema != Schema {
+ return nil, fmt.Errorf("providerbody: the last part of %s is not a %s manifest", rec.ID, Schema)
+ }
+ return &m, nil
+}
+
+// Holds reports the SHA-256 of the body the session holds under id.
+func (s *Session) Holds(id string) (string, bool) {
+ h, ok := s.records[id]
+ if !ok {
+ return "", false
+ }
+ return h.m.SHA256, true
+}
+
+// Manifest returns the manifest of the body held under id.
+func (s *Session) Manifest(id string) (*Manifest, bool) {
+ h, ok := s.records[id]
+ if !ok {
+ return nil, false
+ }
+ return h.m, true
+}
+
+// Add takes a record that landed after every record added before it. It
+// checks that each reference points at something the session already holds,
+// and makes the record's pieces and body available to later records. A
+// record whose id the session holds with the same digest returns ErrRepeat.
+func (s *Session) Add(rec *sessiondata.Record) error {
+ m, err := ManifestOf(rec)
+ if err != nil {
+ return err
+ }
+ if rec.ID == "" {
+ return errors.New("providerbody: a record without an id")
+ }
+ if m.Bytes < 0 || m.Bytes > MaxBytes || m.Depth < 0 || m.Depth > MaxDepth {
+ return fmt.Errorf("providerbody: %s claims %d bytes at depth %d", rec.ID, m.Bytes, m.Depth)
+ }
+ // References are checked first, so a record landed again is held to the
+ // same rules as the first copy before it may count as a repeat.
+ own := len(rec.Parts) - 1
+ copies := 0
+ for _, seg := range m.Segments {
+ switch {
+ case seg.Part != nil:
+ if *seg.Part < 0 || *seg.Part >= own {
+ return fmt.Errorf("providerbody: %s names part %d of %d", rec.ID, *seg.Part, own)
+ }
+ case seg.Piece != "":
+ if _, ok := s.pieces[seg.Piece]; !ok {
+ return fmt.Errorf("providerbody: %s names piece %s, which no earlier record holds", rec.ID, seg.Piece)
+ }
+ case seg.Copy != nil:
+ copies++
+ base, ok := s.records[seg.Copy.From]
+ switch {
+ case !ok:
+ return fmt.Errorf("providerbody: %s copies from %s, which no earlier record holds", rec.ID, seg.Copy.From)
+ case base.m.SHA256 != seg.Copy.SHA256 || seg.Copy.Len < 0 || seg.Copy.Len > base.m.Bytes:
+ return fmt.Errorf("providerbody: %s copies from %s with a digest or length it does not have", rec.ID, seg.Copy.From)
+ case m.Depth != base.m.Depth+1 || m.Depth > MaxDepth:
+ return fmt.Errorf("providerbody: %s claims depth %d over a base of depth %d", rec.ID, m.Depth, base.m.Depth)
+ }
+ }
+ }
+ if copies == 0 && m.Depth != 0 {
+ return fmt.Errorf("providerbody: %s claims depth %d and copies nothing", rec.ID, m.Depth)
+ }
+ if h, ok := s.records[rec.ID]; ok {
+ if h.m.SHA256 != m.SHA256 {
+ return fmt.Errorf("providerbody: %s is held with digest %s and landed again with %s", rec.ID, h.m.SHA256, m.SHA256)
+ }
+ // A repeat is a repeat only if it rebuilds to what it claims; a
+ // record that says so and does not is damage.
+ if _, err := s.rebuild(rec, m); err != nil {
+ return err
+ }
+ return ErrRepeat
+ }
+ s.records[rec.ID] = &held{rec: rec, m: m}
+ for i, p := range rec.Parts[:own] {
+ if p.Kind != sessiondata.PartData {
+ continue
+ }
+ if d := digest(p.Data); !hasPiece(s.pieces, d) {
+ s.pieces[d] = pieceAt{id: rec.ID, part: i}
+ }
+ }
+ if m.Role == RoleRequest && m.Chain != "" {
+ s.setTip(m.Chain, rec.ID)
+ }
+ return nil
+}
+
+func hasPiece(m map[string]pieceAt, d string) bool { _, ok := m[d]; return ok }
+
+// setTip makes id the latest body of its chain, and forgets the chain
+// updated longest ago once more than maxTips are held.
+func (s *Session) setTip(chain, id string) {
+ if _, ok := s.tips[chain]; ok {
+ for i, c := range s.tipOrder {
+ if c == chain {
+ s.tipOrder = append(s.tipOrder[:i], s.tipOrder[i+1:]...)
+ break
+ }
+ }
+ }
+ s.tips[chain] = id
+ s.tipOrder = append(s.tipOrder, chain)
+ if len(s.tipOrder) > maxTips {
+ delete(s.tips, s.tipOrder[0])
+ s.tipOrder = s.tipOrder[1:]
+ }
+}
+
+// Body rebuilds the body held under id and checks its digest. The bytes
+// are the caller's own.
+func (s *Session) Body(id string) ([]byte, error) {
+ b, err := s.body(id)
+ if err != nil {
+ return nil, err
+ }
+ return append([]byte(nil), b...), nil
+}
+
+// body is Body without the copy, for this package only: a later body's copy
+// must read the base as it was rebuilt, and nothing here changes it.
+func (s *Session) body(id string) ([]byte, error) {
+ if b, ok := s.cache[id]; ok {
+ return b, nil
+ }
+ h, ok := s.records[id]
+ if !ok {
+ return nil, fmt.Errorf("providerbody: no body %s", id)
+ }
+ b, err := s.rebuild(h.rec, h.m)
+ if err != nil {
+ return nil, err
+ }
+ s.remember(id, b)
+ return b, nil
+}
+
+func (s *Session) remember(id string, b []byte) {
+ s.cache[id] = b
+ s.cacheOrder = append(s.cacheOrder, id)
+ s.cacheBytes += len(b)
+ for s.cacheBytes > cacheLimit && len(s.cacheOrder) > 1 {
+ old := s.cacheOrder[0]
+ s.cacheOrder = s.cacheOrder[1:]
+ s.cacheBytes -= len(s.cache[old])
+ delete(s.cache, old)
+ }
+}
+
+// rebuild joins a record's segments. The record need not be added yet, so a
+// record can be checked before it lands.
+func (s *Session) rebuild(rec *sessiondata.Record, m *Manifest) ([]byte, error) {
+ if m.Bytes < 0 || m.Bytes > MaxBytes {
+ return nil, fmt.Errorf("providerbody: %s claims %d bytes", rec.ID, m.Bytes)
+ }
+ over := func(int) error {
+ return fmt.Errorf("providerbody: %s rebuilds past the %d bytes it claims", rec.ID, m.Bytes)
+ }
+ // The claimed size is only a claim until the digest agrees, so it sets
+ // no more than a first capacity.
+ out := make([]byte, 0, min(m.Bytes, 1<<20))
+ for _, seg := range m.Segments {
+ switch {
+ case seg.Lit != "":
+ if len(out)+len(seg.Lit) > m.Bytes {
+ return nil, over(len(seg.Lit))
+ }
+ out = append(out, seg.Lit...)
+ case seg.Part != nil:
+ if *seg.Part < 0 || *seg.Part >= len(rec.Parts)-1 {
+ return nil, fmt.Errorf("providerbody: %s names part %d it does not have", rec.ID, *seg.Part)
+ }
+ b, err := partBytes(rec.Parts[*seg.Part])
+ if err != nil {
+ return nil, err
+ }
+ if len(out)+len(b) > m.Bytes {
+ return nil, over(len(b))
+ }
+ out = append(out, b...)
+ case seg.Piece != "":
+ at, ok := s.pieces[seg.Piece]
+ if !ok {
+ return nil, fmt.Errorf("providerbody: %s names piece %s, which no earlier record holds", rec.ID, seg.Piece)
+ }
+ b, err := partBytes(s.records[at.id].rec.Parts[at.part])
+ if err != nil {
+ return nil, err
+ }
+ if len(out)+len(b) > m.Bytes {
+ return nil, over(len(b))
+ }
+ out = append(out, b...)
+ case seg.Copy != nil:
+ base, err := s.body(seg.Copy.From)
+ if err != nil {
+ return nil, err
+ }
+ if seg.Copy.Len < 0 || seg.Copy.Len > len(base) {
+ return nil, fmt.Errorf("providerbody: %s copies %d bytes of a %d byte body", rec.ID, seg.Copy.Len, len(base))
+ }
+ if len(out)+seg.Copy.Len > m.Bytes {
+ return nil, over(seg.Copy.Len)
+ }
+ out = append(out, base[:seg.Copy.Len]...)
+ }
+ }
+ if len(out) != m.Bytes || digest(out) != m.SHA256 {
+ return nil, fmt.Errorf("providerbody: %s rebuilds to %d bytes that do not match its digest", rec.ID, len(out))
+ }
+ return out, nil
+}
+
+func partBytes(p sessiondata.Part) ([]byte, error) {
+ if p.Kind == sessiondata.PartUnknown {
+ return p.Raw()
+ }
+ return p.Data, nil
+}
+
+// Encode cuts a body against what the session holds, checks that the record
+// rebuilds the body after a round trip through the Session Data writer and
+// reader, and adds it. A body that cannot be cut, or does not come back whole,
+// lands whole in one unknown part, and the manifest says why. A body the
+// session already holds returns ErrRepeat.
+func (s *Session) Encode(b Body) (*sessiondata.Record, error) {
+ body := b.Bytes
+ sum := digest(body)
+ if h, ok := s.records[b.ID]; ok {
+ if h.m.SHA256 == sum {
+ return nil, ErrRepeat
+ }
+ return nil, fmt.Errorf("providerbody: %s is held with digest %s and read again with %s", b.ID, h.m.SHA256, sum)
+ }
+ k := b.Keys
+ m := &Manifest{
+ Schema: Schema, Role: b.Role, Src: b.Src, SHA256: sum, Bytes: len(body),
+ Model: k.Model, Session: k.Session, Run: k.Run, Call: k.Call, Request: k.Request, PreviousRequest: k.PreviousRequest,
+ }
+ rec := &sessiondata.Record{Ord: 1, Sha: sum[:12], Bytes: len(body), ID: b.ID, Model: k.Model}
+ switch b.Role {
+ case RoleRequest:
+ rec.Run = k.Run
+ case RoleResponse:
+ rec.Call = k.Call
+ }
+
+ parts, why := s.cut(body, m)
+ if why == "" {
+ rec.Parts = append(parts, manifestPart(m))
+ if err := s.roundTrip(rec, m); err != nil {
+ why = err.Error()
+ }
+ }
+ if why != "" {
+ whole := sessiondata.Part{Kind: sessiondata.PartUnknown, Text: why, State: "available", Bytes: len(body)}
+ whole.SetRaw(body)
+ zero := 0
+ m.Depth, m.Why, m.Segments = 0, why, []Segment{{Part: &zero}}
+ rec.Parts = []sessiondata.Part{whole, manifestPart(m)}
+ if err := s.roundTrip(rec, m); err != nil {
+ return nil, err
+ }
+ }
+ if err := s.Add(rec); err != nil {
+ return nil, err
+ }
+ return rec, nil
+}
+
+// cut builds the parts and segments of a body, or says why it cannot.
+func (s *Session) cut(body []byte, m *Manifest) ([]sessiondata.Part, string) {
+ if !utf8.Valid(body) {
+ return nil, "the body is not valid UTF-8"
+ }
+ if !json.Valid(body) {
+ return nil, "the body is not valid JSON"
+ }
+ spans, first, err := pieces(body)
+ if err != nil {
+ return nil, err.Error()
+ }
+
+ pos := 0
+ if m.Role == RoleRequest && first.z > first.a {
+ m.Chain = chainOf(body[first.a:first.z])
+ if c := s.base(body, spans, m.Chain); c != nil {
+ m.Segments = append(m.Segments, Segment{Copy: c})
+ m.Depth = s.records[c.From].m.Depth + 1
+ pos = c.Len
+ }
+ }
+ var parts []sessiondata.Part
+ local := map[string]int{}
+ for _, sp := range spans {
+ if sp.a < pos {
+ continue
+ }
+ if sp.a > pos {
+ m.Segments = append(m.Segments, Segment{Lit: string(body[pos:sp.a])})
+ }
+ piece := body[sp.a:sp.z]
+ d := digest(piece)
+ switch i, ok := local[d]; {
+ case ok:
+ m.Segments = append(m.Segments, Segment{Part: &i})
+ case hasPiece(s.pieces, d):
+ m.Segments = append(m.Segments, Segment{Piece: d})
+ default:
+ i := len(parts)
+ local[d] = i
+ parts = append(parts, sessiondata.Part{
+ Kind: sessiondata.PartData, Data: json.RawMessage(append([]byte(nil), piece...)),
+ State: "available", Bytes: len(piece),
+ })
+ m.Segments = append(m.Segments, Segment{Part: &i})
+ }
+ pos = sp.z
+ }
+ if pos < len(body) {
+ m.Segments = append(m.Segments, Segment{Lit: string(body[pos:])})
+ }
+ return parts, ""
+}
+
+// base picks the body to copy the front of: the tip of the body's own chain,
+// and the most recently updated tips of other chains, whichever shares the
+// longest front. It says how much to copy. The copy ends before any piece it
+// would cut in two, and on a character boundary, since the rest continues
+// as a literal string.
+func (s *Session) base(body []byte, spans []span, chain string) *Copy {
+ var candidates []string
+ if id, ok := s.tips[chain]; ok {
+ candidates = append(candidates, id)
+ }
+ for i := len(s.tipOrder) - 1; i >= 0 && len(candidates) < otherTips+1; i-- {
+ if c := s.tipOrder[i]; c != chain {
+ candidates = append(candidates, s.tips[c])
+ }
+ }
+ var best *Copy
+ for _, id := range candidates {
+ h := s.records[id]
+ if h.m.Depth+1 > MaxDepth {
+ continue
+ }
+ b, err := s.body(id)
+ if err != nil {
+ continue
+ }
+ n := commonPrefix(b, body)
+ if best == nil || n > best.Len {
+ best = &Copy{From: id, SHA256: h.m.SHA256, Len: n}
+ }
+ }
+ if best == nil {
+ return nil
+ }
+ for _, sp := range spans {
+ if sp.z > best.Len {
+ if sp.a < best.Len {
+ best.Len = sp.a
+ }
+ break
+ }
+ }
+ for best.Len > 0 && best.Len < len(body) && !utf8.RuneStart(body[best.Len]) {
+ best.Len--
+ }
+ if best.Len < MinCopy {
+ return nil
+ }
+ return best
+}
+
+// chainOf names the chain a request belongs to by its first message. That
+// message is the same on every call of a chain, apart from the cache marker,
+// which moves to the newest message, and a one-block text list becoming a
+// plain string once the marker has left it. Both are taken out first. A
+// compaction starts a new first message, and so a new chain.
+func chainOf(first []byte) string {
+ var msg map[string]any
+ if json.Unmarshal(first, &msg) != nil {
+ return digest(first)[:16]
+ }
+ if blocks, ok := msg["content"].([]any); ok {
+ for _, b := range blocks {
+ if block, ok := b.(map[string]any); ok {
+ delete(block, "cache_control")
+ }
+ }
+ if len(blocks) == 1 {
+ if block, ok := blocks[0].(map[string]any); ok && block["type"] == "text" && len(block) == 2 {
+ msg["content"] = block["text"]
+ }
+ }
+ }
+ b, _ := json.Marshal(msg)
+ return digest(b)[:16]
+}
+
+func commonPrefix(a, b []byte) int {
+ n := min(len(a), len(b))
+ for i := 0; i < n; i++ {
+ if a[i] != b[i] {
+ return i
+ }
+ }
+ return n
+}
+
+func manifestPart(m *Manifest) sessiondata.Part {
+ var buf bytes.Buffer
+ enc := json.NewEncoder(&buf)
+ enc.SetEscapeHTML(false)
+ _ = enc.Encode(m)
+ data := bytes.TrimSuffix(buf.Bytes(), []byte{'\n'})
+ return sessiondata.Part{Kind: sessiondata.PartData, Data: data, State: "available", Bytes: len(data)}
+}
+
+// roundTrip writes the record with the Session Data writer, reads it back,
+// and rebuilds the body from what was read. Only that proves the bytes a
+// reader will get, since the writer compacts every data part.
+func (s *Session) roundTrip(rec *sessiondata.Record, m *Manifest) error {
+ var buf bytes.Buffer
+ hdr := &sessiondata.Header{Kind: sessiondata.KindProviderBody, Session: "check", Src: ".", Dialect: "check"}
+ w, err := sessiondata.NewWriter(&buf, hdr)
+ if err != nil {
+ return err
+ }
+ if err := w.Write(rec); err != nil {
+ return err
+ }
+ if err := w.Close(); err != nil {
+ return err
+ }
+ r, err := sessiondata.NewReader(&buf)
+ if err != nil {
+ return err
+ }
+ back, err := r.Next()
+ if err != nil {
+ return err
+ }
+ bm, err := ManifestOf(back)
+ if err != nil {
+ return err
+ }
+ if _, err := s.rebuild(back, bm); err != nil {
+ return fmt.Errorf("the record does not rebuild the body after a round trip: %w", err)
+ }
+ _ = m
+ return nil
+}
+
+// RecordBytes is the size a record takes as a line of a landed file, its
+// newline included. A writer cutting files at a byte budget adds these up.
+func RecordBytes(rec *sessiondata.Record) int64 {
+ var buf bytes.Buffer
+ w, err := sessiondata.NewWriter(&buf, &sessiondata.Header{Kind: sessiondata.KindProviderBody, Session: "s", Src: ".", Dialect: "d"})
+ if err != nil {
+ return 0
+ }
+ start := int64(buf.Len())
+ _ = w.Write(rec)
+ _ = w.Close()
+ // The closing line is measured with the record and taken off with it.
+ end := int64(buf.Len()) - int64(len(`{"t":"end","records":1,"digest":""}`)+64+1)
+ return end - start
+}
+
+// FileOverhead is what a landed provider_body file holds besides its
+// records, at most: its header and its closing line.
+const FileOverhead = 512
diff --git a/pkg/sessionview/sessionview.go b/pkg/sessionview/sessionview.go
index 7120e8c..774f2de 100644
--- a/pkg/sessionview/sessionview.go
+++ b/pkg/sessionview/sessionview.go
@@ -84,6 +84,16 @@ type Conversation struct {
WorkspaceChanges []WorkspaceChange `json:"workspace_changes"`
}
+// ProviderBody names one landed provider body joined to a call: whether it
+// is the request or the response, and the landed record it rebuilds from.
+// The body is not in the document. A reader loads it: the record Ref names,
+// and the session's provider_body files with that sequence or lower, which
+// hold everything it refers to. See Session Data, Provider bodies.
+type ProviderBody struct {
+ Role string `json:"role"`
+ Ref sessionflow.Ref `json:"ref"`
+}
+
// WorkspaceChange is one change record with the step it joins to and where
// it was read from. The record's own fields follow, as changes/1 lists
// them; its captured_by says who observed it. Step is empty when no step
@@ -119,6 +129,11 @@ type Summary struct {
Unresolved int `json:"unresolved"`
// Changes counts the workspace change records.
Changes int `json:"changes"`
+ // ProviderBodies counts the session's landed provider bodies, and
+ // CapturedPrompts the calls whose request is captured: the calls that
+ // list a request.
+ ProviderBodies int `json:"provider_bodies"`
+ CapturedPrompts int `json:"captured_prompts"`
// From and To are when the session began and its last activity, from
// the session node.
@@ -266,6 +281,9 @@ type Node struct {
// Changes names the workspace change records joined to
// this step, by id, in the order WorkspaceChanges lists them.
Changes []string `json:"changes,omitempty"`
+ // ProviderBodies names the provider bodies joined to a call step: its
+ // request, then its response, each only when joined exactly.
+ ProviderBodies []ProviderBody `json:"provider_bodies,omitempty"`
Children []Node `json:"children,omitempty"`
Edges []Edge `json:"edges,omitempty"`
diff --git a/tests/scenarios/provider-bodies-errors.expect.yaml b/tests/scenarios/provider-bodies-errors.expect.yaml
new file mode 100644
index 0000000..02c8b5a
--- /dev/null
+++ b/tests/scenarios/provider-bodies-errors.expect.yaml
@@ -0,0 +1,25 @@
+# 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.
+
+# Two real calls with their request and response each; the synthetic error
+# in between lists none.
+checkpoints:
+ final:
+ view:
+ state: verified
+ provider_bodies: 4
+ captured_prompts: 2
diff --git a/tests/scenarios/provider-bodies-errors.yaml b/tests/scenarios/provider-bodies-errors.yaml
new file mode 100644
index 0000000..9f3ed5d
--- /dev/null
+++ b/tests/scenarios/provider-bodies-errors.yaml
@@ -0,0 +1,27 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# A synthetic API error between two calls. It sits in the stream as a call,
+# but nothing was sent to the provider, so it has no bodies, and the call after
+# it keeps its own request.
+title: an error between two calls
+provider_bodies: true
+steps:
+ - input: summarise the file
+ - call: {text: Reading it.}
+ - error: "API Error: 529 overloaded"
+ - call: {text: It is a configuration file.}
diff --git a/tests/scenarios/provider-bodies-late-child.expect.yaml b/tests/scenarios/provider-bodies-late-child.expect.yaml
new file mode 100644
index 0000000..74c24f7
--- /dev/null
+++ b/tests/scenarios/provider-bodies-late-child.expect.yaml
@@ -0,0 +1,30 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Three calls, each with its request and response, all landed and joined
+# across the two passes.
+checkpoints:
+ launched:
+ kinds: {llm.call: 2}
+ final:
+ kinds: {llm.call: 3}
+ view:
+ state: verified
+ provider_bodies: 6
+ captured_prompts: 3
+push:
+ kinds: [transcript, agent_meta, provider_body, round]
diff --git a/tests/scenarios/provider-bodies-late-child.yaml b/tests/scenarios/provider-bodies-late-child.yaml
new file mode 100644
index 0000000..e4ab6d0
--- /dev/null
+++ b/tests/scenarios/provider-bodies-late-child.yaml
@@ -0,0 +1,34 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# A child that starts late. The first checkpoint lands the child's bodies,
+# which are written ten seconds on; the main stream's next call is planned
+# after that checkpoint and written before them. A body is new when the session
+# does not hold it, not when it comes after what landed before.
+title: a child that starts late
+provider_bodies: true
+steps:
+ - input: start
+ - call:
+ agent:
+ name: helper
+ prompt: look around
+ after: 10s
+ steps:
+ - call: {text: Looking.}
+ checkpoint: launched
+ - call: {text: Back on main.}
diff --git a/tests/scenarios/provider-bodies.expect.yaml b/tests/scenarios/provider-bodies.expect.yaml
new file mode 100644
index 0000000..76769ca
--- /dev/null
+++ b/tests/scenarios/provider-bodies.expect.yaml
@@ -0,0 +1,33 @@
+# 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.
+
+# Seven calls, five on the main stream and two on the searcher's, each with
+# its request and its response joined to it: fourteen bodies, seven captured
+# prompts.
+checkpoints:
+ final:
+ kinds: {llm.call: 7, epoch.boundary: 1}
+ view:
+ state: verified
+ provider_bodies: 14
+ provider_files: 5
+ captured_prompts: 7
+collect:
+ # Small, so the provider bodies spread over many files.
+ max_delta_bytes: 8192
+push:
+ kinds: [transcript, agent_meta, provider_body, round]
diff --git a/tests/scenarios/provider-bodies.yaml b/tests/scenarios/provider-bodies.yaml
new file mode 100644
index 0000000..1e8d591
--- /dev/null
+++ b/tests/scenarios/provider-bodies.yaml
@@ -0,0 +1,50 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Provider bodies beside the steps: every call writes its request and its
+# response, as Claude Code does when OTEL_LOG_RAW_API_BODIES names a
+# directory. A child's calls land between two calls of the main stream, and
+# a compaction starts the main stream's message list again. The bodies land in
+# two passes, into files cut small, so a body refers to bodies and pieces in
+# earlier files. Each body lands in the session, rebuilds from the files up to
+# its own, and leaves the fold as it is.
+title: provider bodies beside the calls
+provider_bodies: true
+steps:
+ - input: read the configuration and summarise it
+ - call:
+ text: Reading it.
+ tool:
+ name: Read
+ input: {file_path: /Users/dev/scenario/config.yaml}
+ result: {text: "timeout: 30\nretries: 3\n", after: 200ms}
+ - call: {text: "The timeout is 30 seconds, with 3 retries."}
+ checkpoint: read
+ - input: find where the timeout is used
+ - call:
+ agent:
+ name: searcher
+ prompt: find every use of the timeout
+ steps:
+ - call:
+ text: Searching.
+ tool: {name: Bash, input: {command: grep -rn timeout .}, result: {text: "server.go:3: var timeout = 30", after: 300ms}}
+ - call: {text: It is used in server.go.}
+ - call: {text: The searcher found it in server.go.}
+ - reset: {summary: "The configuration sets a 30 second timeout, used in server.go."}
+ - input: and the retries?
+ - call: {text: There are 3 retries.}