Collect Claude Code's provider bodies, with repeats taken out

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)

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

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{}

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)
+}

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)
+	}
+}

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,