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/26 00:28:25 UTC
(skywalking-ai-sessionizer) branch mcp-executions created (now 1077fd8)
This is an automated email from the ASF dual-hosted git repository.
wu-sheng pushed a change to branch mcp-executions
in repository https://gitbox.apache.org/repos/asf/skywalking-ai-sessionizer.git
at 1077fd8 Record calls to MCP servers, and derive one metric family for every agent
This branch includes the following new commits:
new 1077fd8 Record calls to MCP servers, and derive one metric family for every agent
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: Record calls to MCP servers, and derive one metric family for every agent
Posted by wu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wu-sheng pushed a commit to branch mcp-executions
in repository https://gitbox.apache.org/repos/asf/skywalking-ai-sessionizer.git
commit 1077fd8a2798fc82102dbf7d99f1d34ead234867
Author: Wu Sheng <wu...@foxmail.com>
AuthorDate: Sat Sep 26 08:28:09 2026 +0800
Record calls to MCP servers, and derive one metric family for every agent
A transcript names an MCP call only as mcp__<server>__<tool>. It does not
say which server ran the call, how it ended or how long it took.
- The claude-code adapter lands the server and the tool on a call part
when the name splits exactly, and the tool step's attrs carry
mcp_server and mcp_tool. A name with a second "__" is left whole.
- The Claude Code plugin's hooks after every mcp__ call write an
execution/1 record, measured on Claude Code 2.1.282: the server and its
configuration source, the outcome, the duration, and the size and
SHA-256 of the arguments and the answer. The changes adapter lands them
as a new kind, execution. Assembly leaves them out. asz.view lists them
as tool_executions, and each step names its records in executions.
- Metrics are one top-level section for the root: metrics.enabled, on by
default, and metrics.lookback, 72h by default. The adapter keys metrics
and metrics_lookback are gone.
- claude_code.token.usage is now agent.token.usage. agent.mcp.calls and
agent.mcp.duration are derived from the execution records, by server,
tool, configuration source, outcome and query source.
- The claude-code-otlp receiver keeps nothing. It accepts metrics, logs
and traces and answers in the request's encoding. Every metric asz sends
is derived from the landed files, so the spool's received-request path
and spool.state are removed.
- The fold properties now compare each stream's record count, so evidence
that entered a stream would fail them. Record times in the view sort by
instant, not as text.
---
README.md | 2 +-
asz.yaml | 34 +--
cmd/asz/main.go | 38 ++--
cmd/asz/pipeline_test.go | 84 ++++++--
cmd/asz/refresh.go | 26 ++-
cmd/asz/view.go | 6 +-
docs/en/adapters/claude-code-plugin.md | 34 +++
docs/en/adapters/claude-code.md | 35 ++--
docs/en/changes/changes.md | 42 ++++
docs/en/formats/asz-view-example.yaml | 1 +
docs/en/formats/asz-view.md | 20 +-
docs/en/formats/otlp.md | 136 +++++++-----
docs/en/formats/session-data.md | 42 +++-
docs/en/formats/storage-root.md | 11 +-
docs/en/guides/scenario.md | 2 +-
docs/en/setup/claude-code-plugin.md | 9 +-
docs/en/setup/configuration.md | 32 +--
docs/en/setup/export-otlp.md | 20 +-
internal/adapters/claudecode/convert.go | 8 +-
internal/adapters/claudecode/mcpname.go | 46 +++++
internal/adapters/claudecodechanges/collect.go | 12 +-
internal/adapters/claudecodechanges/convert.go | 37 ++++
internal/adapters/claudecodechanges/discover.go | 26 ++-
internal/adapters/claudecodeotlp/receiver.go | 136 +++---------
internal/adapters/claudecodeotlp/receiver_test.go | 133 ++----------
internal/assemble/provider.go | 7 +-
internal/assemble/streams.go | 5 +-
internal/assemble/tools.go | 23 ++-
internal/config/config.go | 101 ++++-----
internal/config/config_test.go | 60 ++++--
internal/config/defaults_test.go | 4 +-
internal/export/otlp/otlp_test.go | 4 +-
internal/export/otlp/sent_test.go | 2 +-
internal/index/codec.go | 6 +-
internal/index/fromrecord.go | 3 +
internal/index/index.go | 9 +-
internal/metrics/mcp.go | 117 +++++++++++
internal/metrics/mcp_test.go | 141 +++++++++++++
internal/metrics/metrics.go | 154 +++++++++-----
internal/metrics/metrics_test.go | 31 +--
internal/metrics/parity_test.go | 19 +-
internal/metrics/session.go | 3 +-
internal/scenario/build.go | 11 +-
internal/scenario/changes.go | 34 ++-
internal/scenario/execution.go | 92 +++++++++
internal/scenario/expect/expect.go | 70 ++++++-
internal/scenario/marker.go | 2 +-
internal/scenario/plan.go | 17 +-
internal/scenario/run/push.go | 78 ++++++-
internal/scenario/run/run.go | 45 +++-
internal/scenario/scenario.go | 4 +
internal/scenario/write_claudecode.go | 17 +-
internal/scenario/write_sd.go | 47 ++++-
internal/storage/spool.go | 103 +--------
internal/verify/session.go | 2 +-
internal/view/changes.go | 39 ++--
internal/view/conversationview.go | 34 +++
internal/view/executions.go | 104 ++++++++++
internal/view/order_test.go | 12 ++
pkg/execution/execution.go | 230 +++++++++++++++++++++
pkg/execution/execution_test.go | 47 +++++
pkg/sessiondata/sessiondata.go | 13 ++
pkg/sessionview/sessionview.go | 18 ++
plugins/claude-code/internal/hook/hook.go | 15 ++
plugins/claude-code/internal/output/output.go | 33 ++-
plugins/claude-code/internal/settings/settings.go | 15 ++
plugins/claude-code/main.go | 5 +
plugins/claude-code/mcp.go | 80 +++++++
plugins/claude-code/mcp_test.go | 217 +++++++++++++++++++
plugins/claude-code/plugin/hooks/hooks.json | 8 +-
.../claude-code/testdata/mcp-hooks-2.1.282.jsonl | 9 +
tests/scenarios/all-kinds.expect.yaml | 4 +-
tests/scenarios/all-kinds.yaml | 13 +-
...all-kinds.expect.yaml => mcp-calls.expect.yaml} | 24 ++-
tests/scenarios/mcp-calls.yaml | 76 +++++++
tools/test/collector-check/main.go | 47 +++--
tools/test/e2e-collector.sh | 17 +-
77 files changed, 2478 insertions(+), 765 deletions(-)
diff --git a/README.md b/README.md
index 321dc9a..63da9ef 100644
--- a/README.md
+++ b/README.md
@@ -56,7 +56,7 @@ adapter reports it as `unavailable` rather than approximating it.
| Runtime | Status | Collection |
| --- | --- | --- |
-| [Claude Code](docs/en/adapters/claude-code.md) | collection implemented | local files — no configuration required, and it works on history that already exists. The [plugin](docs/en/setup/claude-code-plugin.md) adds which files each shell command changed. The [receiver](docs/en/setup/configuration.md#the-receiver-adapter), `claude-code-otlp`, is off by default. When Claude Code's own OpenTelemetry exporter is pointed at it, it lands the exporter's metrics requests. It drops the exporter [...]
+| [Claude Code](docs/en/adapters/claude-code.md) | collection implemented | local files — no configuration required, and it works on history that already exists. The [plugin](docs/en/setup/claude-code-plugin.md) adds which files each shell command changed. The [receiver](docs/en/setup/configuration.md#the-receiver-adapter), `claude-code-otlp`, is off by default. It accepts what Claude Code's own OpenTelemetry exporter sends and keeps none of it, so a Claude Code set to export keeps working. |
| Codex | planned | — |
| [LangChain / LangGraph](docs/en/adapters/langsmith.md) | collection implemented | a receiver, `langsmith-ingest`, the tracing client every LangChain application carries is pointed at with four environment variables; nothing changes in the application. It lands the conversation and what each model call was sent. The [plugin](docs/en/setup/langchain-plugin.md) adds which files each tool call changed. |
diff --git a/asz.yaml b/asz.yaml
index b31240b..ce6fe9d 100644
--- a/asz.yaml
+++ b/asz.yaml
@@ -55,15 +55,6 @@ adapters:
exclude:
- /private/tmp/**
- # A reconstructed subset of the runtime's own metric family, derived
- # from the landed files and sent by asz push: claude_code.token.usage
- # under the name Claude Code's OpenTelemetry exporter uses, with the
- # labels a transcript can supply. The first derivation over a root with
- # history reaches back metrics_lookback, 24h unless set; 0 or none
- # means everything.
- metrics: false
- metrics_lookback: 24h
-
collector:
# watch runs the pipeline continuously: every interval it lands what
# is new, parses what moved, and sends what the export block asks
@@ -87,16 +78,13 @@ adapters:
# repack re-cuts an existing root into a new one.
max_delta_bytes: 2097152
- # The runtime's own OpenTelemetry exporter, received here. Point Claude
- # Code at listen with CLAUDE_CODE_ENABLE_TELEMETRY=1,
- # OTEL_METRICS_EXPORTER=otlp and OTEL_EXPORTER_OTLP_ENDPOINT, over gRPC
- # or http/protobuf. Phase one lands its metrics in the spool for asz push;
- # logs and traces are accepted and dropped. metrics may be on here or on
- # claude-code-local, never both, so the same tokens are not counted twice.
+ # A listener for the runtime's own OpenTelemetry exporter. It accepts
+ # metrics, logs and traces over gRPC or HTTP and keeps none of them, so a
+ # Claude Code already set to export keeps working. The metrics asz sends
+ # are derived from the landed files; see metrics below.
- name: claude-code-otlp
enabled: false
listen: 127.0.0.1:4317
- metrics: true
# The workspace change records the asz Claude Code plugin writes: which
# files each shell command changed, and each edit made inside a
@@ -225,6 +213,20 @@ parse:
# a single landed file is published whole even when larger.
max_round_bytes: 2097152
+# The metrics asz derives from the landed files, for every adapter at once:
+# agent.token.usage from the model calls of a Claude Code conversation, and
+# agent.mcp.calls and agent.mcp.duration from the MCP calls the Claude Code
+# plugin saw. They wait in the root's spool, and asz push sends them when
+# export.otlp.metrics is on.
+metrics:
+ enabled: true
+ # How far back the first derivation over a root reaches. It runs once,
+ # before the root holds any metrics, which is usually a new deployment
+ # over a source that already has history. A minute older than this is
+ # not derived. Every later pass derives each new file whole. A duration
+ # such as 72h or 3d; 0 or none means everything.
+ lookback: 72h
+
export:
# asz collect sends every landed file and every round to an OpenTelemetry
# logs receiver at the end of every pass, and asz push sends a root that is
diff --git a/cmd/asz/main.go b/cmd/asz/main.go
index 1402aef..bf30838 100644
--- a/cmd/asz/main.go
+++ b/cmd/asz/main.go
@@ -191,7 +191,7 @@ func main() {
fmt.Fprintf(os.Stderr, "%s: the receiver runs in watch mode only; not started with -once\n", ad.Name)
continue
}
- if err := startReceiver(cfg, ad); err != nil {
+ if err := startReceiver(ad); err != nil {
fatal(err)
}
case config.AdapterLangSmithIngest:
@@ -273,24 +273,13 @@ var receivers int
// startReceiver opens a claude-code-otlp receiver on its address and leaves
// it listening for the life of the process.
-func startReceiver(cfg *config.Config, ad config.Adapter) error {
- zoneRoot, err := cfg.ResolvedRoot()
- if err != nil {
- return err
- }
- if err := os.MkdirAll(zoneRoot, 0o755); err != nil {
- return err
- }
- r := &claudecodeotlp.Receiver{Zone: storage.NewZone(zoneRoot), Listen: ad.Listen, LandMetrics: ad.Metrics}
+func startReceiver(ad config.Adapter) error {
+ r := &claudecodeotlp.Receiver{Listen: ad.Listen}
if err := r.Start(); err != nil {
return err
}
receivers++
- what := "metrics landed in the spool"
- if !ad.Metrics {
- what = "metrics accepted and dropped, metrics: false"
- }
- fmt.Printf("receiver : %s on %s, gRPC and HTTP; %s; logs and traces accepted and dropped\n", ad.Name, r.Addr(), what)
+ fmt.Printf("receiver : %s on %s, gRPC and HTTP; metrics, logs and traces accepted and dropped\n", ad.Name, r.Addr())
return nil
}
@@ -463,7 +452,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",
+ index.KindProviderBody: "provider_body", index.KindExecution: "execution",
}
byKind := map[index.Kind]int{}
msgs := map[uint32]int{}
@@ -807,7 +796,7 @@ func cmdCollect(cfg *config.Config, ads []config.Adapter, once bool) error {
return err
}
zone := storage.NewZone(zoneRoot)
- ref, err := newRefresher(nil, zone, ads, cfg.Parse.MaxRoundBytes, once)
+ ref, err := newRefresher(nil, zone, ads, cfg.Metrics, cfg.Parse.MaxRoundBytes, once)
if err != nil {
return err
}
@@ -890,15 +879,12 @@ func pushOnly(cfg *config.Config) error {
}
}
-// newDeriver is the metrics derivation an adapter asked for, or nil. A
-// single pass is the backfill path over history that already exists, with
-// no later pass to derive what it left waiting, so it never waits for the
-// next file of a stream; only a watching collector does.
-func newDeriver(zone *storage.Zone, ad config.Adapter, once bool) (*metrics.Deriver, error) {
- if !ad.Metrics {
- return nil, nil
- }
- lookback, err := ad.Lookback()
+// newDeriver is the metrics derivation over a root. A single pass is the
+// backfill path over history that already exists, with no later pass to
+// derive what it left waiting, so it never waits for the next file of a
+// stream; only a watching collector does.
+func newDeriver(zone *storage.Zone, mc config.Metrics, once bool) (*metrics.Deriver, error) {
+ lookback, err := mc.LookbackDuration()
if err != nil {
return nil, err
}
diff --git a/cmd/asz/pipeline_test.go b/cmd/asz/pipeline_test.go
index 8899b87..b885b35 100644
--- a/cmd/asz/pipeline_test.go
+++ b/cmd/asz/pipeline_test.go
@@ -28,6 +28,7 @@ import (
collogspb "go.opentelemetry.io/proto/otlp/collector/logs/v1"
collmetricspb "go.opentelemetry.io/proto/otlp/collector/metrics/v1"
+ "google.golang.org/protobuf/proto"
"github.com/apache/skywalking-ai-sessionizer/internal/config"
"github.com/apache/skywalking-ai-sessionizer/internal/export/otlp"
@@ -76,7 +77,7 @@ func TestOnePassLandsParsesAndSends(t *testing.T) {
{Name: config.AdapterClaudeCodeChanges, Enabled: true, SourceRoot: filepath.Join(source, "plugins", "data"),
Collector: config.Collector{Mode: config.ModeOnce, Interval: time.Second, MaxDeltaBytes: 1 << 20}},
}
- ref, err := newRefresher(view.New(zone, nil), zone, ads, 2<<20, true)
+ ref, err := newRefresher(view.New(zone, nil), zone, ads, noMetrics, 2<<20, true)
if err != nil {
t.Fatal(err)
}
@@ -144,7 +145,7 @@ func TestAPassWithNoReceiverStillLandsAndParses(t *testing.T) {
source := filepath.Join(root, "_source")
ads := []config.Adapter{{Name: config.AdapterClaudeCodeLocal, Enabled: true, SourceRoot: source,
Collector: config.Collector{Mode: config.ModeOnce, Interval: time.Second, MaxDeltaBytes: 1 << 20}}}
- ref, err := newRefresher(view.New(zone, nil), zone, ads, 2<<20, true)
+ ref, err := newRefresher(view.New(zone, nil), zone, ads, noMetrics, 2<<20, true)
if err != nil {
t.Fatal(err)
}
@@ -262,7 +263,7 @@ func testRefresher(t *testing.T, zone *storage.Zone, root string) *refresher {
source := filepath.Join(root, "_source")
ads := []config.Adapter{{Name: config.AdapterClaudeCodeLocal, Enabled: true, SourceRoot: source,
Collector: config.Collector{Mode: config.ModeOnce, Interval: time.Second, MaxDeltaBytes: 1 << 20}}}
- ref, err := newRefresher(view.New(zone, nil), zone, ads, 2<<20, true)
+ ref, err := newRefresher(view.New(zone, nil), zone, ads, noMetrics, 2<<20, true)
if err != nil {
t.Fatal(err)
}
@@ -285,16 +286,15 @@ func testRefresher(t *testing.T, zone *storage.Zone, root string) *refresher {
func TestTheModeIsSettledBeforeAnythingIsBuilt(t *testing.T) {
root := t.TempDir()
zone := storage.NewZone(root)
- // The metrics-enabled adapter says once and comes first; the second
- // says watch. The pipeline watches, and so must the derivation.
+ // The first adapter says once; the second says watch. The pipeline
+ // watches, and so must the derivation.
ads := []config.Adapter{
{Name: config.AdapterClaudeCodeLocal, Enabled: true, SourceRoot: filepath.Join(root, "_source"),
- Metrics: true, MetricsLookback: "none",
Collector: config.Collector{Mode: config.ModeOnce, Interval: time.Minute, MaxDeltaBytes: 1 << 20}},
{Name: config.AdapterClaudeCodeChanges, Enabled: true, SourceRoot: filepath.Join(root, "_source", "plugins", "data"),
Collector: config.Collector{Mode: config.ModeWatch, Interval: 2 * time.Second, MaxDeltaBytes: 1 << 20}},
}
- ref, err := newRefresher(view.New(zone, nil), zone, ads, 2<<20, false)
+ ref, err := newRefresher(view.New(zone, nil), zone, ads, allMetrics, 2<<20, false)
if err != nil {
t.Fatal(err)
}
@@ -307,7 +307,7 @@ func TestTheModeIsSettledBeforeAnythingIsBuilt(t *testing.T) {
t.Fatalf("the pipeline period is %s, want the shortest asked for, 2s", ref.interval)
}
if ref.deriver == nil {
- t.Fatal("no metrics derivation for an adapter that asked for one")
+ t.Fatal("no metrics derivation with metrics on")
}
// A single pass sets Grace to -1, never wait, because there is no later
// pass to derive what it left. A watching pipeline leaves it at the
@@ -317,6 +317,50 @@ func TestTheModeIsSettledBeforeAnythingIsBuilt(t *testing.T) {
}
}
+// TestMetricsAreDerivedWhicheverAdapterCollected. Metrics are the root's,
+// not an adapter's. A pipeline that collects only the plugin's records, with
+// no transcript adapter, still derives the calls to MCP servers from them.
+func TestMetricsAreDerivedWhicheverAdapterCollected(t *testing.T) {
+ root, _ := scenarioRoot(t, "mcp-calls.yaml")
+ ref := pipelineOver(t, root, bothAdapters(root, "")[1:])
+ if ref.col != nil {
+ t.Fatal("the pipeline reads transcripts; the test needs the plugin's records alone")
+ }
+ if err := ref.pass(); err != nil {
+ t.Fatal(err)
+ }
+ files, err := storage.NewSpool(storage.NewZone(root)).List()
+ if err != nil {
+ t.Fatal(err)
+ }
+ calls := 0.0
+ for _, f := range files {
+ data, err := os.ReadFile(f.Path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var req collmetricspb.ExportMetricsServiceRequest
+ if err := proto.Unmarshal(data, &req); err != nil {
+ t.Fatal(err)
+ }
+ for _, rm := range req.GetResourceMetrics() {
+ for _, sm := range rm.GetScopeMetrics() {
+ for _, m := range sm.GetMetrics() {
+ for _, dp := range m.GetSum().GetDataPoints() {
+ if m.GetName() == metrics.MCPCalls {
+ calls += dp.GetAsDouble()
+ }
+ }
+ }
+ }
+ }
+ }
+ // Four calls have a record; the first one's is written twice.
+ if calls != 4 {
+ t.Fatalf("%v calls to MCP servers derived from %d spool files, want 4", calls, len(files))
+ }
+}
+
// TestASourceThatAppearsLaterIsCollected. The plugin may be installed after
// the pipeline starts. A source directory that is absent at the start must
// not be dropped for the life of the process.
@@ -326,7 +370,7 @@ func TestASourceThatAppearsLaterIsCollected(t *testing.T) {
changes := filepath.Join(root, "_source", "plugins", "data")
ads := []config.Adapter{{Name: config.AdapterClaudeCodeChanges, Enabled: true, SourceRoot: changes,
Collector: config.Collector{Mode: config.ModeWatch, Interval: time.Second, MaxDeltaBytes: 1 << 20}}}
- ref, err := newRefresher(view.New(zone, nil), zone, ads, 2<<20, false)
+ ref, err := newRefresher(view.New(zone, nil), zone, ads, noMetrics, 2<<20, false)
if err != nil {
t.Fatal(err)
}
@@ -503,16 +547,24 @@ func scenarioRoot(t *testing.T, name string) (string, *scenario.Built) {
return root, b
}
+// allMetrics derives every metric with no look-back, as a build's
+// configuration does; noMetrics derives none.
+var (
+ allMetrics = config.Metrics{Lookback: "none"}
+ noMetrics = config.Metrics{Enabled: new(bool)}
+)
+
// bothAdapters is what a build's configuration enables: both local
-// adapters, the plugin's reading under the build's source unless another
-// directory is given, and metrics derived with no look-back.
+// adapters, and the plugin's reading under the build's source unless another
+// directory is given. pipelineOver derives metrics over them with no
+// look-back.
func bothAdapters(root, changes string) []config.Adapter {
source := filepath.Join(root, "_source")
if changes == "" {
changes = filepath.Join(source, "plugins", "data")
}
return []config.Adapter{
- {Name: config.AdapterClaudeCodeLocal, Enabled: true, SourceRoot: source, Metrics: true, MetricsLookback: "none",
+ {Name: config.AdapterClaudeCodeLocal, Enabled: true, SourceRoot: source,
Collector: config.Collector{Mode: config.ModeOnce, Interval: time.Second, MaxDeltaBytes: 1 << 20}},
{Name: config.AdapterClaudeCodeChanges, Enabled: true, SourceRoot: changes,
Collector: config.Collector{Mode: config.ModeOnce, Interval: time.Second, MaxDeltaBytes: 1 << 20}},
@@ -523,7 +575,7 @@ func bothAdapters(root, changes string) []config.Adapter {
func pipelineOver(t *testing.T, root string, ads []config.Adapter) *refresher {
t.Helper()
zone := storage.NewZone(root)
- ref, err := newRefresher(view.New(zone, nil), zone, ads, 2<<20, true)
+ ref, err := newRefresher(view.New(zone, nil), zone, ads, allMetrics, 2<<20, true)
if err != nil {
t.Fatal(err)
}
@@ -906,7 +958,7 @@ func TestProviderBodiesReachesTheReceiverCollector(t *testing.T) {
zone := storage.NewZone(root)
ads := []config.Adapter{{Name: config.AdapterLangSmithIngest, Enabled: true,
Listen: "127.0.0.1:0", ProviderBodies: on}}
- ref, err := newRefresher(view.New(zone, nil), zone, ads, 2<<20, true)
+ ref, err := newRefresher(view.New(zone, nil), zone, ads, noMetrics, 2<<20, true)
if err != nil {
t.Fatal(err)
}
@@ -969,7 +1021,7 @@ func TestTwoRecorderDirectoriesAreBothCollected(t *testing.T) {
{Name: config.AdapterChanges, Enabled: true, SourceRoot: pluginData, Collector: once},
{Name: config.AdapterChanges, Enabled: true, SourceRoot: shimData, Collector: once},
}
- ref, err := newRefresher(view.New(zone, nil), zone, ads, 2<<20, true)
+ ref, err := newRefresher(view.New(zone, nil), zone, ads, noMetrics, 2<<20, true)
if err != nil {
t.Fatal(err)
}
@@ -1009,7 +1061,7 @@ func TestTwoSpellingsOfOneRecorderDirectoryAreRefused(t *testing.T) {
{Name: config.AdapterChanges, Enabled: true, SourceRoot: data, Collector: once},
{Name: config.AdapterChanges, Enabled: true, SourceRoot: alias, Collector: once},
}
- if _, err := newRefresher(view.New(zone, nil), zone, ads, 2<<20, true); err == nil || !strings.Contains(err.Error(), "both read") && !strings.Contains(err.Error(), "read "+data) {
+ if _, err := newRefresher(view.New(zone, nil), zone, ads, noMetrics, 2<<20, true); err == nil || !strings.Contains(err.Error(), "both read") && !strings.Contains(err.Error(), "read "+data) {
t.Fatalf("two spellings of one directory were accepted: %v", err)
}
}
diff --git a/cmd/asz/refresh.go b/cmd/asz/refresh.go
index 61b6474..6f521b8 100644
--- a/cmd/asz/refresh.go
+++ b/cmd/asz/refresh.go
@@ -171,7 +171,7 @@ type refresher struct {
// plugin may be installed after the pipeline starts, and a source that
// appears later has to be picked up without a restart; every pass looks
// again.
-func newRefresher(srv *view.Server, zone *storage.Zone, ads []config.Adapter, maxRound int64, once bool) (*refresher, error) {
+func newRefresher(srv *view.Server, zone *storage.Zone, ads []config.Adapter, mc config.Metrics, maxRound int64, once bool) (*refresher, error) {
r := &refresher{srv: srv, zone: zone, maxRound: maxRound, full: true, retry: map[string]bool{}}
// One pipeline serves every adapter, so it needs one mode and one
@@ -199,11 +199,6 @@ func newRefresher(srv *view.Server, zone *storage.Zone, ads []config.Adapter, ma
if err != nil {
return nil, err
}
- deriver, err := newDeriver(zone, ad, mode == config.ModeOnce)
- if err != nil {
- return nil, err
- }
- r.deriver = deriver
r.col = claudecode.New(src, zone, ad.Collector.MaxDeltaBytes)
r.match = claudecode.NewMatcher(ad.Include, ad.Exclude).Match
r.colSource = src
@@ -252,6 +247,15 @@ func newRefresher(srv *view.Server, zone *storage.Zone, ads []config.Adapter, ma
fmt.Fprintln(os.Stderr, "source : no local adapter enabled; nothing is collected")
return nil, nil
}
+ // Metrics are the root's, not an adapter's: one derivation reads every
+ // landed file, whichever adapter landed it.
+ if mc.On() {
+ deriver, err := newDeriver(zone, mc, mode == config.ModeOnce)
+ if err != nil {
+ return nil, err
+ }
+ r.deriver = deriver
+ }
r.base = view.Status{Mode: mode, Adapter: strings.Join(names, "+"), Source: strings.Join(sources, ", ")}
if mode == config.ModeWatch {
r.base.IntervalMS = r.interval.Milliseconds()
@@ -420,6 +424,10 @@ func (r *refresher) pass() error {
changed := map[string]bool{}
var cs *claudecode.Stats
+ // collected says a collector reached its source this pass. The first
+ // derivation over a root is bounded by the look-back and runs once, so
+ // it waits for a source to be there.
+ collected := false
colHere := false
if r.col != nil {
var cerr error
@@ -428,6 +436,7 @@ func (r *refresher) pass() error {
}
}
if colHere {
+ collected = true
var err error
cs, err = r.col.CollectAll(r.match)
if err != nil {
@@ -453,6 +462,7 @@ func (r *refresher) pass() error {
if !changesHere {
continue
}
+ collected = true
ch, err := cs.col.CollectAll(cs.match)
if err != nil {
errs = append(errs, err)
@@ -480,6 +490,7 @@ func (r *refresher) pass() error {
}
}
if providerHere {
+ collected = true
projects := r.colSource
if projects == "" {
projects, _ = claudecode.ResolveSourceRoot("")
@@ -503,6 +514,7 @@ func (r *refresher) pass() error {
}
if r.langsmith != nil {
+ collected = true
landedRuns, err := r.langsmith.Collect()
if err != nil {
errs = append(errs, err)
@@ -533,7 +545,7 @@ func (r *refresher) pass() error {
sessions = all
}
}
- if r.deriver != nil && cs != nil {
+ if r.deriver != nil && collected {
// The first pass derives history once. A later pass derives what
// either adapter moved, what waited on a lock, and what either the
// deriver or removal left for another pass.
diff --git a/cmd/asz/view.go b/cmd/asz/view.go
index 0522e31..743efe3 100644
--- a/cmd/asz/view.go
+++ b/cmd/asz/view.go
@@ -146,7 +146,7 @@ func cmdServer(cfg *config.Config, ads []config.Adapter, once bool) error {
zone := storage.NewZone(zoneRoot)
srv := view.NewWithGlossaries(zone, glossaries())
- ref, err := newRefresher(srv, zone, ads, cfg.Parse.MaxRoundBytes, once)
+ ref, err := newRefresher(srv, zone, ads, cfg.Metrics, cfg.Parse.MaxRoundBytes, once)
if err != nil {
return err
}
@@ -213,8 +213,8 @@ func printPipeline(ref *refresher, cfg *config.Config) {
fmt.Printf("source : %s (once)\n", ref.base.Source)
}
if ref.deriver != nil {
- fmt.Printf("metrics : %s, derived from the landed files, look-back %s on the first pass\n",
- metrics.TokenUsage, lookbackWord(ref.deriver.Lookback))
+ fmt.Printf("metrics : %s, %s and %s, derived from the landed files, look-back %s on the first pass\n",
+ metrics.TokenUsage, metrics.MCPCalls, metrics.MCPDuration, lookbackWord(ref.deriver.Lookback))
}
if ref.pusher == nil {
fmt.Println("export : none; set export.otlp.endpoint to send what is landed")
diff --git a/docs/en/adapters/claude-code-plugin.md b/docs/en/adapters/claude-code-plugin.md
index a85d1c3..2fe4198 100644
--- a/docs/en/adapters/claude-code-plugin.md
+++ b/docs/en/adapters/claude-code-plugin.md
@@ -67,6 +67,40 @@ the model's own words, not in Claude Code's shape. Records of a different shape
dialect, even when the same runtime produced them, as the
[adapter contract](../concepts-and-designs/unified-conversation-model.md#adapter-contract) says.
+## Calls to MCP servers
+
+Claude Code runs the plugin's hook after every call to an MCP server, on the main stream and inside
+subagents. The hooks match `mcp__.*`: `PostToolUse` after a call that returned, and
+`PostToolUseFailure` after one that did not. The plugin appends one `execution/1` record per call,
+to a file beside the stream's change records:
+
+```text
+${CLAUDE_PLUGIN_DATA}/output/<session-id>/main.execution.jsonl
+${CLAUDE_PLUGIN_DATA}/output/<session-id>/<agent-id>.execution.jsonl
+```
+
+A record names the call by its tool-use id. It carries the server's name and where its
+configuration came from, how the call ended, `returned`, `failed` or `interrupted`, and the time
+Claude Code measured around it. It keeps the size and the SHA-256 of the arguments and of the
+answer, never their text. `pkg/execution` in the repository defines the shape, and
+[Execution records](../formats/session-data.md#execution-records) describes it as asz lands it.
+
+Measured on Claude Code 2.1.282 with a test MCP server, in the fixture
+`plugins/claude-code/testdata/mcp-hooks-2.1.282.jsonl`:
+
+- An error the server returned, a server that exited and a call that timed out all reach
+ `PostToolUseFailure`.
+- Every hook carried `duration_ms`, failures included.
+- A call the permission check denied reaches no hook and leaves no record.
+- The arguments a record measures are what the server received. A `PreToolUse` hook that rewrote
+ the input changed what the server received and what the hook after the call saw. The transcript
+ kept the model's own input.
+- A server started with `--mcp-config` has the source `dynamic`.
+- Claude Code passes a tool's annotations, such as a hint that it only reads, to neither the model,
+ the hooks nor the transcript. So a record cannot say whether a tool reads or writes.
+
+`mcp.enabled: false` in the settings turns the records off.
+
## Exclusions
Exclusion means no observation: a change under an excluded directory is never seen. Every record
diff --git a/docs/en/adapters/claude-code.md b/docs/en/adapters/claude-code.md
index e3e0f06..53ed491 100644
--- a/docs/en/adapters/claude-code.md
+++ b/docs/en/adapters/claude-code.md
@@ -682,8 +682,12 @@ attachment travels whole as one `data` part, so all its fields are kept. System
journal lines and manifests are of this kind.
**An MCP tool use is an ordinary `call`.** Its server is only a prefix on the name,
-`mcp__<server>__<tool>`. This was observed on 39 of 26,940 tool uses (0.1%), too few to state as a
-rule.
+`mcp__<server>__<tool>`. The adapter lands the server and the tool as the part's `server` and
+`server_tool` only when exactly one `__` follows `mcp__`. A server or tool whose own name holds
+`__` cannot be told apart from the separator, so such a name is left whole. MCP tool uses were 39
+of 26,940 tool uses (0.1%) in the corpus. Which server ran a call, how it ended and how long it
+took are not in the transcript. The plugin's [execution records](claude-code-plugin.md#calls-to-mcp-servers)
+carry them.
Five record fields come from places the glossary does not list:
@@ -853,11 +857,11 @@ the binary are unreachable in shipped builds.
Its content defaults are restrictive, as [Collection](#collection) describes.
-### The same metrics from the transcripts
+### Metrics from the transcripts
-With `metrics: true`, this adapter derives a reconstructed subset of the runtime's own metric
-family from what it landed, under the exporter's metric name, so a receiver holds one name
-whichever produced the points. Phase one is `claude_code.token.usage`: the usage of a call is its
+asz derives `agent.token.usage` from what this adapter landed. It counts what the runtime's own
+exporter calls `claude_code.token.usage`, with the same description, unit, kind and token types,
+under asz's own name, because a runtime's vocabulary stops at its adapter. The usage of a call is its
last fragment's in line order, as the assembler reads it, and only a finished call counts;
`query_source` is `main` for the session's own transcript and `subagent` for a child's; `model`
is `message.model`; `session.id` is the session. The exporter's account, organisation, terminal,
@@ -869,20 +873,17 @@ the derivation is tested against.
What the transcripts do not carry is not derived: cost, per-request latency, tool execution time,
active time, lines of code, commits, pull requests, the session start type, and the tokens of the
-runtime's auxiliary calls, which never reach a transcript. Those come from the exporter alone.
-The first derivation over a root with history reaches back `metrics_lookback`, 24 hours unless
-set. See [Metrics](../setup/export-otlp.md#metrics).
+runtime's auxiliary calls, which never reach a transcript. asz does not send them. The first
+derivation over a root with history reaches back `metrics.lookback`, 72 hours unless set. See
+[Metrics](../setup/export-otlp.md#metrics).
### The runtime's exporter, received
-`claude-code-otlp` is the other adapter for this runtime: an OpenTelemetry receiver its
-exporter is pointed at, over gRPC or HTTP with protobuf on one port. Phase one lands the metrics
-requests it receives in the spool, bytes as received, and `asz push` sends them under asz's
-identity; logs and traces are accepted and dropped. It lands no transcript and reads none, so it
-adds no structure; the two adapters meet only in the spool, and `metrics` is on for one of them.
-The runtime's events, with the latency and cost a transcript never carries, are the second phase,
-and the question there is whether they land as Session Data joined to `llm.call` by the request
-id or travel untouched.
+`claude-code-otlp` is an OpenTelemetry receiver the runtime's exporter can be pointed at, over
+gRPC or HTTP on one port. It accepts metrics, logs and traces, and keeps none of them. A runtime
+already set to export keeps working, and nothing it sends is counted beside what asz derives. It
+lands no transcript and reads none, so it adds no structure. The runtime's events carry the
+latency and cost a transcript never has. Landing them is not implemented.
The receiver speaks OTLP because Claude Code has exactly one outward protocol. Its exporter setting
accepts `console`, `otlp` or `prometheus` for metrics, and `console` or `otlp` for logs and traces.
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index 5fd2ff6..5a10a52 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -32,8 +32,50 @@
is now external input that opens its talk. A session collected before this keeps the gap, because
landed records are never rewritten.
+## MCP calls
+
+- A call to an MCP server names its server and its tool. Claude Code calls an MCP tool
+ `mcp__<server>__<tool>`. A landed call part now carries `server` and `server_tool` when the name
+ splits exactly, with one `__` after `mcp__`, and the step's `attrs` carry `mcp_server` and
+ `mcp_tool`. A name with a second `__` is left whole, since the separator cannot be told apart
+ from a name that holds it. A session collected before this has neither, because landed records
+ are never rewritten.
+- The Claude Code plugin records each call to an MCP server. Its hooks after every `mcp__` call
+ write an `execution/1` record: the server and where its configuration came from, how the call
+ ended, the time Claude Code measured around it, and the size and SHA-256 of the arguments and of
+ the answer, never their text. The `changes` adapter lands them as a new kind, `execution`.
+ Assembly does not read them. `asz.view` lists them as `tool_executions`, and a step names its
+ records in `executions`, joined by the tool-use id. `mcp.enabled: false` in the plugin's
+ settings turns them off. The shapes were measured on Claude Code 2.1.282, and the real hook
+ payloads are the plugin's test data.
+
+## Metrics
+
+- Metrics are one section of the configuration, for the whole root. `metrics.enabled`, on by
+ default, derives every metric from every landed file, whichever adapter landed it.
+ `metrics.lookback`, 72 hours by default, bounds the first derivation over a root that already
+ has history. The adapter keys `metrics` and `metrics_lookback` are gone, and a file that still
+ writes them has them ignored. Metrics used to be off by default, with a look-back of 24 hours.
+- `claude_code.token.usage` is now `agent.token.usage`, with the same description, unit, kind and
+ labels. The family is asz's and is meant for any agent, and a runtime's vocabulary stops at its
+ adapter. A receiver's rules must read the new name.
+- Two new metrics count the calls to MCP servers, from the plugin's execution records:
+ `agent.mcp.calls` and `agent.mcp.duration`, by server, tool, the source of the server's
+ configuration, outcome and query source. A server and a tool together name the target a call
+ reached, which a receiver can treat as an endpoint of the agent.
+- The `claude-code-otlp` receiver no longer keeps what Claude Code's exporter sends. It accepts
+ metrics, logs and traces, over gRPC or over HTTP in any encoding, and keeps none of them. Its
+ `metrics` key is gone. Every metric asz sends is derived from the landed files, so one root has one
+ source for each count. The exporter's labels a transcript does not carry, such as the user and
+ the organisation, and its other metrics, such as cost, are no longer sent. The spool holds only
+ derived requests, and `spool.state` is no longer written.
+
## Development
+- The `all-kinds` scenario holds a call to an MCP server, so the push to a real OpenTelemetry
+ Collector carries an `execution` file and the MCP metrics, and the check compares them with the
+ root. The same run sends a real exporter's metrics to the `claude-code-otlp` receiver, and fails
+ if any of them reaches the spool or the Collector.
- The first benchmark: `BenchmarkLangChainRequests` measures cutting every request body of one
LangChain conversation, for context windows from 8K to 1M tokens, trimmed, summarised or neither.
`make test` and CI only compile it. Its comment gives the command that runs it, the reference
diff --git a/docs/en/formats/asz-view-example.yaml b/docs/en/formats/asz-view-example.yaml
index fe39bae..a522e57 100644
--- a/docs/en/formats/asz-view-example.yaml
+++ b/docs/en/formats/asz-view-example.yaml
@@ -846,4 +846,5 @@ workspace_changes:
- ' '
- -const Version = "0.2.0"
- +const Version = "0.3.0"
+tool_executions: []
diff --git a/docs/en/formats/asz-view.md b/docs/en/formats/asz-view.md
index 201048a..9c5dd7d 100644
--- a/docs/en/formats/asz-view.md
+++ b/docs/en/formats/asz-view.md
@@ -84,6 +84,7 @@ producer of them.
| `relations` | one per relation of the fold: `id`, `type`, `from`, `to`, `quality`, `via`, `evidence` |
| `unresolved` | one per reference the assembler could not resolve, open or since resolved: `id`, `kind`, `ref`, `reason`, `state` |
| `workspace_changes` | one per workspace change record the session's files carry, joined to its step. See below. |
+| `tool_executions` | one per tool execution record the session's files carry, joined to its step. See below. |
Verification is content, not an error. A gap in the chain or a failed digest is written into
`summary.state` and `summary.problems`, each round says whether it verified, and the rest of the
@@ -113,7 +114,8 @@ usable round at all is an error, because there is nothing to show.
| `text`, `state`, `bytes` | the part the node stands on: its readable text, clipped to the longest prefix of whole characters within 2,000 bytes, whether the content is `available`, and its full size. For a `data` part the text is the data as compact JSON. Earlier writers put `\u003c`, `\u003e` and `\u0026` there for `<`, `>` and `&`, so their step text can differ (see [What data holds](session-data.md#what-data-holds)). A reader wanting the whole record reads it by address. |
| `usage`, `flags`, `dropped` | what else the referenced record says, copied once: on an `llm.call`, the token counts `in`, `out`, `cache_read`, `cache_write` from the one record `usage_at` names, never a sum over fragments; the record's `flags`; and its `dropped` list, so a viewer can say what was left out and why |
| a talk adds | `label` and `reply`, clipped the same way and described below, then `runs`, `steps`, `tools`, `from`, `to`, `child`, `segment` |
-| a tool or agent call adds | `name`, `failed`, `result`, `result_state`, `result_bytes`, `request_to_result_ms` and `request_to_result_join`, the time from the request record to the result record where the assembler joined them exactly |
+| a tool or agent call adds | `name`, `failed`, `result`, `result_state`, `result_bytes`, `request_to_result_ms` and `request_to_result_join`, the time from the request record to the result record where the assembler joined them exactly; `changes` and `executions`, the records joined to it |
+| a call to an MCP server has in `attrs` | `mcp_server` and `mcp_tool`, the server and the tool the runtime's name for the call addresses, where the name splits exactly (see [Parts](session-data.md#parts)). They are the runtime's names; which server ran the call, by its configured name, is in the call's execution record |
| a `turn.duration` step adds | `duration_ms`, `duration_measured_by` |
| `children` | containment, in record order: a talk holds runs, a run holds steps, a call holds what it produced |
| `edges` | every relation touching the node, in both directions, as `{type, other, dir, quality, via}`, ordered by relation id and then direction, so a viewer draws cross-stream flow without searching `relations` and the same fold gives the same list |
@@ -163,6 +165,22 @@ 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.
+## Tool executions
+
+`tool_executions` lists what an observer around a tool call saw it do, one entry per execution
+record, in time order. An entry carries where the record was read from and the step it joins to,
+then the record's own fields as [`execution/1`](session-data.md#execution-records) lists them:
+
+| Key | Value |
+| --- | --- |
+| `step` | the id of the tool step whose tool-use id the record names. Empty when no step of the document carries the id; the record is kept, not dropped |
+| `ref` | the landed record the entry was built from, in the shape every node's `ref` has. `seq` names an `execution` file under `files` |
+| `schema`, `id`, `observed_by`, `boundary`, `session`, `stream`, `tool`, `tool_name`, `cwd`, `protocol` | the record's identity: `id` is the observation's own, so one call can have several entries |
+| `server`, `time`, `duration_ms`, `outcome`, `arguments`, `result` | which server ran the call and where its configuration came from, when the observation ended, the time the runtime measured around the call, how it ended, and the size and digest of what was sent and what came back |
+
+A tool step lists the ids of its records under `executions`, in the order `tool_executions` lists
+them. The records are not counted in `summary`, and they add no node.
+
## Provider bodies
A call step lists the provider bodies joined to it under `provider_bodies`: its request, then its
diff --git a/docs/en/formats/otlp.md b/docs/en/formats/otlp.md
index 3b97ac6..0105821 100644
--- a/docs/en/formats/otlp.md
+++ b/docs/en/formats/otlp.md
@@ -2,7 +2,7 @@
What [Export over OpenTelemetry](../setup/export-otlp.md) sends, for anyone building or checking a
receiver: how files become log records, how they travel, the attributes every record carries, and
-how the token metrics are derived.
+how the metrics are derived.
## One log record per file
@@ -120,21 +120,37 @@ fails the build. See [Scenarios](../guides/scenario.md).
## Metrics
-`asz push` sends metrics as well, when an adapter produces them, under the name Claude Code's
-own OpenTelemetry exporter uses, so a receiver holds one metric name whichever produced the
-points. What the local adapter derives is a reconstructed subset of the exporter's family, not a
-copy of it, and this table says exactly which part:
+asz derives these metrics from the landed files when `metrics.enabled` is on, and `asz push` sends
+them when `export.otlp.metrics` is on. Both are on by default. The metrics are asz's own family,
+named for any agent rather than for one runtime:
-| | The runtime's exporter | Derived from the transcripts |
+| Metric | Unit | One point | Attributes |
+| --- | --- | --- | --- |
+| `agent.token.usage` | `tokens` | the tokens of the model calls that finished in a minute, of one type | `session.id`, `model`, `query_source`, `type` |
+| `agent.mcp.calls` | `{call}` | the calls to MCP servers observed in a minute | `session.id`, `mcp_server.name`, `mcp_server.source`, `mcp_tool.name`, `outcome`, `query_source` |
+| `agent.mcp.duration` | `ms` | the time the runtime measured around those calls, summed | the same as `agent.mcp.calls` |
+
+Each is a monotonic delta sum, and every value is a double. A request's resource says
+`service.name` `claude-code`, which asz normalises on the way out, and its instrumentation scope is
+asz's own. The `claude-code-otlp` adapter accepts what Claude Code's own exporter sends and keeps
+none of it, so one root has one source for each count.
+
+### Token usage
+
+`agent.token.usage` is derived from Claude Code transcripts. It counts what Claude Code's own
+OpenTelemetry exporter calls `claude_code.token.usage`, and differs from it exactly where this table
+says:
+
+| | Claude Code's exporter | Derived from the transcripts |
| --- | --- | --- |
-| metric | `claude_code.token.usage`, `Number of tokens used`, unit `tokens`, a monotonic delta sum | the same name, description, unit and kind |
+| metric | `claude_code.token.usage`, `Number of tokens used`, unit `tokens`, a monotonic delta sum | `agent.token.usage`, with the same description, unit and kind |
| `type` | `input`, `output`, `cacheRead`, `cacheCreation`, a point for each even when zero | the same four, zero included |
| `query_source` | `main`, `subagent`, `auxiliary` | `main` and `subagent`, from the stream the call was made on; the auxiliary calls, such as the Haiku call that names a session, never reach a transcript |
| `model`, `session.id` | yes | yes |
| `user.id`, `user.email`, `user.account_uuid`, `user.account_id`, `organization.id`, `terminal.type`, `effort`, `speed`, and the agent, skill, plugin and MCP attribution | yes | no; a transcript does not carry them |
| the other seven metrics: cost, active time, lines of code, commits, pull requests, sessions started, edit decisions | yes | no, and never estimated |
-| resource | `service.name` `claude-code`, `service.version`, `host.arch`, `os.type`, `os.version` | `service.name` `claude-code`; asz normalises both on the way out |
-| instrumentation scope | `com.anthropic.claude_code`, versioned as the runtime | asz's own, so a receiver that keys on the scope sees two streams of one name; the OAP keys on the name and the labels |
+| resource | `service.name` `claude-code`, `service.version`, `host.arch`, `os.type`, `os.version` | `service.name` `claude-code` |
+| instrumentation scope | `com.anthropic.claude_code`, versioned as the runtime | asz's own |
| a point's window | the exporter's export interval, wall clock | the minute the call's last fragment ended, see below |
| value | a double | a double |
@@ -145,19 +161,19 @@ account for. A new capture enters that directory only through the same test, whi
identifying values:
`go test ./internal/metrics -run TestCaptures -capture "$PWD/FILE.pb" -capture-name NAME`.
-With `metrics: true` on the `claude-code-local` adapter, the collector derives the points from
-the landed files by the assembler's own rule: the usage of a call is its last fragment's in line
-order, never a sum, and only a call that finished counts. A main transcript repeats the final usage
-on every fragment. A child's carries streaming partials on all but the last. A call has finished
-when one of its fragments reports a stop reason. A call that never does is left out, because its
-usage block is a streaming stub, not a count. That is about 10% of all calls in a corpus of 2,970
-files and 365,825 records written by Claude Code 2.1.220 to 2.1.251. It is 14.5% of the calls on
-child streams and 0.02% of those on main transcripts, as
-[Provider calls](../adapters/claude-code.md#provider-calls) reports.
+The collector derives the points from the landed files by the assembler's own rule: the usage of a
+call is its last fragment's in line order, never a sum, and only a call that finished counts. A main
+transcript repeats the final usage on every fragment. A child's carries streaming partials on all
+but the last. A call has finished when one of its fragments reports a stop reason. A call that never
+does is left out, because its usage block is a streaming stub, not a count. That is about 10% of all
+calls in a corpus of 2,970 files and 365,825 records written by Claude Code 2.1.220 to 2.1.251. It
+is 14.5% of the calls on child streams and 0.02% of those on main transcripts, as [Provider
+calls](../adapters/claude-code.md#provider-calls) reports.
A call cut at a landed file boundary is read through following transcript files of its stream
-when the file's last record is a fragment of that call. There is no one-file limit. Metadata and
-change files do not interrupt the reading; the first unrelated transcript record does. A tool
+when the file's last record is a fragment of that call. There is no one-file limit. Metadata,
+change and execution files do not interrupt the reading; the first unrelated transcript record
+does. A tool
result carries no call id, and in the same corpus a call's own tool results sit between its
fragments on 24% of multi-fragment calls. Such a result ends the read-ahead. An unfinished child
call remains eligible when its final fragment arrives later. On a main transcript every fragment
@@ -171,39 +187,53 @@ with a tool result does not wait. A single pass, the backfill over history that
derives with what is there and never waits. A call is never counted twice, however many files its
records reach, and a record the runtime re-emitted before a context reset is the same call again.
-Points are summed per minute and attribute set, and the windows of one series never overlap: a
-point takes its minute unless the series already has a point at or past it, as when two children
-ran in the same minute or a child's file landed later, in which case it follows the series' last
-point. No point of a series is ever thrown away for another. The first derivation over a root
-with history is bounded by `metrics_lookback`, 24 hours unless set, and by the newest request the
-receiver adapter landed, so switching the flag on sends neither a year of tokens nor what the
-runtime's exporter already sent. A session the first pass did not finish keeps that look-back
-across retries and restarts, whether a file waited for its grace or an error stopped the session
-before its later files. Once every file of the session is derived, a later new file is derived
-whole. Before a request enters the
-spool, an immutable receipt under the session's `metrics/` records its exact bytes and the calls
-it counted. If saving progress fails, the retry uses that receipt even when more source fragments
-have arrived, so its accounting still matches the request already written. The retry applies every
-such receipt before it derives any file again, and a receipt only moves a series forward. So a file
-that waited for its grace in the failed pass takes a window after the ones already in the spool,
-and no two windows of a series overlap.
-
-The other source of the same family is the runtime's exporter itself: the `claude-code-otlp`
-adapter receives what Claude Code sends and lands each metrics request in the same spool, bytes as
-received. The receiver does not check whether it has landed a request before. A request the
-exporter sends twice lands twice, and `asz push` sends both copies. One root sends one source:
-`metrics` may be on for the local adapter or for the receiver, and the configuration refuses both.
-
-The points wait in the storage root's `_metrics/` spool, one write-once file per landed file
-with points or per request received, and go out in order under the same budget and the same
-once-only rule as the files. `export.otlp.logs` and `export.otlp.metrics` switch the two things a
-push sends, the files and rounds as logs and the spool as metrics, so a receiver that takes one
-and not the other is sent what it takes.
-On the way out the resource is normalised to asz's identity, the service, the layer, the sender,
-so the OAP holds one service for the runtime. A receiver that answers with a partial success has
-taken the request, and the protocol says not to send it again: the rejected records or points are
-counted on the pass line as `rejected`, the file is marked sent, and `push.state` gets a `rejected`
-line for it.
+### Calls to MCP servers
+
+`agent.mcp.calls` and `agent.mcp.duration` are derived from the execution records the
+[Claude Code plugin](../adapters/claude-code-plugin.md#calls-to-mcp-servers) writes, never from a
+transcript, which does not say which server ran a call or how long it took. No runtime writes a
+metric of its own for these, so there is no exporter to compare them with.
+
+One record is one call. It is counted once per session by the record's own id, in the minute it was
+observed. `mcp_server.name` is the server the plugin reported. `mcp_tool.name` is the runtime's
+name for the call after `mcp__<server>__`, where that name splits exactly, and otherwise the whole
+name. The server and the tool together name the target a call reached, which a receiver can treat
+as an endpoint of the agent. `mcp_server.source` is where the server's configuration came from,
+such as `user`, `project` or `dynamic`. `outcome` is `returned`, `failed` or `interrupted`.
+`query_source` is `main` or `subagent`, from the stream the call ran on.
+
+The duration is Claude Code's own time around the call. It includes any waiting before the call
+started, and it is not the server's own time. Divided by the calls of the same series, it is the
+mean. Measured on Claude Code 2.1.282, every record carried a duration, failures included. A call
+the permission check denied runs no hook, so it is not counted. Nothing says whether a tool reads
+or writes, because Claude Code passes a tool's annotations to neither the hooks nor the transcript.
+
+### Windows and the first pass
+
+Points are summed per minute and attribute set, and the windows of one series never overlap: a point
+takes its minute unless the series already has a point at or past it, as when two children ran in
+the same minute or a child's file landed later, in which case it follows the series' last point. No
+point of a series is ever thrown away for another. The first derivation over a root with history is
+bounded by `metrics.lookback`, 72 hours unless set, so a new deployment over a source with months of
+history does not send all of it. A session the first pass did not finish keeps that look-back across
+retries and restarts, whether a file waited for its grace or an error stopped the session before its
+later files. Once every file of the session is derived, a later new file is derived whole. Before a
+request enters the spool, an immutable receipt under the session's `metrics/` records its exact
+bytes and the calls it counted. If saving progress fails, the retry uses that receipt even when more
+source fragments have arrived, so its accounting still matches the request already written. The
+retry applies every such receipt before it derives any file again, and a receipt only moves a series
+forward. So a file that waited for its grace in the failed pass takes a window after the ones
+already in the spool, and no two windows of a series overlap.
+
+The points wait in the storage root's `_metrics/` spool, one write-once file per landed file with
+points, and go out in order under the same budget and the same once-only rule as the files.
+`export.otlp.logs` and `export.otlp.metrics` switch the two things a push sends, the files and
+rounds as logs and the spool as metrics, so a receiver that takes one and not the other is sent what
+it takes. On the way out the resource is normalised to asz's identity, the service, the layer, the
+sender, so the OAP holds one service for the runtime. A receiver that answers with a partial success
+has taken the request, and the protocol says not to send it again: the rejected records or points
+are counted on the pass line as `rejected`, the file is marked sent, and `push.state` gets a
+`rejected` line for it.
## Rate
diff --git a/docs/en/formats/session-data.md b/docs/en/formats/session-data.md
index 8d21487..9791380 100644
--- a/docs/en/formats/session-data.md
+++ b/docs/en/formats/session-data.md
@@ -79,6 +79,7 @@ is a directory with one file per body, and each record names its own file.
| `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 [...]
+| `execution` | what the asz Claude Code plugin's hook saw of each call to an MCP server, one line per call, in a file of its own beside the stream's change file | one record per observation, with `id`, `tool` and `time` lifted from it and the line whole as one `data` part, an `execution/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`, and one step can have several. See [Execution records](#execution-records) |
| `provider_body` | what a model call was sent and what came back: the bodies Claude Code wrote for its provider, one file per body, or a LangChain run's `inputs` and `outputs` | one record per body, with `id` unique in the session (for Claude Code, the body's file name without `.json`; for LangChain, the run id followed by `:request` or `:response`), `run` on a request, `call` on a response, `model`, and the parts described in [Provider bodies](#provider-bodies); no `from` and no flags, [...]
A source that is one document, not a stream of lines, lands as one record with `ord` 1 and `off` 0.
@@ -233,12 +234,19 @@ reads every landed file this way.
| --- | --- | --- |
| `text` | readable text | `text` |
| `reasoning` | the model's own reasoning | `text` when the runtime kept it |
-| `call` | a request to run something | `id`, `name`, `data` |
+| `call` | a request to run something | `id`, `name`, `data`, and `server` and `server_tool` for a call to an MCP server |
| `result` | what a call returned | `of`, `failed`, and `text`, `data` or both |
| `media` | an image or a document | `media`, `data` |
| `data` | structure that is not prose: a record the runtime keeps for itself, a manifest | `data` |
| `unknown` | content the dialect could not describe | the bytes in `data` as one JSON string, the bytes themselves or their base64 when `encoding` is `base64`, and the reason in `text` |
+`server` and `server_tool` say which MCP server a call is addressed to and which of its tools. An
+adapter sets them only where its runtime's name for the call splits exactly. Claude Code names such a
+call `mcp__<server>__<tool>`, and the split is made only when exactly one `__` follows `mcp__`: a
+name that holds a second `__` could split two ways, so it is left whole. The server is Claude Code's
+form of the configured name, so "claude.ai Claude Docs" is `claude_ai_Claude_Docs`. The name the
+server was configured with is in the call's execution record, when there is one.
+
Every part carries `state` and `bytes`. `state` is one of `available`, `truncated`, `redacted`,
`omitted` or `unavailable`, and `bytes` is the size of the original even when the part holds less.
A reader is always told how much of the original it has.
@@ -333,6 +341,38 @@ Claude Code files, six content-block shapes exist and four cover 99.99% of conte
Claude Code [Session Data mapping](../adapters/claude-code.md#session-data-mapping) gives the
counts.
+## Execution records
+
+An `execution/1` record says what one observer around a tool call saw it do. The transcript says what
+the model asked for and what came back. It does not say which server ran the call, how long the call
+took, or what the server was sent. The one observer today is the asz Claude Code plugin, whose hook
+runs after every call to an MCP server. Measured on Claude Code 2.1.282:
+
+- The hook after a call names the server and where its configuration came from, the time the
+ runtime measured around the call, and the input as the server received it. A hook before the call
+ that rewrote the input changed what the server received and what this hook sees, while the
+ transcript kept the model's own input.
+- An error the server returned, a lost connection and a timeout all reach the failure hook, not the
+ success hook.
+- A call the runtime refused never reaches a hook, so it has no record.
+- A server's declaration that a tool only reads, or can destroy, reaches neither the model, nor the
+ hooks, nor the transcript.
+
+| Field | Holds |
+| --- | --- |
+| `schema` | `execution/1` |
+| `id` | the observation's own identity: the tool-use id and where it was observed. One call can have several records, each with its own id. The same event handed over twice is one record |
+| `observed_by`, `boundary` | who observed the call, `asz-plugin`, and where it stood, `client_hook`: the runtime's hook after the call |
+| `session`, `stream` | the session, and `main` or the agent id the call ran under |
+| `tool`, `tool_name` | the call's tool-use id, the only join to its step, and the name the runtime called it by |
+| `cwd` | the working directory the runtime reported, which the collector's session filters are judged by |
+| `protocol` | `mcp` |
+| `server` | `name`, the server's configured name, and `source`, where its configuration came from in the runtime's own word: for Claude Code `user`, `project`, `local`, or `dynamic` for a server given on the command line |
+| `time` | when the observation ended |
+| `duration_ms` | the time the runtime measured around the call, which includes any waiting before it started. It is not the server's own time. Absent when the observer did not say |
+| `outcome` | `returned`, `failed` or `interrupted` |
+| `arguments`, `result` | only the `bytes` and the `sha256` of the value as compact JSON with sorted keys, and `state` `size_only`. Two observations of one value have one digest, so a digest that differs from the model's input shows the input was changed on the way. `result` is only on a call that returned |
+
## Dropped
```json
diff --git a/docs/en/formats/storage-root.md b/docs/en/formats/storage-root.md
index 49e9711..076fd88 100644
--- a/docs/en/formats/storage-root.md
+++ b/docs/en/formats/storage-root.md
@@ -18,6 +18,8 @@ session.
transcript-<collected-at>-<seq>.sd
changes.cursor the plugin's change records for the stream, when
changes-<collected-at>-<seq>.sd the asz Claude Code plugin is installed
+ execution.cursor the plugin's execution records for the stream: which
+ execution-<collected-at>-<seq>.sd MCP server each call went to, and how it ended
<agent-id>/ one directory per child agent
transcript.cursor
transcript-<collected-at>-<seq>.sd
@@ -43,11 +45,8 @@ session.
_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
- .lock
- metrics-<received-at>-<seq>-otlp.pb one request the receiver adapter was sent
+ _metrics/ the metrics spool, when metrics are on
+ metrics.state what the derivation has counted
metrics-<session-id>-<seq>-local.pb the points derived from the session's landed file <seq>
push.state what has been sent: each file's digest, the receivers,
and the files a receiver rejected records of
@@ -63,7 +62,7 @@ session.
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
+derive. [OTLP Records](otlp.md) says how the metrics spool fills and
what `push.state` records. `_scenario/` and `_removed/` exist only in a root a scenario build
wrote. [Retention](#retention) says what they are for.
diff --git a/docs/en/guides/scenario.md b/docs/en/guides/scenario.md
index 2b44580..f29ae89 100644
--- a/docs/en/guides/scenario.md
+++ b/docs/en/guides/scenario.md
@@ -150,7 +150,7 @@ A duration counts from the session's last record, never from when it landed, so
data always gives the same answer. A feed stamps each session so its last record falls when it was
written, so a feed's session stays about that long after it was written. A one-shot build with an
old `--at` is already past its duration, and goes as soon as it is sent. `d` counts whole days, as
-`metrics_lookback` does. One run applies one policy to every session it writes.
+`metrics.lookback` does. One run applies one policy to every session it writes.
The policy belongs to the scenario, never to the product. The configuration has no removal
setting, and a session no build marked, which is every real Claude Code session, is never removed.
diff --git a/docs/en/setup/claude-code-plugin.md b/docs/en/setup/claude-code-plugin.md
index 9207f8e..91190ef 100644
--- a/docs/en/setup/claude-code-plugin.md
+++ b/docs/en/setup/claude-code-plugin.md
@@ -2,8 +2,10 @@
The plugin records which files each tool call changed, including shell commands and edits inside
subagents. asz collects the records, and the conversation page shows each change beside the step
-that made it. [Claude Code Plugin Internals](../adapters/claude-code-plugin.md) explains what is
-recorded and how.
+that made it. The plugin also records each call to an MCP server: which server ran it, how it ended
+and how long it took. The conversation document lists each of these records with its step, and asz
+derives the [MCP metrics](export-otlp.md#metrics) from them.
+[Claude Code Plugin Internals](../adapters/claude-code-plugin.md) explains what is recorded and how.
## Install
@@ -127,6 +129,8 @@ exclude:
remove: []
read_only:
enabled: true
+mcp:
+ enabled: true
tools:
scope: [Bash, PowerShell, Monitor]
retention:
@@ -142,6 +146,7 @@ size_cap: 1048576
adds rules and `remove` drops a default. A rule that starts with `/` is relative to the root; any
other name matches at any depth.
- `read_only.enabled`: skip the scan around a shell command that only reads.
+- `mcp.enabled`: record each call to an MCP server.
- `tools.scope`: the tools watched with a scan.
- `retention.idle`: drop the saved file contents of a root that has been idle this long.
- `retention.ttl`: delete an output file not written for this long.
diff --git a/docs/en/setup/configuration.md b/docs/en/setup/configuration.md
index 10e4c24..1472252 100644
--- a/docs/en/setup/configuration.md
+++ b/docs/en/setup/configuration.md
@@ -25,9 +25,9 @@ A list of the sources asz collects from.
| Adapter | On by default | Collects |
| --- | --- | --- |
| `claude-code-local` | yes | Claude Code's transcripts on this machine |
-| `changes` | yes | what a tool call changed on disk, from the [Claude Code](claude-code-plugin.md) or [LangChain](langchain-plugin.md) plugin; see [below](#the-changes-adapter) |
+| `changes` | yes | what a tool call changed on disk, from the [Claude Code](claude-code-plugin.md) or [LangChain](langchain-plugin.md) plugin, and each call to an MCP server the Claude Code plugin saw; see [below](#the-changes-adapter) |
| `claude-code-provider` | yes | the request and response bodies Claude Code writes when asked; see [below](#the-provider-adapter) |
-| `claude-code-otlp` | no | what Claude Code's own OpenTelemetry exporter sends; see [below](#the-receiver-adapter) |
+| `claude-code-otlp` | no | nothing: it accepts what Claude Code's own OpenTelemetry exporter sends, and keeps none of it; see [below](#the-receiver-adapter) |
| `langsmith-ingest` | no | what the LangSmith tracing client sends, from a LangChain or LangGraph application; see [LangChain and LangGraph](langchain.md) |
### More than one agent
@@ -54,8 +54,6 @@ A list of the sources asz collects from.
| `source_root` | empty | The directory Claude Code keeps its projects in, one directory per project. Empty means `projects` under `CLAUDE_CONFIG_DIR`, else under `XDG_CONFIG_HOME/claude`, else under `~/.claude`. To collect from a copy, name the copy's `projects` directory, not the directory above it. |
| `include` | empty | [Session filters](#session-filters) a session must match. Empty means every session. |
| `exclude` | `/private/tmp/**` | [Session filters](#session-filters) that leave a session out. `exclude: []` clears the default. |
-| `metrics` | `false` | Derive Claude Code's token metrics from the collected data. See [Metrics](export-otlp.md#metrics). |
-| `metrics_lookback` | `24h` | How far back the first derivation reaches, such as `24h` or `7d`. `0` or `none` derives everything. |
| `listen` | none | `claude-code-otlp` only: the address to receive on, such as `127.0.0.1:4317`, gRPC and HTTP on one port. |
### Session filters
@@ -92,6 +90,16 @@ data later.
| --- | --- | --- |
| `max_round_bytes` | `2097152` | The largest round file the parser writes, 2 MiB. |
+## metrics
+
+The metrics asz derives from the collected data, whichever adapter collected it. See
+[Metrics](export-otlp.md#metrics).
+
+| Key | Default | Meaning |
+| --- | --- | --- |
+| `enabled` | `true` | Derive the metrics. |
+| `lookback` | `72h` | How far back the first derivation reaches, such as `72h` or `3d`. `0` or `none` derives everything. Later ones derive everything new. |
+
## export
Where to send the collected data. See [Export over OpenTelemetry](export-otlp.md).
@@ -128,7 +136,8 @@ export:
## The changes adapter
`changes` collects the records `asz-changes` writes: what each tool call changed on disk, for
-the [Claude Code plugin](claude-code-plugin.md) or the [LangChain plugin](langchain-plugin.md).
+the [Claude Code plugin](claude-code-plugin.md) or the [LangChain plugin](langchain-plugin.md),
+and each call to an MCP server the Claude Code plugin saw.
With no `source_root` it reads Claude Code's plugin directory, `plugins/data` under
`CLAUDE_CONFIG_DIR`, else `XDG_CONFIG_HOME/claude`, else `~/.claude`, and does nothing when the
plugin is not installed. A LangChain recorder writes where `ASZ_CHANGES_DATA` points, so that is
@@ -160,15 +169,8 @@ adapters:
- name: claude-code-otlp
enabled: true
listen: 127.0.0.1:4317
- metrics: true
-```
-
-`claude-code-otlp` receives Claude Code's own metrics. Start Claude Code with:
-
-```sh
-CLAUDE_CODE_ENABLE_TELEMETRY=1 OTEL_METRICS_EXPORTER=otlp OTEL_EXPORTER_OTLP_ENDPOINT=http://127.0.0.1:4317 claude
```
-It runs while `asz collect` or `asz server` runs, but not with `-once`. It keeps the metrics for
-[export](#export), and drops logs and traces. Turn `metrics` on here or on `claude-code-local`, not
-on both, since both would count the same tokens.
+`claude-code-otlp` is for a Claude Code already set to send its own telemetry. It accepts metrics,
+logs and traces over gRPC or HTTP, and keeps none of them. The metrics asz sends come from the
+collected data. It runs while `asz collect` or `asz server` runs, but not with `-once`.
diff --git a/docs/en/setup/export-otlp.md b/docs/en/setup/export-otlp.md
index 93bb2c7..447d560 100644
--- a/docs/en/setup/export-otlp.md
+++ b/docs/en/setup/export-otlp.md
@@ -2,8 +2,7 @@
asz can send what it collects to an OpenTelemetry receiver, such as the SkyWalking OAP or an
OpenTelemetry Collector: every collected file and every conversation round as a log record, and
-Claude Code's token usage as metrics. [OTLP Records](../formats/otlp.md) describes what a receiver
-gets.
+the metrics derived from them. [OTLP Records](../formats/otlp.md) describes what a receiver gets.
## Send
@@ -24,13 +23,20 @@ everything not sent yet, and exits. Each file is sent once.
## Metrics
-To send Claude Code's token usage, `claude_code.token.usage`, choose one source:
+asz derives these metrics from what it collects, and sends them with the files:
-- Derive it from the transcripts: set `metrics: true` on the `claude-code-local` adapter.
-- Or receive what Claude Code's own exporter sends, with the
- [receiver adapter](configuration.md#the-receiver-adapter).
+- `agent.token.usage`: the tokens of each Claude Code model call.
+- `agent.mcp.calls` and `agent.mcp.duration`: each call to an MCP server and the time it took.
+ They need the [Claude Code plugin](claude-code-plugin.md).
-Not both, since both would count the same tokens.
+They are on by default. The first run reaches back three days. To change that, or to turn the
+metrics off:
+
+```yaml
+metrics:
+ enabled: true
+ lookback: 72h # 0 or none sends all the history
+```
## Limit the rate
diff --git a/internal/adapters/claudecode/convert.go b/internal/adapters/claudecode/convert.go
index 652016f..5286148 100644
--- a/internal/adapters/claudecode/convert.go
+++ b/internal/adapters/claudecode/convert.go
@@ -238,8 +238,12 @@ func partOf(raw json.RawMessage, tur *toolResult, hasTUR bool) (sessiondata.Part
}
case "tool_use", "server_tool_use":
- return sessiondata.Part{Kind: sessiondata.PartCall, ID: b.ID, Name: b.Name,
- Data: b.Input, State: model.ContentAvailable, Bytes: len(b.Input)}, nil
+ p := sessiondata.Part{Kind: sessiondata.PartCall, ID: b.ID, Name: b.Name,
+ Data: b.Input, State: model.ContentAvailable, Bytes: len(b.Input)}
+ if b.Type == "tool_use" {
+ p.Server, p.ServerTool, _ = MCPName(b.Name)
+ }
+ return p, nil
case "tool_result":
p := sessiondata.Part{Kind: sessiondata.PartResult, Of: b.ToolUseID,
diff --git a/internal/adapters/claudecode/mcpname.go b/internal/adapters/claudecode/mcpname.go
new file mode 100644
index 0000000..82042fa
--- /dev/null
+++ b/internal/adapters/claudecode/mcpname.go
@@ -0,0 +1,46 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package claudecode
+
+import "strings"
+
+// mcpPrefix starts the name Claude Code gives a call to an MCP server.
+const mcpPrefix = "mcp__"
+
+// MCPName splits the name Claude Code gives a call to an MCP server,
+// mcp__<server>__<tool>, into the server and the tool, and reports whether
+// the split is exact.
+//
+// It is exact only when exactly one "__" follows the prefix. Claude Code
+// builds the name by joining the two with "__", so a server or a tool whose
+// own name holds "__" would make two splits possible, and then none is made
+// and the name is left whole. Measured on the local corpus on 2026-09-25, no
+// MCP tool name held a second "__". The server part is Claude Code's form of
+// the configured name: "claude.ai Claude Docs" is called
+// "claude_ai_Claude_Docs", so it is not the configured name itself.
+func MCPName(name string) (server, tool string, ok bool) {
+ rest, found := strings.CutPrefix(name, mcpPrefix)
+ if !found {
+ return "", "", false
+ }
+ server, tool, found = strings.Cut(rest, "__")
+ if !found || server == "" || tool == "" || strings.Contains(tool, "__") {
+ return "", "", false
+ }
+ return server, tool, true
+}
diff --git a/internal/adapters/claudecodechanges/collect.go b/internal/adapters/claudecodechanges/collect.go
index 847d16f..f93c6c2 100644
--- a/internal/adapters/claudecodechanges/collect.go
+++ b/internal/adapters/claudecodechanges/collect.go
@@ -69,9 +69,6 @@ func (s *Stats) Complete() bool {
// maxDrainRounds bounds how many windows one source may land in a pass.
const maxDrainRounds = 64
-// prefix is the landed file name prefix, which is also the kind.
-const prefix = "changes"
-
// New returns a Collector with sensible defaults.
func New(sourceRoot string, zone *storage.Zone, maxDelta int64) *Collector {
if maxDelta <= 0 {
@@ -222,6 +219,9 @@ func (c *Collector) collectSession(s Session, st *Stats) error {
// rather than reading its file against another's position, which is the
// conflict that stopped both for good.
func cursorFor(dir, root string, src Source) (string, error) {
+ // A landed file's name prefix is its kind, and so is its cursor's, so
+ // a stream's change file and execution file keep separate places.
+ prefix := string(src.Kind)
plain := filepath.Join(dir, prefix+".cursor")
held, err := storage.LoadCursor(plain, storage.CursorAppend, src.Rel)
if err != nil {
@@ -329,10 +329,10 @@ func (c *Collector) collectSource(src Source, ix *index.Index, state *storage.Se
}
seq := state.Take()
- name := storage.LandedName(prefix, storage.Stamp(now), seq)
+ name := storage.LandedName(string(src.Kind), storage.Stamp(now), seq)
hdr := &sessiondata.Header{
H: 1, Seq: seq, At: now.UTC().Format(time.RFC3339Nano),
- Kind: sessiondata.KindChanges, Adapter: Name + "/" + Version, Dialect: Dialect,
+ Kind: src.Kind, Adapter: Name + "/" + Version, Dialect: Dialect,
Src: src.Rel, Session: src.Session, Stream: src.Stream,
}
var written int64
@@ -342,7 +342,7 @@ func (c *Collector) collectSource(src Source, ix *index.Index, state *storage.Se
return err
}
for row, ln := range chunk.Lines {
- rec := Convert(ln.Ord, ln.Off, ln.Bytes)
+ rec := ConvertKind(src.Kind, ln.Ord, ln.Off, ln.Bytes)
if err := rw.Write(rec); err != nil {
return err
}
diff --git a/internal/adapters/claudecodechanges/convert.go b/internal/adapters/claudecodechanges/convert.go
index 1f10587..fd7cb1b 100644
--- a/internal/adapters/claudecodechanges/convert.go
+++ b/internal/adapters/claudecodechanges/convert.go
@@ -23,6 +23,7 @@ import (
"encoding/json"
"github.com/apache/skywalking-ai-sessionizer/pkg/changes"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/execution"
"github.com/apache/skywalking-ai-sessionizer/pkg/model"
"github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
)
@@ -59,6 +60,42 @@ func Convert(ord, off uint64, payload []byte) *sessiondata.Record {
return rec
}
+// ConvertKind turns one line into a record of the kind its file holds.
+func ConvertKind(kind sessiondata.Kind, ord, off uint64, payload []byte) *sessiondata.Record {
+ if kind == sessiondata.KindExecution {
+ return ConvertExecution(ord, off, payload)
+ }
+ return Convert(ord, off, payload)
+}
+
+// ConvertExecution turns one execution record the plugin wrote into a
+// Session Data record, the same framing as a change record: its identity,
+// the call it joins to and its time lifted onto the record, and the line
+// kept whole, byte for byte, as one data part.
+func ConvertExecution(ord, off uint64, payload []byte) *sessiondata.Record {
+ sum := sha256.Sum256(payload)
+ rec := &sessiondata.Record{
+ Ord: ord, Off: off,
+ Sha: hex.EncodeToString(sum[:])[:12],
+ Bytes: len(payload),
+ }
+ r, ok := execution.Decode(payload)
+ if !ok {
+ rec.Parts = []sessiondata.Part{unknownPart(payload, "the line is not an execution/1 record")}
+ return rec
+ }
+ if err := r.Validate(); err != nil {
+ rec.Parts = []sessiondata.Part{unknownPart(payload, err.Error())}
+ return rec
+ }
+ rec.ID, rec.Tool, rec.Time = r.ID, r.Tool, r.Time
+ rec.Parts = []sessiondata.Part{{
+ Kind: sessiondata.PartData, Data: json.RawMessage(payload),
+ State: model.ContentAvailable, Bytes: len(payload),
+ }}
+ return rec
+}
+
// unknownPart keeps a line the adapter could not read, every byte of it.
// SetRaw is what keeps a byte that is not valid UTF-8, which a plain JSON
// string would replace. A write cut short and joined to the next line is
diff --git a/internal/adapters/claudecodechanges/discover.go b/internal/adapters/claudecodechanges/discover.go
index 20d5bd0..deadbbc 100644
--- a/internal/adapters/claudecodechanges/discover.go
+++ b/internal/adapters/claudecodechanges/discover.go
@@ -32,6 +32,8 @@ import (
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecode"
"github.com/apache/skywalking-ai-sessionizer/pkg/changes"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/execution"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
)
// Source is one file the plugin writes: the records of one stream of one
@@ -41,6 +43,9 @@ type Source struct {
Rel string // relative to the adapter's source root; kept in the landed header
Session string
Stream string // main or an agent id
+ // Kind is what the file holds, and what it lands as: change records,
+ // or execution records, which the plugin writes to a file of their own.
+ Kind sessiondata.Kind
}
// Session is one Claude Code session with every change file the plugin
@@ -158,7 +163,7 @@ func DiscoverWithWarnings(root string) (sessions []Session, warnings []error, er
continue
}
for _, f := range files {
- stream, ok := strings.CutSuffix(f.Name(), ".jsonl")
+ stream, kind, ok := streamOf(f.Name())
if f.IsDir() || !ok || (stream != "main" && !claudecode.IsAgentID(stream)) {
continue
}
@@ -169,7 +174,7 @@ func DiscoverWithWarnings(root string) (sessions []Session, warnings []error, er
}
abs := filepath.Join(out, id, f.Name())
s.Sources = append(s.Sources, Source{
- Path: abs, Rel: path.Join(name, "output", id, f.Name()), Session: id, Stream: stream,
+ Path: abs, Rel: path.Join(name, "output", id, f.Name()), Session: id, Stream: stream, Kind: kind,
})
if s.Root == "" {
s.Root = firstRoot(abs)
@@ -185,8 +190,20 @@ func DiscoverWithWarnings(root string) (sessions []Session, warnings []error, er
return sessions, warnings, nil
}
+// streamOf reads a file's stream and kind from its name: <stream>.jsonl for
+// change records, <stream>.execution.jsonl for execution records.
+func streamOf(name string) (string, sessiondata.Kind, bool) {
+ if stream, ok := strings.CutSuffix(name, ".execution.jsonl"); ok {
+ return stream, sessiondata.KindExecution, true
+ }
+ stream, ok := strings.CutSuffix(name, ".jsonl")
+ return stream, sessiondata.KindChanges, ok
+}
+
// firstRoot reads the workspace path the first complete record names, or
-// nothing when the file has no complete record yet.
+// nothing when the file has no complete record yet. A change record names
+// the root it observed; an execution record, the working directory the
+// runtime reported.
func firstRoot(path string) string {
f, err := os.Open(path)
if err != nil {
@@ -201,6 +218,9 @@ func firstRoot(path string) string {
if r, ok := changes.Decode(line); ok && r.Root != nil {
return r.Root.Path
}
+ if r, ok := execution.Decode(line); ok {
+ return r.Cwd
+ }
return ""
}
diff --git a/internal/adapters/claudecodeotlp/receiver.go b/internal/adapters/claudecodeotlp/receiver.go
index 5fb24eb..30e3f37 100644
--- a/internal/adapters/claudecodeotlp/receiver.go
+++ b/internal/adapters/claudecodeotlp/receiver.go
@@ -16,13 +16,14 @@
// under the License.
// Package claudecodeotlp is the claude-code-otlp adapter: an OpenTelemetry
-// receiver Claude Code's own exporter is pointed at. Phase one takes the
-// runtime's metrics and lands each request in the storage root's metrics
-// spool, bytes as received, for asz push to send on under asz's identity.
-// Logs and traces are accepted and dropped, counted so the status says so,
-// which keeps a runtime configured with every exporter working. Nothing
-// here reads a transcript: the two adapters for the one runtime meet only
-// in the spool, and the configuration refuses to run both with metrics on.
+// receiver Claude Code's own exporter can be pointed at. It accepts metrics,
+// logs and traces, counts each request so the status says what the runtime
+// sends, and keeps none of them. A runtime configured with an exporter keeps
+// working, and nothing it sends is counted twice.
+//
+// asz does not land the exporter's metrics. Every metric asz sends is derived
+// from the landed files, under one name for every runtime, so one root has
+// one source for each count.
package claudecodeotlp
import (
@@ -32,6 +33,7 @@ import (
"io"
"net"
"net/http"
+ "strings"
"sync"
"sync/atomic"
"time"
@@ -40,60 +42,35 @@ import (
collmetricspb "go.opentelemetry.io/proto/otlp/collector/metrics/v1"
colltracepb "go.opentelemetry.io/proto/otlp/collector/trace/v1"
"google.golang.org/grpc"
- "google.golang.org/grpc/codes"
- "google.golang.org/grpc/status"
- "google.golang.org/protobuf/proto"
-
- "github.com/apache/skywalking-ai-sessionizer/internal/storage"
)
const (
// Name is the adapter's name in the configuration.
Name = "claude-code-otlp"
- // RuntimeName is the runtime the requests concern, the same service the
- // local adapter's sessions are attributed to.
- RuntimeName = "Claude Code"
// Version is the adapter's contract version.
Version = "0.1.0"
- // SourceOTLP names the spool files this adapter writes.
- SourceOTLP = "otlp"
// MaxRequest bounds one request, well above what the runtime's exporter
// sends in a minute.
MaxRequest = 64 << 20
)
-// Stats counts what the receiver did since it started.
+// Stats counts the requests accepted and dropped since the receiver started.
type Stats struct {
- Metrics int64 // metrics requests landed in the spool
- // MetricsDropped counts metrics requests accepted and dropped because
- // the adapter has metrics off: the local adapter derives them instead.
- MetricsDropped int64
- Logs int64 // logs requests accepted and dropped
- Traces int64 // traces requests accepted and dropped
- Bytes int64 // metrics bytes landed
- Errors int64
+ Metrics int64
+ Logs int64
+ Traces int64
}
// Receiver listens for the runtime's exporter on one address, over gRPC
-// and over HTTP with protobuf, the two protocols the runtime offers, and
-// lands what it receives.
+// and over HTTP with protobuf, the two protocols the runtime offers.
type Receiver struct {
- Zone *storage.Zone
// Listen is the address, such as 127.0.0.1:4317. Both protocols share
// it: gRPC and HTTP/1.1 are told apart on the first bytes.
Listen string
- // LandMetrics says metrics are landed in the spool. Off, they are
- // accepted and dropped, for a root whose local adapter derives them.
- LandMetrics bool
- Now func() time.Time
- spool *storage.Spool
- metricsDropped atomic.Int64
- metrics atomic.Int64
- logs atomic.Int64
- traces atomic.Int64
- bytes atomic.Int64
- errs atomic.Int64
+ metrics atomic.Int64
+ logs atomic.Int64
+ traces atomic.Int64
mu sync.Mutex
ln net.Listener
@@ -104,7 +81,7 @@ type Receiver struct {
// Stats reports the counts so far.
func (r *Receiver) Stats() Stats {
- return Stats{Metrics: r.metrics.Load(), MetricsDropped: r.metricsDropped.Load(), Logs: r.logs.Load(), Traces: r.traces.Load(), Bytes: r.bytes.Load(), Errors: r.errs.Load()}
+ return Stats{Metrics: r.metrics.Load(), Logs: r.logs.Load(), Traces: r.traces.Load()}
}
// Addr is the address the receiver listens on, once started.
@@ -121,13 +98,9 @@ func (r *Receiver) Addr() string {
// both protocols: a connection that opens with an HTTP/2 preface is gRPC,
// anything else is HTTP/1.1.
func (r *Receiver) Start() error {
- if r.Now == nil {
- r.Now = time.Now
- }
if r.Listen == "" {
return errors.New("claude-code-otlp: no listen address")
}
- r.spool = storage.NewSpool(r.Zone)
ln, err := net.Listen("tcp", r.Listen)
if err != nil {
return fmt.Errorf("claude-code-otlp: %w", err)
@@ -137,7 +110,7 @@ func (r *Receiver) Start() error {
collogspb.RegisterLogsServiceServer(r.rpc, &logsServer{r: r})
colltracepb.RegisterTraceServiceServer(r.rpc, &traceServer{r: r})
mux := http.NewServeMux()
- mux.HandleFunc("/v1/metrics", r.serveMetrics)
+ mux.HandleFunc("/v1/metrics", r.serveDropped(&r.metrics))
mux.HandleFunc("/v1/logs", r.serveDropped(&r.logs))
mux.HandleFunc("/v1/traces", r.serveDropped(&r.traces))
r.web = &http.Server{Handler: mux, ReadHeaderTimeout: 10 * time.Second}
@@ -191,58 +164,20 @@ func (r *Receiver) Stop() {
<-done
}
-// land writes one metrics request to the spool, bytes as received, or
-// drops it when the adapter has metrics off.
-func (r *Receiver) land(data []byte) error {
- if !r.LandMetrics {
- r.metricsDropped.Add(1)
- return nil
- }
- if _, err := r.spool.Put(SourceOTLP, data, r.Now()); err != nil {
- r.errs.Add(1)
- return err
- }
- r.metrics.Add(1)
- r.bytes.Add(int64(len(data)))
- return nil
-}
-
-func (r *Receiver) serveMetrics(w http.ResponseWriter, req *http.Request) {
- if req.Method != http.MethodPost {
- http.Error(w, "POST a metrics request", http.StatusMethodNotAllowed)
- return
- }
- if ct := req.Header.Get("Content-Type"); ct != "application/x-protobuf" {
- // The runtime can also send http/json; phase one lands protobuf
- // only, which is what the runtime's default protocol sends.
- http.Error(w, "content type "+ct+" is not application/x-protobuf; set OTEL_EXPORTER_OTLP_PROTOCOL to grpc or http/protobuf", http.StatusUnsupportedMediaType)
- return
- }
- data, err := io.ReadAll(io.LimitReader(req.Body, MaxRequest+1))
- if err != nil || len(data) > MaxRequest {
- http.Error(w, "the request is too large or could not be read", http.StatusRequestEntityTooLarge)
- return
- }
- var msg collmetricspb.ExportMetricsServiceRequest
- if err := proto.Unmarshal(data, &msg); err != nil {
- http.Error(w, "not an ExportMetricsServiceRequest: "+err.Error(), http.StatusBadRequest)
- return
- }
- if err := r.land(data); err != nil {
- http.Error(w, err.Error(), http.StatusServiceUnavailable)
- return
- }
- out, _ := proto.Marshal(&collmetricspb.ExportMetricsServiceResponse{})
- w.Header().Set("Content-Type", "application/x-protobuf")
- _, _ = w.Write(out)
-}
-
-// serveDropped accepts a logs or traces request and keeps nothing of it,
-// counting it so the status can say what the runtime is sending.
+// serveDropped accepts a request and keeps nothing of it, counting it so the
+// status can say what the runtime is sending. The answer is an empty export
+// response in the request's own encoding, as OTLP asks: no bytes in
+// protobuf, and {} in JSON.
func (r *Receiver) serveDropped(counter *atomic.Int64) http.HandlerFunc {
return func(w http.ResponseWriter, req *http.Request) {
_, _ = io.Copy(io.Discard, io.LimitReader(req.Body, MaxRequest))
counter.Add(1)
+ if strings.HasPrefix(req.Header.Get("Content-Type"), "application/json") {
+ w.Header().Set("Content-Type", "application/json")
+ w.WriteHeader(http.StatusOK)
+ _, _ = w.Write([]byte("{}"))
+ return
+ }
w.Header().Set("Content-Type", "application/x-protobuf")
w.WriteHeader(http.StatusOK)
}
@@ -253,17 +188,8 @@ type metricsServer struct {
r *Receiver
}
-func (s *metricsServer) Export(_ context.Context, req *collmetricspb.ExportMetricsServiceRequest) (*collmetricspb.ExportMetricsServiceResponse, error) {
- // Landed as the bytes of the message as received: the runtime's
- // request, re-encoded from the decoded message, is the same bytes for
- // the same message, and what push sends is what the spool holds.
- data, err := proto.Marshal(req)
- if err != nil {
- return nil, status.Error(codes.InvalidArgument, err.Error())
- }
- if err := s.r.land(data); err != nil {
- return nil, status.Error(codes.Unavailable, err.Error())
- }
+func (s *metricsServer) Export(context.Context, *collmetricspb.ExportMetricsServiceRequest) (*collmetricspb.ExportMetricsServiceResponse, error) {
+ s.r.metrics.Add(1)
return &collmetricspb.ExportMetricsServiceResponse{}, nil
}
diff --git a/internal/adapters/claudecodeotlp/receiver_test.go b/internal/adapters/claudecodeotlp/receiver_test.go
index 91e5bdb..9e135ec 100644
--- a/internal/adapters/claudecodeotlp/receiver_test.go
+++ b/internal/adapters/claudecodeotlp/receiver_test.go
@@ -19,8 +19,8 @@ package claudecodeotlp_test
import (
"bytes"
+ "io"
"net/http"
- "os"
"testing"
"time"
@@ -29,12 +29,9 @@ import (
commonpb "go.opentelemetry.io/proto/otlp/common/v1"
metricspb "go.opentelemetry.io/proto/otlp/metrics/v1"
resourcepb "go.opentelemetry.io/proto/otlp/resource/v1"
- "google.golang.org/protobuf/proto"
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecodeotlp"
"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/storage"
)
func request() *collmetricspb.ExportMetricsServiceRequest {
@@ -47,18 +44,17 @@ func request() *collmetricspb.ExportMetricsServiceRequest {
}}}
}
-// The receiver takes the runtime's metrics over either protocol on one
-// port and lands each request in the spool, bytes as sent; logs are
-// accepted, counted and dropped.
-func TestReceiverLandsMetricsFromBothProtocols(t *testing.T) {
- z := storage.NewZone(t.TempDir())
- rcv := &claudecodeotlp.Receiver{Zone: z, Listen: "127.0.0.1:0", LandMetrics: true, Now: func() time.Time { return time.Date(2026, 9, 7, 0, 0, 0, 0, time.UTC) }}
+// The receiver takes what the runtime's exporter sends over either protocol
+// on one port, answers each request as accepted, and counts it. It keeps
+// nothing: asz derives every metric it sends from the landed files, so a
+// runtime that also exports must not reach the spool.
+func TestReceiverAcceptsAndDropsBothProtocols(t *testing.T) {
+ rcv := &claudecodeotlp.Receiver{Listen: "127.0.0.1:0"}
if err := rcv.Start(); err != nil {
t.Fatal(err)
}
defer rcv.Stop()
addr := rcv.Addr()
- want, _ := proto.Marshal(request())
for _, o := range []otlp.Options{
{Protocol: otlp.ProtocolGRPC, Endpoint: addr, Timeout: 10 * time.Second},
@@ -69,123 +65,26 @@ func TestReceiverLandsMetricsFromBothProtocols(t *testing.T) {
t.Fatal(err)
}
if _, err := c.ExportMetrics(request()); err != nil {
- t.Fatalf("%s: %v", o.Protocol, err)
+ t.Fatalf("%s: a metrics request must be accepted and dropped: %v", o.Protocol, err)
}
if _, err := c.Export(&collogspb.ExportLogsServiceRequest{}); err != nil {
t.Fatalf("%s: a logs request must be accepted and dropped: %v", o.Protocol, err)
}
_ = c.Close()
}
- files, err := storage.NewSpool(z).List()
- if err != nil {
- t.Fatal(err)
- }
- if len(files) != 2 {
- t.Fatalf("the spool holds %d files, want one per metrics request", len(files))
- }
- for _, f := range files {
- data, err := os.ReadFile(f.Path)
- if err != nil {
- t.Fatal(err)
- }
- if !bytes.Equal(data, want) || f.Source != claudecodeotlp.SourceOTLP {
- t.Fatalf("%s: not the request as sent, or not marked as the runtime's", f.Path)
- }
- if fi, _ := os.Stat(f.Path); fi.Mode().Perm()&0o222 != 0 {
- t.Fatalf("%s is writable; a spooled request is write-once", f.Path)
- }
- }
- st := rcv.Stats()
- if st.Metrics != 2 || st.Logs != 2 || st.Traces != 0 || st.Bytes != int64(2*len(want)) || st.Errors != 0 {
- t.Fatalf("stats %+v", st)
- }
-
- // A JSON body is refused with a word about the protocol to set.
- resp, err := http.Post("http://"+addr+"/v1/metrics", "application/json", bytes.NewReader([]byte("{}")))
+ // A JSON body is accepted too: nothing is decoded, so its encoding does
+ // not matter, and a runtime set to http/json keeps working. The answer is
+ // in JSON, as OTLP asks.
+ resp, err := http.Post("http://"+addr+"/v1/traces", "application/json", bytes.NewReader([]byte("{}")))
if err != nil {
t.Fatal(err)
}
+ body, _ := io.ReadAll(resp.Body)
resp.Body.Close()
- if resp.StatusCode != http.StatusUnsupportedMediaType {
- t.Fatalf("a JSON request answered %d", resp.StatusCode)
- }
-}
-
-// With metrics off the receiver accepts the runtime's metrics and keeps
-// nothing, so a root whose local adapter derives them counts them once.
-func TestReceiverWithMetricsOffDropsThem(t *testing.T) {
- z := storage.NewZone(t.TempDir())
- rcv := &claudecodeotlp.Receiver{Zone: z, Listen: "127.0.0.1:0"}
- if err := rcv.Start(); err != nil {
- t.Fatal(err)
- }
- defer rcv.Stop()
- c, err := otlp.NewClient(otlp.Options{Protocol: otlp.ProtocolGRPC, Endpoint: rcv.Addr(), Timeout: 10 * time.Second})
- if err != nil {
- t.Fatal(err)
- }
- defer c.Close()
- if _, err := c.ExportMetrics(request()); err != nil {
- t.Fatal(err)
- }
- if files, _ := storage.NewSpool(z).List(); len(files) != 0 {
- t.Fatalf("the spool holds %d files; metrics off must land nothing", len(files))
+ if resp.StatusCode != http.StatusOK || resp.Header.Get("Content-Type") != "application/json" || string(body) != "{}" {
+ t.Fatalf("a JSON request answered %d %q %q", resp.StatusCode, resp.Header.Get("Content-Type"), body)
}
- if st := rcv.Stats(); st.Metrics != 0 || st.MetricsDropped != 1 {
+ if st := rcv.Stats(); st.Metrics != 2 || st.Logs != 2 || st.Traces != 1 {
t.Fatalf("stats %+v", st)
}
}
-
-// The whole path of the runtime's own metrics: sent by the exporter to the
-// receiver, landed in the spool, sent on by asz push under asz's identity
-// with the points untouched, and sent once.
-func TestReceivedMetricsReachTheReceiverOfThePush(t *testing.T) {
- z := storage.NewZone(t.TempDir())
- rcv := &claudecodeotlp.Receiver{Zone: z, Listen: "127.0.0.1:0", LandMetrics: true}
- if err := rcv.Start(); err != nil {
- t.Fatal(err)
- }
- defer rcv.Stop()
- runtime, err := otlp.NewClient(otlp.Options{Protocol: otlp.ProtocolGRPC, Endpoint: rcv.Addr(), Timeout: 10 * time.Second})
- if err != nil {
- t.Fatal(err)
- }
- defer runtime.Close()
- if _, err := runtime.ExportMetrics(request()); err != nil {
- t.Fatal(err)
- }
-
- oap, err := otlptest.Start()
- if err != nil {
- t.Fatal(err)
- }
- defer oap.Close()
- for _, protocol := range []string{otlp.ProtocolGRPC, otlp.ProtocolHTTP} {
- client, err := oap.Client(protocol)
- if err != nil {
- t.Fatal(err)
- }
- p := &otlp.Pusher{Zone: z, Client: client, Version: "test", ServiceName: "Claude Code", InstanceID: "sender-1", Layer: "AI_AGENT"}
- st, err := p.Pass()
- if err != nil || len(st.Errors) != 0 {
- t.Fatal(err, st.Errors)
- }
- _ = client.Close()
- got := oap.MetricsRequests()
- if protocol == otlp.ProtocolGRPC {
- if st.Metrics != 1 || len(got) != 1 {
- t.Fatalf("%s: sent %d, the receiver holds %d", protocol, st.Metrics, len(got))
- }
- res := otlptest.Attrs(got[0].GetResourceMetrics()[0].GetResource().GetAttributes())
- if res["service.name"] != "Claude Code" || res["service.instance.id"] != "sender-1" || res["service.layer"] != "AI_AGENT" {
- t.Fatalf("resource on the wire: %v", res)
- }
- m := got[0].GetResourceMetrics()[0].GetScopeMetrics()[0].GetMetrics()[0]
- if m.GetName() != "claude_code.token.usage" || m.GetSum().GetDataPoints()[0].GetAsInt() != 42 {
- t.Fatalf("the runtime's points did not travel unchanged: %v", m)
- }
- } else if st.Metrics != 0 || len(got) != 1 {
- t.Fatalf("a second pass over the other transport sent the request again: %d, %d", st.Metrics, len(got))
- }
- }
-}
diff --git a/internal/assemble/provider.go b/internal/assemble/provider.go
index 2afda8a..04125a3 100644
--- a/internal/assemble/provider.go
+++ b/internal/assemble/provider.go
@@ -331,12 +331,13 @@ func (b *builder) entryAt(r sessionflow.Ref) *index.Entry {
// transcriptKind reports whether a record came from a stream's transcript, whose
// lines are numbered as one run. A sidecar, a journal, a manifest, a script, a
-// change record and a provider body are each their own file, with their own line
-// numbers, so counting them together would find gaps that are not there.
+// change record, an execution record and a provider body are each their own
+// file, with their own line numbers, so counting them together would find gaps
+// that are not there.
func transcriptKind(k index.Kind) bool {
switch k {
case index.KindMeta, index.KindJournal, index.KindManifest, index.KindScript,
- index.KindChanges, index.KindProviderBody:
+ index.KindChanges, index.KindProviderBody, index.KindExecution:
return false
}
return true
diff --git a/internal/assemble/streams.go b/internal/assemble/streams.go
index dd0f74a..1734f4f 100644
--- a/internal/assemble/streams.go
+++ b/internal/assemble/streams.go
@@ -50,14 +50,15 @@ func (b *builder) stage1Canonical() {
// record count and time range, so a session with such records would
// 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.
+ // joined to the call by a view, and no step of any stream. So is an
+ // execution record, which says what an observer saw a call do.
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 || e.Kind == index.KindProviderBody {
+ if e.Kind == index.KindChanges || e.Kind == index.KindProviderBody || e.Kind == index.KindExecution {
continue
}
bounded = append(bounded, i)
diff --git a/internal/assemble/tools.go b/internal/assemble/tools.go
index 00e78a1..c6799af 100644
--- a/internal/assemble/tools.go
+++ b/internal/assemble/tools.go
@@ -35,12 +35,15 @@ import (
type toolUse struct {
ToolID uint32
Name string
- Use *index.Entry
- UseOrd uint16
- Result *index.Entry
- ResOrd uint16
- Stream *streamInfo
- NodeID string
+ // Server and ServerTool are the MCP server and tool the call's name
+ // addresses, where the adapter could split the name exactly.
+ Server, ServerTool string
+ Use *index.Entry
+ UseOrd uint16
+ Result *index.Entry
+ ResOrd uint16
+ Stream *streamInfo
+ NodeID string
// Ambiguous means several results carry this id. An exact identifier match
// does not guarantee a unique match, and where it is not unique the
@@ -82,6 +85,7 @@ func (b *builder) stage4Tools() {
case index.BlockToolUse:
if t.Use == nil {
t.Use, t.UseOrd, t.Name = e, blk.Ord, b.str(blk.Name)
+ t.Server, t.ServerTool = b.str(blk.Server), b.str(blk.ServerTool)
}
case index.BlockToolResult:
if t.Result != nil {
@@ -146,6 +150,13 @@ func (b *builder) emitTool(t *toolUse, parent string) {
// The request lives in the tool_use block; the reference below locates it.
"timing": model.Unavailable,
}
+ // A call to an MCP server names the server and the tool, as the adapter
+ // read them from the call's name. They are the runtime's names, which
+ // can differ from the configured ones; which server ran the call is an
+ // observer's to say, in an execution record.
+ if t.Server != "" {
+ a["mcp_server"], a["mcp_tool"] = t.Server, t.ServerTool
+ }
refs := []sessionflow.Ref{blockRef(t.Use, t.UseOrd)}
quality := model.Unresolved
switch {
diff --git a/internal/config/config.go b/internal/config/config.go
index fdb8547..5853764 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -35,9 +35,35 @@ type Config struct {
Storage Storage `yaml:"storage"`
Adapters []Adapter `yaml:"adapters"`
Parse Parse `yaml:"parse"`
+ Metrics Metrics `yaml:"metrics"`
Export Export `yaml:"export"`
}
+// Metrics configures what asz derives from the landed files. It is one
+// switch for the root, whichever adapters landed the files: every metric
+// asz defines is derived from every landed file it applies to, the token
+// usage of a Claude Code conversation and the calls the plugin saw reach an
+// MCP server alike. export.otlp.metrics says whether a push sends them.
+type Metrics struct {
+ // Enabled derives the metrics. On unless set to false.
+ Enabled *bool `yaml:"enabled"`
+ // Lookback bounds the first derivation over a root, the one pass that
+ // runs before the root has any metrics state: a minute older than this
+ // is not derived, so a new deployment over months of history does not
+ // send all of it at once. Every later pass derives each new file whole.
+ // A duration such as 72h or 3d; empty means 72h; 0 or none means
+ // everything.
+ Lookback string `yaml:"lookback"`
+}
+
+// DefaultLookback is three days. A new deployment usually starts cold over a
+// source that already holds history, and three days is enough to fill a
+// dashboard without sending everything the source keeps.
+const DefaultLookback = 72 * time.Hour
+
+// On reports whether metrics are derived.
+func (m Metrics) On() bool { return m.Enabled == nil || *m.Enabled }
+
// Parse configures assembly into rounds.
type Parse struct {
// MaxRoundBytes caps a round file. A round travels whole as one log
@@ -103,27 +129,27 @@ func (o OTLP) SendMetrics() bool { return o.Metrics == nil || *o.Metrics }
func boolPtr(b bool) *bool { return &b }
-// Lookback is the metrics look-back as a duration: 24h when unset, zero
-// for 0 or none, and a d suffix counts days, since a look-back is spoken of
-// in days.
-func (a Adapter) Lookback() (time.Duration, error) {
- s := strings.TrimSpace(a.MetricsLookback)
+// LookbackDuration is the look-back as a duration: 72h when unset, zero for
+// 0 or none, and a d suffix counts days, since a look-back is spoken of in
+// days.
+func (m Metrics) LookbackDuration() (time.Duration, error) {
+ s := strings.TrimSpace(m.Lookback)
switch s {
case "":
- return 24 * time.Hour, nil
+ return DefaultLookback, nil
case "0", "none":
return 0, nil
}
if days, ok := strings.CutSuffix(s, "d"); ok {
n, err := strconv.Atoi(days)
if err != nil || n < 0 {
- return 0, fmt.Errorf("config: metrics_lookback %q is not a number of days", s)
+ return 0, fmt.Errorf("config: metrics.lookback %q is not a number of days", s)
}
return time.Duration(n) * 24 * time.Hour, nil
}
d, err := time.ParseDuration(s)
if err != nil || d < 0 {
- return 0, fmt.Errorf("config: metrics_lookback %q is not a duration such as 24h or 7d", s)
+ return 0, fmt.Errorf("config: metrics.lookback %q is not a duration such as 24h or 7d", s)
}
return d, nil
}
@@ -156,13 +182,6 @@ type Adapter struct {
Include []string `yaml:"include"`
Exclude []string `yaml:"exclude"`
- // Metrics, on claude-code-local, turns on the runtime's own metric
- // family derived from the landed files: the same names and attributes
- // the runtime's exporter sends, so a receiver sees one family whichever
- // produced it. Phase one is token usage. On claude-code-otlp it says
- // the received metrics are landed and pushed. It cannot be on for both
- // at once, since the two would count the same tokens twice.
- Metrics bool `yaml:"metrics"`
// Listen, on claude-code-otlp, is the address the receiver listens on
// for the runtime's exporter, such as 127.0.0.1:4317, over gRPC and
// HTTP with protobuf on the one port.
@@ -192,10 +211,6 @@ type Adapter struct {
// it was. The Claude Code provider adapter is its own adapter and does
// not take this.
ProviderBodies bool `yaml:"provider_bodies"`
- // MetricsLookback bounds the first derivation over a root that has
- // history: a minute older than this is not derived. A duration such as
- // 24h or 7d; empty means 24h; 0 or none means everything.
- MetricsLookback string `yaml:"metrics_lookback"`
Collector Collector `yaml:"collector"`
}
@@ -286,9 +301,6 @@ func Default() *Config {
Name: AdapterClaudeCodeLocal,
Enabled: true,
Exclude: []string{"/private/tmp/**"},
- // The look-back is written out, as every default is, so the
- // file says what the first derivation reaches back to.
- MetricsLookback: "24h",
Collector: Collector{
Mode: ModeWatch,
Interval: DefaultInterval,
@@ -296,12 +308,10 @@ func Default() *Config {
},
}, {
// The runtime's own exporter, received. Off until pointed at:
- // the file names the address the runtime would be given, and
- // metrics on here means the local adapter leaves them to it.
+ // the file names the address the runtime would be given.
Name: AdapterClaudeCodeOTLP,
Enabled: false,
Listen: "127.0.0.1:4317",
- Metrics: true,
}, {
// The LangSmith tracing client, received. Off until pointed
// at: the file names the address an application would be
@@ -340,6 +350,9 @@ func Default() *Config {
},
}},
Parse: Parse{MaxRoundBytes: 2 << 20},
+ // Written out, as every default is, so the file says what the first
+ // derivation reaches back to.
+ Metrics: Metrics{Enabled: boolPtr(true), Lookback: "72h"},
Export: Export{OTLP: OTLP{
Protocol: "grpc",
Layer: "AI_AGENT",
@@ -406,6 +419,12 @@ func Load(path string) (*Config, error) {
if loaded.Parse.MaxRoundBytes > 0 {
cfg.Parse.MaxRoundBytes = loaded.Parse.MaxRoundBytes
}
+ if loaded.Metrics.Enabled != nil {
+ cfg.Metrics.Enabled = loaded.Metrics.Enabled
+ }
+ if loaded.Metrics.Lookback != "" {
+ cfg.Metrics.Lookback = loaded.Metrics.Lookback
+ }
o := &loaded.Export.OTLP
if o.Protocol != "" {
cfg.Export.OTLP.Protocol = o.Protocol
@@ -467,7 +486,6 @@ func (c *Config) Validate() error {
// other adapter is one source, and an old name for an adapter is that
// adapter: two entries that read one directory would land it twice.
changesRoots := map[string]string{}
- localMetrics, receiverMetrics := false, false
for i, a := range c.Adapters {
if a.Name == "" {
return fmt.Errorf("config: adapters[%d] has no name", i)
@@ -494,10 +512,6 @@ func (c *Config) Validate() error {
if a.Listen == "" {
return fmt.Errorf("config: adapter %q needs listen, the address the runtime's exporter is pointed at, such as 127.0.0.1:4317", a.Name)
}
- receiverMetrics = receiverMetrics || a.Metrics
- }
- if a.Name == AdapterClaudeCodeLocal && a.Enabled && a.Metrics {
- localMetrics = true
}
if a.Name != AdapterLangSmithIngest && a.ProviderBodies {
return fmt.Errorf("config: adapter %q does not take provider_bodies; only %q does, and %q lands Claude Code's bodies on its own", a.Name, AdapterLangSmithIngest, AdapterClaudeCodeProvider)
@@ -506,9 +520,8 @@ func (c *Config) Validate() error {
if a.Enabled && a.Listen == "" {
return fmt.Errorf("config: adapter %q needs listen, the address LANGSMITH_ENDPOINT is pointed at, such as 127.0.0.1:1985", a.Name)
}
- if a.Metrics || a.MetricsLookback != "" || a.SourceRoot != "" ||
- len(a.Include) > 0 || len(a.Exclude) > 0 {
- return fmt.Errorf("config: adapter %q is a receiver: it takes listen, token, thread_keys, scope, provider_bodies and collector, not source_root, include, exclude, metrics or metrics_lookback", a.Name)
+ if a.SourceRoot != "" || len(a.Include) > 0 || len(a.Exclude) > 0 {
+ return fmt.Errorf("config: adapter %q is a receiver: it takes listen, token, thread_keys, scope, provider_bodies and collector, not source_root, include or exclude", a.Name)
}
// It takes a period and a byte budget, unlike the metrics
// receiver: what it accepts waits in an inbox until a pass
@@ -524,23 +537,20 @@ func (c *Config) Validate() error {
continue
}
if a.Name == AdapterClaudeCodeOTLP {
- if a.Collector != (Collector{}) || a.SourceRoot != "" || len(a.Include) > 0 || len(a.Exclude) > 0 || a.MetricsLookback != "" {
- return fmt.Errorf("config: adapter %q is a receiver: it takes listen and metrics, not collector, source_root, include, exclude or metrics_lookback", a.Name)
+ if a.Collector != (Collector{}) || a.SourceRoot != "" || len(a.Include) > 0 || len(a.Exclude) > 0 {
+ return fmt.Errorf("config: adapter %q is a receiver: it takes listen, not collector, source_root, include or exclude", a.Name)
}
continue
}
- if isChanges(a.Name) && (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 isChanges(a.Name) && a.Listen != "" {
+ return fmt.Errorf("config: adapter %q reads the plugin's records only: it takes source_root, include, exclude and collector, not 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.Name == AdapterClaudeCodeProvider && a.Listen != "" {
+ return fmt.Errorf("config: adapter %q reads provider bodies only: it takes source_root, include, exclude and collector, not 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)
}
- if _, err := a.Lookback(); err != nil {
- return fmt.Errorf("adapter %q: %w", a.Name, err)
- }
}
if p := c.Export.OTLP.Protocol; p != "grpc" && p != "http" {
return fmt.Errorf("config: export.otlp.protocol is %q, want grpc or http", p)
@@ -548,11 +558,8 @@ func (c *Config) Validate() error {
if !c.Export.OTLP.SendLogs() && !c.Export.OTLP.SendMetrics() {
return errors.New("config: export.otlp has both logs and metrics off; a push would send nothing")
}
- // The same tokens must not be counted twice: derived from the
- // transcripts and received from the runtime's exporter are two sources
- // of one metric, and a root sends one of them.
- if localMetrics && receiverMetrics {
- return fmt.Errorf("config: %s has metrics on while %s is enabled with metrics; the two would count the same tokens twice, so turn one off", AdapterClaudeCodeLocal, AdapterClaudeCodeOTLP)
+ if _, err := c.Metrics.LookbackDuration(); err != nil {
+ return err
}
return nil
}
diff --git a/internal/config/config_test.go b/internal/config/config_test.go
index 90580f6..51c819d 100644
--- a/internal/config/config_test.go
+++ b/internal/config/config_test.go
@@ -102,10 +102,9 @@ func TestRepoConfigSpellsOutEveryValue(t *testing.T) {
}
}
-// The receiver adapter needs an address, and the same tokens are never
-// counted twice: local derivation and the runtime's exporter cannot both
-// have metrics on.
-func TestReceiverNeedsAnAddressAndMetricsComeFromOneSource(t *testing.T) {
+// The receiver adapter needs an address, and it is a server with no
+// collector.
+func TestReceiverNeedsAnAddress(t *testing.T) {
cfg := Default()
if cfg.Adapters[1].Name != AdapterClaudeCodeOTLP {
t.Fatalf("the defaults list %s second, want the receiver", cfg.Adapters[1].Name)
@@ -121,22 +120,51 @@ func TestReceiverNeedsAnAddressAndMetricsComeFromOneSource(t *testing.T) {
t.Fatal("a receiver with collector settings was accepted; it is a server")
}
cfg.Adapters[1].Collector = Collector{}
- cfg.Adapters[1].Listen = "127.0.0.1:4317"
if err := cfg.Validate(); err != nil {
- t.Fatalf("a receiver with metrics beside a local adapter without metrics must be accepted: %v", err)
+ t.Fatalf("a receiver with an address must be accepted: %v", err)
}
- cfg.Adapters[0].Metrics = true
- if err := cfg.Validate(); err == nil {
- t.Fatal("metrics on both adapters was accepted")
+}
+
+// Metrics are one section for the root, read as written, false included.
+// A section left out derives every metric with a look-back of three days.
+func TestMetricsAreOneSectionForTheRoot(t *testing.T) {
+ path := filepath.Join(t.TempDir(), "asz.yaml")
+ load := func(body string) (*Config, error) {
+ t.Helper()
+ if err := os.WriteFile(path, []byte(body), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ return Load(path)
}
- cfg.Adapters[1].Metrics = false
- if err := cfg.Validate(); err != nil {
- t.Fatalf("local metrics beside a receiver without metrics must be accepted: %v", err)
+ cfg, err := load("parse:\n max_round_bytes: 1048576\n")
+ if err != nil {
+ t.Fatal(err)
}
- cfg.Adapters[1].Enabled = false
- cfg.Adapters[1].Metrics = true
- if err := cfg.Validate(); err != nil {
- t.Fatalf("a disabled receiver must not count: %v", err)
+ if d, err := cfg.Metrics.LookbackDuration(); !cfg.Metrics.On() || err != nil || d != 72*time.Hour {
+ t.Fatalf("a file without a metrics section: on=%v look-back %s %v, want on and 72h", cfg.Metrics.On(), d, err)
+ }
+ // A section built in code, with nothing set, is the same.
+ if d, err := (Metrics{}).LookbackDuration(); !(Metrics{}).On() || err != nil || d != 72*time.Hour {
+ t.Fatalf("an empty metrics section: look-back %s %v, want on and 72h", d, err)
+ }
+ if cfg, err = load("metrics:\n enabled: false\n"); err != nil || cfg.Metrics.On() {
+ t.Fatalf("enabled: false was not read: %v", err)
+ }
+ for body, want := range map[string]time.Duration{
+ "metrics:\n lookback: 3d\n": 72 * time.Hour,
+ "metrics:\n lookback: 6h\n": 6 * time.Hour,
+ "metrics:\n lookback: none\n": 0,
+ } {
+ cfg, err := load(body)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if d, err := cfg.Metrics.LookbackDuration(); err != nil || d != want {
+ t.Fatalf("%q: look-back %s %v, want %s", body, d, err, want)
+ }
+ }
+ if _, err := load("metrics:\n lookback: soon\n"); err == nil {
+ t.Fatal("a look-back that is not a duration was accepted")
}
}
diff --git a/internal/config/defaults_test.go b/internal/config/defaults_test.go
index 3715d89..564e61d 100644
--- a/internal/config/defaults_test.go
+++ b/internal/config/defaults_test.go
@@ -43,10 +43,10 @@ func TestNamedAdaptersKeepDefaultsAndExplicitOverrides(t *testing.T) {
if len(c.Adapters) != 1 || !reflect.DeepEqual(c.Adapters[0], def) {
t.Fatalf("named defaults: got %+v, want %+v", c.Adapters, def)
}
- override := "adapters:\n - name: " + def.Name + "\n enabled: false\n exclude: []\n metrics: false\n"
+ override := "adapters:\n - name: " + def.Name + "\n enabled: false\n exclude: []\n"
c = write(override)
a := c.Adapters[0]
- if a.Enabled || a.Metrics || len(a.Exclude) != 0 {
+ if a.Enabled || len(a.Exclude) != 0 {
t.Fatalf("explicit false/empty values were lost: %+v", a)
}
})
diff --git a/internal/export/otlp/otlp_test.go b/internal/export/otlp/otlp_test.go
index dc21324..a3ff0aa 100644
--- a/internal/export/otlp/otlp_test.go
+++ b/internal/export/otlp/otlp_test.go
@@ -566,7 +566,7 @@ func TestPushSendsTheMetricsSpoolUnderTheServiceIdentity(t *testing.T) {
if err != nil {
t.Fatal(err)
}
- if _, err := storage.NewSpool(z).Put("otlp", data, time.Now()); err != nil {
+ if _, _, err := storage.NewSpool(z).PutNamed("metrics-test-000001-local.pb", data); err != nil {
t.Fatal(err)
}
rcv := startReceiver(t)
@@ -605,7 +605,7 @@ func TestPushSendsTheMetricsSpoolUnderTheServiceIdentity(t *testing.T) {
func TestPushSendsOnlyWhatIsSwitchedOn(t *testing.T) {
spooled := func(z *storage.Zone) {
data, _ := proto.Marshal(&collmetricspb.ExportMetricsServiceRequest{})
- if _, err := storage.NewSpool(z).Put("otlp", data, time.Now()); err != nil {
+ if _, _, err := storage.NewSpool(z).PutNamed("metrics-test-000001-local.pb", data); err != nil {
t.Fatal(err)
}
}
diff --git a/internal/export/otlp/sent_test.go b/internal/export/otlp/sent_test.go
index a1acb00..47631a1 100644
--- a/internal/export/otlp/sent_test.go
+++ b/internal/export/otlp/sent_test.go
@@ -66,7 +66,7 @@ func spool(t *testing.T, z *storage.Zone) (string, string) {
if err != nil {
t.Fatal(err)
}
- path, err := storage.NewSpool(z).Put("otlp", data, time.Now())
+ path, _, err := storage.NewSpool(z).PutNamed("metrics-test-000001-local.pb", data)
if err != nil {
t.Fatal(err)
}
diff --git a/internal/index/codec.go b/internal/index/codec.go
index 6d7fc1b..c003a0d 100644
--- a/internal/index/codec.go
+++ b/internal/index/codec.go
@@ -34,7 +34,7 @@ var magic = [4]byte{'A', 'S', 'I', 'X'}
const (
// seq row stream batch kind trigger flags ts record parent call run continues tool child started_by label ord first count
entryWidth = 4 + 4 + 4 + 4 + 1 + 1 + 2 + 8 + 4*9 + 8 + 4 + 4 // 80
- blockWidth = 4 + 2 + 1 + 4 + 4 + 4 + 4 // 23
+ blockWidth = 4 + 2 + 1 + 4 + 4 + 4 + 4 + 4 + 4 // 31
// entry role request previous
bodyWidth = 4 + 1 + 4 + 4 // 13
)
@@ -289,6 +289,8 @@ func encodeBlock(b []byte, k *Block) {
le.PutUint32(b[11:], k.Name)
le.PutUint32(b[15:], k.Adds)
le.PutUint32(b[19:], k.Dels)
+ le.PutUint32(b[23:], k.Server)
+ le.PutUint32(b[27:], k.ServerTool)
}
func decodeBlock(b []byte, k *Block) {
@@ -299,6 +301,8 @@ func decodeBlock(b []byte, k *Block) {
k.Name = le.Uint32(b[11:])
k.Adds = le.Uint32(b[15:])
k.Dels = le.Uint32(b[19:])
+ k.Server = le.Uint32(b[23:])
+ k.ServerTool = le.Uint32(b[27:])
}
// LoadFor loads a session's index only if it agrees with its saved state.
diff --git a/internal/index/fromrecord.go b/internal/index/fromrecord.go
index 9330e0d..acc46b7 100644
--- a/internal/index/fromrecord.go
+++ b/internal/index/fromrecord.go
@@ -126,6 +126,8 @@ func kindOf(hdr *sessiondata.Header, rec *sessiondata.Record) Kind {
return KindChanges
case sessiondata.KindProviderBody:
return KindProviderBody
+ case sessiondata.KindExecution:
+ return KindExecution
}
switch rec.From {
case sessiondata.FromAgent:
@@ -200,6 +202,7 @@ func blocksOf(in *Interner, rec *sessiondata.Record) []Block {
switch p.Kind {
case sessiondata.PartCall:
b.Kind, b.ToolID, b.Name = BlockToolUse, in.ID(p.ID), in.ID(p.Name)
+ b.Server, b.ServerTool = in.ID(p.Server), in.ID(p.ServerTool)
case sessiondata.PartResult:
b.Kind, b.ToolID = BlockToolResult, in.ID(p.Of)
case sessiondata.PartText:
diff --git a/internal/index/index.go b/internal/index/index.go
index 6ef6429..8f90d0d 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, Block or Body
// changes; a mismatch discards the index and rebuilds rather than migrating.
-const Schema = 14
+const Schema = 15
// Kind classifies a record without reading it.
type Kind uint8
@@ -53,6 +53,9 @@ const (
// KindProviderBody is a body the runtime exchanged with its model
// provider; evidence, never a step, and in no derived lookup.
KindProviderBody
+ // KindExecution is what an observer saw a tool call do; evidence, never
+ // a step.
+ KindExecution
)
// Trigger says what caused a prompt cycle.
@@ -284,4 +287,8 @@ type Block struct {
// summed over its files; zero for every other kind.
Adds uint32
Dels uint32
+ // Server and ServerTool are the MCP server and tool a tool_use block's
+ // name addresses, interned, where the adapter could split the name.
+ Server uint32
+ ServerTool uint32
}
diff --git a/internal/metrics/mcp.go b/internal/metrics/mcp.go
new file mode 100644
index 0000000..018fced
--- /dev/null
+++ b/internal/metrics/mcp.go
@@ -0,0 +1,117 @@
+// 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 metrics
+
+import (
+ "encoding/hex"
+ "errors"
+ "hash"
+ "io"
+ "os"
+ "time"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecode"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/execution"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+)
+
+// executionKey is how an observation is remembered as counted, beside the
+// calls the token metric counts. The prefix keeps the two apart.
+func executionKey(id string) string { return "execution:" + id }
+
+// deriveExecutions turns one execution file's records into the MCP family's
+// points.
+//
+// One observation is one call counted, once per session by its own id: the
+// same line landed again in a later file is the same observation. A record
+// is counted in the minute it was observed. Only calls to MCP servers are
+// counted. The server is the name the observer gave, the tool the part of
+// the runtime's name after the server where that name splits exactly, and
+// otherwise the whole name: together they name the target a call reached,
+// which a receiver can treat as an endpoint of the agent. The server's
+// source is where its configuration came from. Measured on Claude Code
+// 2.1.282, every hook after a call carries its duration, failures included,
+// so the time divided by the calls of a series is the mean. A record without
+// a duration would still count as a call and add no time.
+//
+// Nothing says whether a tool reads or writes. An MCP server can declare
+// that per tool, but measured on Claude Code 2.1.282 the declaration reaches
+// neither the model, nor the hooks, nor the transcript, so no label is made
+// up for it.
+func deriveExecutions(rd *sessiondata.Reader, f *os.File, h hash.Hash, hdr *sessiondata.Header, since time.Time, ss *sessionState, res *result) (*result, error) {
+ source := SourceSubagent
+ if hdr.Stream == "main" {
+ source = SourceMain
+ }
+ sums := map[point]int64{}
+ here := map[string]bool{}
+ for {
+ rec, err := rd.Next()
+ if errors.Is(err, io.EOF) {
+ break
+ }
+ if err != nil {
+ return nil, err
+ }
+ t, ok := recordTime(rec)
+ if ok && t.After(res.latest) {
+ res.latest = t
+ }
+ for _, p := range rec.Parts {
+ if p.Kind != sessiondata.PartData {
+ continue
+ }
+ r, isExec := execution.Decode(p.Data)
+ if !isExec || r.Protocol != execution.ProtocolMCP || r.Server == nil || !ok {
+ continue
+ }
+ key := executionKey(r.ID)
+ if ss.Calls[key] || here[key] {
+ continue
+ }
+ here[key] = true
+ res.counted = append(res.counted, key)
+ minute := t.Truncate(time.Minute)
+ if !since.IsZero() && minute.Add(time.Minute).Before(since) {
+ continue
+ }
+ tool := r.ToolName
+ if _, split, exact := claudecode.MCPName(r.ToolName); exact {
+ tool = split
+ }
+ s := series{server: r.Server.Name, origin: r.Server.Source, tool: tool, outcome: r.Outcome, source: source}
+ s.metric = MCPCalls
+ sums[point{series: s, session: hdr.Session, minute: minute}]++
+ if r.DurationMS != nil {
+ s.metric = MCPDuration
+ sums[point{series: s, session: hdr.Session, minute: minute}] += *r.DurationMS
+ }
+ }
+ }
+ if _, err := io.Copy(io.Discard, f); err != nil {
+ return nil, err
+ }
+ res.digest = hex.EncodeToString(h.Sum(nil))
+ points := make([]point, 0, len(sums))
+ for k, v := range sums {
+ k.value = v
+ points = append(points, k)
+ }
+ res.points = windows(points, res, ss)
+ return res, nil
+}
diff --git a/internal/metrics/mcp_test.go b/internal/metrics/mcp_test.go
new file mode 100644
index 0000000..620b848
--- /dev/null
+++ b/internal/metrics/mcp_test.go
@@ -0,0 +1,141 @@
+// 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 metrics_test
+
+import (
+ "io"
+ "os"
+ "path/filepath"
+ "testing"
+ "time"
+
+ collmetricspb "go.opentelemetry.io/proto/otlp/collector/metrics/v1"
+ "google.golang.org/protobuf/proto"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/export/otlp/otlptest"
+ "github.com/apache/skywalking-ai-sessionizer/internal/metrics"
+ "github.com/apache/skywalking-ai-sessionizer/internal/storage"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/execution"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/model"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+)
+
+// observed is one record the plugin writes after a call to an MCP server.
+func observed(tool, name, server string, ms int64, outcome string) *execution.Record {
+ return &execution.Record{Schema: execution.Schema, ID: tool + "/" + execution.BoundaryClientHook,
+ ObservedBy: execution.ObservedByASZPlugin, Boundary: execution.BoundaryClientHook,
+ Session: "s1", Stream: "main", Tool: tool, ToolName: name, Protocol: execution.ProtocolMCP,
+ Server: &execution.Server{Name: server, Source: "user"}, Time: base.Format(time.RFC3339Nano),
+ DurationMS: &ms, Outcome: outcome}
+}
+
+// landExecutions lands one execution file, each record as the changes
+// adapter lands a line: whole, as one data part.
+func landExecutions(t *testing.T, z *storage.Zone, seq uint64, recs ...*execution.Record) {
+ t.Helper()
+ path := filepath.Join(z.StreamDir("s1", "main"), storage.LandedName("execution", storage.Stamp(time.Unix(int64(seq), 0)), seq))
+ err := storage.WriteAtomic(path, storage.PermLanded, func(w io.Writer) error {
+ sw, err := sessiondata.NewWriter(w, &sessiondata.Header{Seq: seq, At: "2026-09-07T00:00:00Z", Kind: sessiondata.KindExecution,
+ Adapter: "test/0", Dialect: "execution/1", Src: "s1/main.execution", Session: "s1", Stream: "main"})
+ if err != nil {
+ return err
+ }
+ for i, r := range recs {
+ line, err := r.Marshal()
+ if err != nil {
+ return err
+ }
+ if err := sw.Write(&sessiondata.Record{Ord: uint64(i + 1), Sha: "x", Bytes: len(line), ID: r.ID, Tool: r.Tool, Time: r.Time,
+ Parts: []sessiondata.Part{{Kind: sessiondata.PartData, Data: line, State: model.ContentAvailable, Bytes: len(line)}}}); err != nil {
+ return err
+ }
+ }
+ return sw.Close()
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ aged(t, path, base.Add(-time.Hour))
+}
+
+// mcpTotals sums the MCP family in the spool by metric, server, tool and
+// outcome.
+func mcpTotals(t *testing.T, z *storage.Zone) map[string]float64 {
+ t.Helper()
+ files, err := storage.NewSpool(z).List()
+ if err != nil {
+ t.Fatal(err)
+ }
+ out := map[string]float64{}
+ for _, sf := range files {
+ data, err := os.ReadFile(sf.Path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var req collmetricspb.ExportMetricsServiceRequest
+ if err := proto.Unmarshal(data, &req); err != nil {
+ t.Fatal(err)
+ }
+ for _, rm := range req.ResourceMetrics {
+ for _, sm := range rm.ScopeMetrics {
+ for _, m := range sm.Metrics {
+ for _, dp := range m.GetSum().GetDataPoints() {
+ a := otlptest.Attrs(dp.Attributes)
+ out[m.Name+" "+a["mcp_server.name"]+"/"+a["mcp_tool.name"]+"/"+a["outcome"]] += dp.GetAsDouble()
+ }
+ }
+ }
+ }
+ }
+ return out
+}
+
+// TestAnObservationIsCountedOnceAcrossFiles. The plugin appends a line for
+// every event its hook is handed, and the runtime can hand one event over
+// again later, so the same record can land in a later file than the first.
+// It is still one call. A tool name that does not split exactly is kept
+// whole, so a call is never put under a tool it did not name.
+func TestAnObservationIsCountedOnceAcrossFiles(t *testing.T) {
+ z := storage.NewZone(t.TempDir())
+ lookup := observed("toolu_1", "mcp__status__lookup", "status", 300, execution.OutcomeReturned)
+ odd := observed("toolu_2", "mcp__odd__name__twice", "odd", 20, execution.OutcomeFailed)
+ landExecutions(t, z, 1, lookup, odd)
+ d := deriver(z, base.Add(time.Hour), metrics.Options{})
+ if _, err := d.Pass(nil); err != nil {
+ t.Fatal(err)
+ }
+ landExecutions(t, z, 2, lookup)
+ if st, err := d.Pass(nil); err != nil || len(st.Errors) != 0 {
+ t.Fatal(err, st.Errors)
+ }
+ want := map[string]float64{
+ metrics.MCPCalls + " status/lookup/returned": 1,
+ metrics.MCPDuration + " status/lookup/returned": 300,
+ metrics.MCPCalls + " odd/mcp__odd__name__twice/failed": 1,
+ metrics.MCPDuration + " odd/mcp__odd__name__twice/failed": 20,
+ }
+ got := mcpTotals(t, z)
+ for k, v := range want {
+ if got[k] != v {
+ t.Fatalf("%s: %v, want %v; the spool holds %v", k, got[k], v, got)
+ }
+ }
+ if len(got) != len(want) {
+ t.Fatalf("the spool holds %v, want %v", got, want)
+ }
+}
diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go
index f00f4e1..8b0eca9 100644
--- a/internal/metrics/metrics.go
+++ b/internal/metrics/metrics.go
@@ -53,14 +53,17 @@ import (
"github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
)
-// The metric, as the runtime's exporter names it.
+// The token metric. asz names it for an agent, not for one runtime, so every
+// runtime's tokens are one family. Claude Code's own exporter calls the same
+// count by a name of its own, and the claude-code-otlp adapter gives it this
+// name when it lands it, so a receiver holds one name whichever produced the
+// points.
const (
// TokenUsage counts tokens, by type and model, per session and per query
- // source; the exporter's claude_code.token.usage.
- TokenUsage = "claude_code.token.usage"
+ // source.
+ TokenUsage = "agent.token.usage"
tokenUnit = "tokens"
tokenDesc = "Number of tokens used"
-
// RuntimeService is the service name the runtime's exporter puts on its
// resource. A derived request carries the same, and asz push normalises
// both to asz's identity on the way out.
@@ -83,6 +86,26 @@ const (
StateFile = "metrics.state"
)
+// The MCP family. No runtime writes a metric of a call to an MCP server, so
+// it is asz's alone, named beside the token metric. It is derived from
+// execution records, one observation of one call each, and never from a
+// transcript, which does not say which server ran a call or how long it
+// took. A server and a tool together name the target a call reached, which
+// a receiver can treat as an endpoint of the agent.
+const (
+ // MCPCalls counts calls to MCP servers, by server, tool and outcome.
+ MCPCalls = "agent.mcp.calls"
+ mcpCallsUnit = "{call}"
+ mcpCallsDesc = "Number of calls to MCP servers an observer recorded"
+ // MCPDuration sums the time the observer measured around each call,
+ // in milliseconds. Divided by the calls of the same series it is the
+ // mean time a call took as the runtime saw it, which includes waiting
+ // before the call started and is not the server's own time.
+ MCPDuration = "agent.mcp.duration"
+ mcpDurationUnit = "ms"
+ mcpDurationDesc = "Time the observer measured around calls to MCP servers"
+)
+
// The token types, spelled as the exporter spells them.
const (
TypeInput = "input"
@@ -185,13 +208,6 @@ func (d *Deriver) Pass(sessions []string) (*Stats, error) {
if d.Lookback > 0 {
since = d.Now().Add(-d.Lookback)
}
- // The runtime's exporter may have been the source until now: what
- // it sent is in the spool with the time it was received, and the
- // derivation starts after the last of it, so a switch of source
- // counts nothing twice.
- if received, err := latestReceived(spool); err == nil && received.After(since) {
- since = received
- }
}
if sessions == nil {
if sessions, err = sessionDirs(d.Zone.Root()); err != nil {
@@ -292,22 +308,6 @@ func (d *Deriver) Pass(sessions []string) (*Stats, error) {
return st, nil
}
-// latestReceived is when the newest request from the runtime's exporter was
-// put in the spool; zero when there is none.
-func latestReceived(spool *storage.Spool) (time.Time, error) {
- files, err := spool.List()
- if err != nil {
- return time.Time{}, err
- }
- var latest time.Time
- for _, f := range files {
- if f.Source != SourceLocal && f.At.After(latest) {
- latest = f.At
- }
- }
- return latest, nil
-}
-
// call is what one provider call's fragments said, as far as they were read:
// the last fragment's usage and model, its time, and whether the call
// finished.
@@ -322,9 +322,21 @@ type call struct {
// series names one time series of the metric: what a point's attributes
// name, apart from the session, which is the state's own.
-type series struct{ model, source, typ string }
+type series struct {
+ metric string
+ model, source, typ string
+ server, tool, outcome string
+ // origin is where the runtime's configuration of the MCP server came
+ // from, one value per server.
+ origin string
+}
-func (s series) key() string { return s.model + "|" + s.source + "|" + s.typ }
+func (s series) key() string {
+ if s.metric == "" || s.metric == TokenUsage {
+ return s.model + "|" + s.source + "|" + s.typ
+ }
+ return s.metric + "|" + s.server + "|" + s.origin + "|" + s.tool + "|" + s.outcome + "|" + s.source
+}
// point is one minute of one series, with the window it was given.
type point struct {
@@ -372,10 +384,13 @@ func deriveFile(lf storage.LandedFile, following []storage.LandedFile, since tim
// The metric is Claude Code's own family, derived from its transcripts
// alone. A root can hold another runtime's conversations beside them,
// and those carry calls and usage too, so without this a LangChain
- // session's tokens went out as claude_code.token.usage. Every landed
+ // session's tokens went out as Claude Code's token metric. Every landed
// header names a dialect; the reader refuses one that does not. The
// mock dialect is a scenario writing Claude Code's shape directly, and
// the scenarios hold its metrics to the Claude Code build's.
+ if hdr.Kind == sessiondata.KindExecution {
+ return deriveExecutions(rd, f, h, &hdr, since, ss, res)
+ }
if d, _, _ := strings.Cut(hdr.Dialect, "/"); d != "claude-code" && d != "mock" {
if _, err := io.Copy(h, f); err != nil {
return nil, err
@@ -477,8 +492,14 @@ func deriveFile(lf storage.LandedFile, following []storage.LandedFile, since tim
k.value = v
points = append(points, k)
}
- // The same file derives the same bytes: points in one order, and the
- // windows follow from it.
+ res.points = windows(points, res, ss)
+ return res, nil
+}
+
+// windows orders a file's points and gives each its window. The same file
+// derives the same bytes: points in one order, and the windows follow from
+// it.
+func windows(points []point, res *result, ss *sessionState) []point {
sort.Slice(points, func(i, j int) bool {
a, b := points[i], points[j]
if !a.minute.Equal(b.minute) {
@@ -506,8 +527,7 @@ func deriveFile(lf storage.LandedFile, following []storage.LandedFile, since tim
}
res.lastEnd[p.series.key()] = p.end
}
- res.points = points
- return res, nil
+ return points
}
// take folds one fragment into the call: the last fragment's usage, model
@@ -566,37 +586,69 @@ func continueIn(path string, c *call) (more, matched bool, err error) {
// scope, one metric, delta sums that are monotonic, as the exporter's SDK
// writes them.
func request(points []point, version string) *collmetricspb.ExportMetricsServiceRequest {
- dps := make([]*metricspb.NumberDataPoint, 0, len(points))
+ byMetric := map[string][]*metricspb.NumberDataPoint{}
+ var names []string
for _, p := range points {
- attrs := []*commonpb.KeyValue{
- str("session.id", p.session),
- str("type", p.series.typ),
- str("query_source", p.series.source),
+ name := p.series.metric
+ if name == "" {
+ name = TokenUsage
+ }
+ var attrs []*commonpb.KeyValue
+ if name == TokenUsage {
+ attrs = []*commonpb.KeyValue{
+ str("session.id", p.session),
+ str("type", p.series.typ),
+ str("query_source", p.series.source),
+ }
+ if p.series.model != "" {
+ attrs = append(attrs, str("model", p.series.model))
+ }
+ } else {
+ attrs = []*commonpb.KeyValue{
+ str("session.id", p.session),
+ str("mcp_server.name", p.series.server),
+ str("mcp_server.source", p.series.origin),
+ str("mcp_tool.name", p.series.tool),
+ str("outcome", p.series.outcome),
+ str("query_source", p.series.source),
+ }
}
- if p.series.model != "" {
- attrs = append(attrs, str("model", p.series.model))
+ if _, ok := byMetric[name]; !ok {
+ names = append(names, name)
}
// A double, as the exporter's SDK encodes its counters, so a
// receiver reads one value kind from both sources.
- dps = append(dps, &metricspb.NumberDataPoint{
+ byMetric[name] = append(byMetric[name], &metricspb.NumberDataPoint{
Attributes: attrs,
StartTimeUnixNano: uint64(p.start.UnixNano()),
TimeUnixNano: uint64(p.end.UnixNano()),
Value: &metricspb.NumberDataPoint_AsDouble{AsDouble: float64(p.value)},
})
}
+ sort.Strings(names)
+ metrics := make([]*metricspb.Metric, 0, len(names))
+ for _, name := range names {
+ desc, unit := tokenDesc, tokenUnit
+ switch name {
+ case MCPCalls:
+ desc, unit = mcpCallsDesc, mcpCallsUnit
+ case MCPDuration:
+ desc, unit = mcpDurationDesc, mcpDurationUnit
+ }
+ metrics = append(metrics, &metricspb.Metric{
+ Name: name, Description: desc, Unit: unit,
+ Data: &metricspb.Metric_Sum{Sum: &metricspb.Sum{
+ DataPoints: byMetric[name],
+ AggregationTemporality: metricspb.AggregationTemporality_AGGREGATION_TEMPORALITY_DELTA,
+ IsMonotonic: true,
+ }},
+ })
+ }
return &collmetricspb.ExportMetricsServiceRequest{ResourceMetrics: []*metricspb.ResourceMetrics{{
Resource: &resourcepb.Resource{Attributes: []*commonpb.KeyValue{str("service.name", RuntimeService)}},
ScopeMetrics: []*metricspb.ScopeMetrics{{
- Scope: &commonpb.InstrumentationScope{Name: ScopeName, Version: version},
- Metrics: []*metricspb.Metric{{
- Name: TokenUsage, Description: tokenDesc, Unit: tokenUnit,
- Data: &metricspb.Metric_Sum{Sum: &metricspb.Sum{
- DataPoints: dps,
- AggregationTemporality: metricspb.AggregationTemporality_AGGREGATION_TEMPORALITY_DELTA,
- IsMonotonic: true,
- }},
- }},
+ Scope: &commonpb.InstrumentationScope{Name: ScopeName, Version: version},
+ Metrics: metrics,
}},
}}}
}
diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go
index babbbf4..f47d1dd 100644
--- a/internal/metrics/metrics_test.go
+++ b/internal/metrics/metrics_test.go
@@ -365,11 +365,8 @@ func TestAPassRunAgainWritesNothingMore(t *testing.T) {
}
}
-// The look-back bounds the first pass only, and the runtime's own metrics
-// already in the spool bound it further: the derivation starts after the
-// last request the receiver landed, so a switch of source counts nothing
-// twice.
-func TestLookbackAndReceivedMetricsBoundTheFirstPass(t *testing.T) {
+// The look-back bounds the first pass only.
+func TestLookbackBoundsTheFirstPass(t *testing.T) {
z := storage.NewZone(t.TempDir())
now := base
land(t, z, "s1", "main", 1, []call{{id: "old", model: "m", at: now.Add(-48 * time.Hour), frags: 1, in: 100, out: 100}})
@@ -391,30 +388,14 @@ func TestLookbackAndReceivedMetricsBoundTheFirstPass(t *testing.T) {
t.Fatalf("a later pass must derive an old file whole: requests=%d err=%v", st.Requests, err)
}
- z2 := storage.NewZone(t.TempDir())
- land(t, z2, "s1", "main", 1, []call{
- {id: "before", model: "m", at: now.Add(-30 * time.Minute), frags: 1, in: 5, out: 5},
- {id: "after", model: "m", at: now.Add(-5 * time.Minute), frags: 1, in: 6, out: 6},
- })
- data, _ := proto.Marshal(&collmetricspb.ExportMetricsServiceRequest{})
- if _, err := storage.NewSpool(z2).Put("otlp", data, now.Add(-10*time.Minute)); err != nil {
- t.Fatal(err)
- }
- d2 := deriver(z2, now, metrics.Options{Lookback: 24 * time.Hour})
- if _, err := d2.Pass(nil); err != nil {
- t.Fatal(err)
- }
- if got, _ := points(t, z2); total(got["main/input/m#s1"]) != 6 {
- t.Fatalf("what the receiver already landed must not be derived again: %v", got)
- }
}
// TestAnotherRuntimesTranscriptDerivesNothing.
//
-// The metric is Claude Code's own family. A root can hold a LangChain
-// conversation beside Claude Code's, landed by the receiver with calls and
-// usage of its own, and those tokens went out as claude_code.token.usage
-// because the derivation read every transcript in the root.
+// The token metric counts the model calls of Claude Code transcripts only. A
+// root can hold a LangChain conversation beside Claude Code's, landed by the
+// receiver with calls and usage of its own, and those tokens went out as
+// Claude Code's because the derivation read every transcript in the root.
func TestAnotherRuntimesTranscriptDerivesNothing(t *testing.T) {
z := storage.NewZone(t.TempDir())
c := call{id: "c1", model: "m", at: base, frags: 1, in: 4, out: 60}
diff --git a/internal/metrics/parity_test.go b/internal/metrics/parity_test.go
index 6e6c933..5883b0c 100644
--- a/internal/metrics/parity_test.go
+++ b/internal/metrics/parity_test.go
@@ -52,6 +52,10 @@ func capture(t *testing.T) *collmetricspb.ExportMetricsServiceRequest {
return &req
}
+// exporterTokenUsage is what Claude Code's exporter calls the metric asz
+// derives as agent.token.usage.
+const exporterTokenUsage = "claude_code.token.usage"
+
func find(req *collmetricspb.ExportMetricsServiceRequest, name string) *metricspb.Metric {
for _, rm := range req.ResourceMetrics {
for _, sm := range rm.ScopeMetrics {
@@ -91,10 +95,15 @@ func contains(list []string, s string) bool {
// page lists as not derived. Every other metric the exporter sent is on
// the page's list of metrics not derived. A newer runtime that sends
// another label or another metric fails here, and the page is updated.
+//
+// The exporter calls the metric by the runtime's name, and asz by its own,
+// agent.token.usage, since a runtime's vocabulary stops at its adapter. The
+// name is the one difference the page lists for the metric itself.
func TestDerivedPointsMatchTheCapturedExporter(t *testing.T) {
- theirs := find(capture(t), metrics.TokenUsage)
+ landed := capture(t)
+ theirs := find(landed, exporterTokenUsage)
if theirs == nil {
- t.Fatal("the capture carries no token metric")
+ t.Fatal("the capture carries no token metric under the exporter's name")
}
z := storage.NewZone(t.TempDir())
@@ -118,7 +127,7 @@ func TestDerivedPointsMatchTheCapturedExporter(t *testing.T) {
}
// The metric itself.
- if ours.Name != theirs.Name || ours.Unit != theirs.Unit || ours.Description != theirs.Description {
+ if ours.Name == theirs.Name || ours.Unit != theirs.Unit || ours.Description != theirs.Description {
t.Fatalf("metric: derived %q %q %q, exporter %q %q %q", ours.Name, ours.Unit, ours.Description, theirs.Name, theirs.Unit, theirs.Description)
}
if ours.GetSum().GetAggregationTemporality() != theirs.GetSum().GetAggregationTemporality() || ours.GetSum().GetIsMonotonic() != theirs.GetSum().GetIsMonotonic() {
@@ -181,13 +190,13 @@ func TestDerivedPointsMatchTheCapturedExporter(t *testing.T) {
}
// Every other metric in the capture is one the page says is not derived.
- for _, rm := range capture(t).ResourceMetrics {
+ for _, rm := range landed.ResourceMetrics {
for _, sm := range rm.ScopeMetrics {
if sm.GetScope().GetName() != "com.anthropic.claude_code" {
t.Fatalf("the exporter's scope is %q; the export page names another", sm.GetScope().GetName())
}
for _, m := range sm.Metrics {
- if m.Name != metrics.TokenUsage && !contains(metrics.NotDerivedMetrics, m.Name) {
+ if m.Name != exporterTokenUsage && !contains(metrics.NotDerivedMetrics, m.Name) {
t.Fatalf("the exporter sent %s, which is neither derived nor listed as not derived", m.Name)
}
}
diff --git a/internal/metrics/session.go b/internal/metrics/session.go
index 02a1f68..3829f69 100644
--- a/internal/metrics/session.go
+++ b/internal/metrics/session.go
@@ -42,8 +42,7 @@ func SpoolName(session string, seq uint64) string {
// part before the source, and it has at least six digits. Everything before
// it is the session. A pattern such as metrics-<session>-* is never used,
// because one session id can be another followed by a dash and digits, and
-// the other's files would then be taken for its own. A request the receiver
-// adapter put in the spool names no session, so no session owns it.
+// the other's files would then be taken for its own.
func SpoolOwner(name string) (string, bool) {
rest, ok := strings.CutPrefix(name, "metrics-")
if !ok {
diff --git a/internal/scenario/build.go b/internal/scenario/build.go
index 7e51339..d2f1a7b 100644
--- a/internal/scenario/build.go
+++ b/internal/scenario/build.go
@@ -190,14 +190,15 @@ adapters:
- name: claude-code-local
enabled: true
source_root: %s
- # The runtime's own metric family, derived from what is landed, with
- # no look-back: a scenario is history by construction.
- metrics: true
- metrics_lookback: none
%s - name: claude-code-changes
enabled: true
source_root: %s/plugins/data
-%s%s# To export the session with asz push, name the receiver:
+%s%s# The metrics, derived from what is landed with no look-back: a scenario is
+# history by construction.
+metrics:
+ enabled: true
+ lookback: none
+# To export the session with asz push, name the receiver:
# export:
# otlp:
# protocol: grpc
diff --git a/internal/scenario/changes.go b/internal/scenario/changes.go
index 06f8f8a..28dc0e2 100644
--- a/internal/scenario/changes.go
+++ b/internal/scenario/changes.go
@@ -189,30 +189,36 @@ func (p *Plan) runtimeResult(e *Event) map[string]any {
// nothing with the original, so a scenario can be built again without its
// changes and the two folds compared.
func stripChanges(steps []Step) []Step {
+ return mapTools(steps, func(t *Tool) { t.Changes = nil })
+}
+
+// mapTools returns the steps with edit applied to a copy of every tool,
+// nested ones included, sharing nothing with the original.
+func mapTools(steps []Step, edit func(*Tool)) []Step {
out := make([]Step, len(steps))
for i, s := range steps {
if s.Call != nil {
c := *s.Call
- if c.Tool != nil && len(c.Tool.Changes) > 0 {
+ if c.Tool != nil {
t := *c.Tool
- t.Changes = nil
+ edit(&t)
c.Tool = &t
}
if c.Agent != nil {
a := *c.Agent
- a.Steps = stripChanges(a.Steps)
+ a.Steps = mapTools(a.Steps, edit)
c.Agent = &a
}
if c.Skill != nil {
k := *c.Skill
- k.Steps = stripChanges(k.Steps)
+ k.Steps = mapTools(k.Steps, edit)
c.Skill = &k
}
if c.Workflow != nil {
w := *c.Workflow
w.Children = make([]Child, len(w.Children))
for j, child := range c.Workflow.Children {
- child.Steps = stripChanges(child.Steps)
+ child.Steps = mapTools(child.Steps, edit)
w.Children[j] = child
}
c.Workflow = &w
@@ -224,25 +230,26 @@ func stripChanges(steps []Step) []Step {
return out
}
-// HasChanges reports whether any tool in the steps changes a file.
-func HasChanges(steps []Step) bool {
+// anyTool reports whether any tool in the steps, nested ones included,
+// satisfies has.
+func anyTool(steps []Step, has func(*Tool) bool) bool {
for _, s := range steps {
if s.Call == nil {
continue
}
c := s.Call
- if c.Tool != nil && len(c.Tool.Changes) > 0 {
+ if c.Tool != nil && has(c.Tool) {
return true
}
- if c.Agent != nil && HasChanges(c.Agent.Steps) {
+ if c.Agent != nil && anyTool(c.Agent.Steps, has) {
return true
}
- if c.Skill != nil && HasChanges(c.Skill.Steps) {
+ if c.Skill != nil && anyTool(c.Skill.Steps, has) {
return true
}
if c.Workflow != nil {
for _, child := range c.Workflow.Children {
- if HasChanges(child.Steps) {
+ if anyTool(child.Steps, has) {
return true
}
}
@@ -251,6 +258,11 @@ func HasChanges(steps []Step) bool {
return false
}
+// HasChanges reports whether any tool in the steps changes a file.
+func HasChanges(steps []Step) bool {
+ return anyTool(steps, func(t *Tool) bool { return len(t.Changes) > 0 })
+}
+
// WithoutChanges is the scenario with every change removed.
func (sc *Scenario) WithoutChanges() *Scenario {
out := *sc
diff --git a/internal/scenario/execution.go b/internal/scenario/execution.go
new file mode 100644
index 0000000..6dad29b
--- /dev/null
+++ b/internal/scenario/execution.go
@@ -0,0 +1,92 @@
+// 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 (
+ "encoding/json"
+ "time"
+
+ "github.com/apache/skywalking-ai-sessionizer/pkg/execution"
+)
+
+// Execution is what the plugin's hook saw of one call to an MCP server.
+type Execution struct {
+ Server string `yaml:"server"`
+ Source string `yaml:"source"`
+ // Outcome is returned, failed or interrupted. Empty follows the result:
+ // failed when the result failed, returned otherwise.
+ Outcome string `yaml:"outcome"`
+ Duration time.Duration `yaml:"duration"`
+ // Arguments are what the server was sent, when a hook before the call
+ // changed them. Empty means the tool's own input.
+ Arguments map[string]any `yaml:"arguments"`
+ // Twice writes the record a second time, as the plugin does when the
+ // runtime hands its hook the same event again. It is one call.
+ Twice bool `yaml:"twice"`
+}
+
+// executionRecord is the record the plugin writes for a call to an MCP
+// server, from the hook after it.
+func (p *Plan) executionRecord(e *Event) *execution.Record {
+ x := e.Execution
+ outcome := x.Outcome
+ if outcome == "" {
+ outcome = execution.OutcomeReturned
+ if e.Failed != nil && *e.Failed {
+ outcome = execution.OutcomeFailed
+ }
+ }
+ args := x.Arguments
+ if args == nil {
+ args = e.ToolInput
+ }
+ if args == nil {
+ args = map[string]any{}
+ }
+ r := &execution.Record{
+ Schema: execution.Schema, ID: e.Of + "/" + execution.BoundaryClientHook,
+ ObservedBy: execution.ObservedByASZPlugin, Boundary: execution.BoundaryClientHook,
+ Session: p.Session, Stream: e.Stream, Tool: e.Of, ToolName: e.ToolName, Cwd: p.Cwd(),
+ Protocol: execution.ProtocolMCP, Server: &execution.Server{Name: x.Server, Source: x.Source},
+ Time: ccTime(e.At), Outcome: outcome,
+ Arguments: execution.Measure(jsLine(args)),
+ }
+ if x.Duration > 0 {
+ ms := x.Duration.Milliseconds()
+ r.DurationMS = &ms
+ }
+ if outcome == execution.OutcomeReturned {
+ // The runtime hands the hook the MCP answer's content blocks.
+ answer, _ := json.Marshal([]map[string]any{{"type": "text", "text": e.Text}})
+ r.Result = execution.Measure(answer)
+ }
+ return r
+}
+
+// HasExecutions reports whether any tool in the steps has an execution.
+func HasExecutions(steps []Step) bool {
+ return anyTool(steps, func(t *Tool) bool { return t.Execution != nil })
+}
+
+// WithoutExecutions is the scenario with every execution removed, so it can
+// be built again and the two folds compared.
+func (sc *Scenario) WithoutExecutions() *Scenario {
+ out := *sc
+ out.Steps = mapTools(sc.Steps, func(t *Tool) { t.Execution = nil })
+ return &out
+}
diff --git a/internal/scenario/expect/expect.go b/internal/scenario/expect/expect.go
index bc98744..6cf90b9 100644
--- a/internal/scenario/expect/expect.go
+++ b/internal/scenario/expect/expect.go
@@ -97,6 +97,10 @@ 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"`
+ // ExecutionsLeaveTheFold says the same scenario without its execution
+ // records folds to the same nodes, relations and talks: a record is
+ // evidence beside a step, never a step.
+ ExecutionsLeaveTheFold *bool `yaml:"executions_leave_the_fold"`
// ProviderBodiesLeaveTheFold says the same scenario without its provider
// bodies folds to the same nodes and relations. Checked when a scenario
// writes provider bodies.
@@ -183,7 +187,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",
+ "provider_body": "provider_body", "execution": "execution",
}
// Resolve finds the landed file a Lose names among a session's files. The
@@ -277,6 +281,12 @@ type View struct {
ChangedFiles *int `yaml:"changed_files"`
CapturedBy map[string]int `yaml:"captured_by"`
ChangesJoined *int `yaml:"changes_joined"`
+ // Executions counts the tool execution records, ExecutionsJoined the
+ // records that found their step, and Outcomes the records by how the
+ // call ended.
+ Executions *int `yaml:"executions"`
+ ExecutionsJoined *int `yaml:"executions_joined"`
+ Outcomes map[string]int `yaml:"outcomes"`
// ProviderBodies counts the landed provider bodies, and CapturedPrompts
// the calls that list a request.
// ProviderFiles counts the session's provider_body files.
@@ -566,6 +576,27 @@ func checkView(root, session string, want *View) ([]string, error) {
if want.CapturedPrompts != nil && doc.Summary.CapturedPrompts != *want.CapturedPrompts {
bad("view.captured_prompts is %d, want %d", doc.Summary.CapturedPrompts, *want.CapturedPrompts)
}
+ if want.Executions != nil && len(doc.ToolExecutions) != *want.Executions {
+ bad("view.executions is %d, want %d", len(doc.ToolExecutions), *want.Executions)
+ }
+ if want.ExecutionsJoined != nil || want.Outcomes != nil {
+ joined := 0
+ outcomes := map[string]int{}
+ for _, te := range doc.ToolExecutions {
+ outcomes[te.Outcome]++
+ if te.Step != "" {
+ joined++
+ }
+ }
+ if want.ExecutionsJoined != nil && joined != *want.ExecutionsJoined {
+ bad("view.executions_joined is %d, want %d", joined, *want.ExecutionsJoined)
+ }
+ for o, n := range want.Outcomes {
+ if outcomes[o] != n {
+ bad("view.outcomes[%s] is %d, want %d", o, outcomes[o], n)
+ }
+ }
+ }
if want.ChangedFiles != nil || want.CapturedBy != nil || want.ChangesJoined != nil {
files, joined := 0, 0
bySource := map[string]int{}
@@ -618,6 +649,8 @@ type Summary struct {
TalksOn map[string]int
Unresolved map[string]int // kind -> open references
Streams []string
+ // Records is how many records each stream holds, by stream.
+ Records map[string]int
}
// Summarize reads a fold into a Summary.
@@ -626,7 +659,7 @@ func Summarize(root, session string) (*Summary, error) {
if err != nil {
return nil, err
}
- s := &Summary{Kinds: map[string]int{}, Relations: map[string]int{}, TalksOn: map[string]int{}, Unresolved: map[string]int{}}
+ s := &Summary{Kinds: map[string]int{}, Relations: map[string]int{}, TalksOn: map[string]int{}, Unresolved: map[string]int{}, Records: map[string]int{}}
for _, n := range v.Nodes {
s.Kinds[n.Kind]++
if n.Kind == model.KindTalk {
@@ -634,6 +667,11 @@ func Summarize(root, session string) (*Summary, error) {
}
if n.Kind == model.KindStream {
s.Streams = append(s.Streams, n.Stream)
+ var attrs struct {
+ Records int `json:"records"`
+ }
+ _ = json.Unmarshal(n.Attrs, &attrs)
+ s.Records[n.Stream] = attrs.Records
}
}
for _, r := range v.Relations {
@@ -678,6 +716,34 @@ func Compare(a, b *Summary) []string {
return out
}
+// CompareRecords says how the record counts of the streams in two summaries
+// differ. Evidence beside a stream, such as a change record, a provider body
+// or an execution record, must not enter the stream: it would add no node,
+// and still change the stream's record count and time range, so the same
+// session would fold differently with and without it. Formats are not
+// compared this way, because each lands its own records.
+func CompareRecords(a, b *Summary) []string {
+ var out []string
+ keys := map[string]bool{}
+ for k := range a.Records {
+ keys[k] = true
+ }
+ for k := range b.Records {
+ keys[k] = true
+ }
+ var names []string
+ for k := range keys {
+ names = append(names, k)
+ }
+ sort.Strings(names)
+ for _, k := range names {
+ if a.Records[k] != b.Records[k] {
+ out = append(out, fmt.Sprintf("records on %s: %d against %d", k, a.Records[k], b.Records[k]))
+ }
+ }
+ return out
+}
+
// Properties that read the root alone.
// FoldEqualsParse checks that folding every round equals one full parse of
diff --git a/internal/scenario/marker.go b/internal/scenario/marker.go
index d9f36fa..06a5556 100644
--- a/internal/scenario/marker.go
+++ b/internal/scenario/marker.go
@@ -125,7 +125,7 @@ func shortDuration(d time.Duration) string {
}
// ParseRemove reads a removal policy: immediately, a Go duration such as 30m
-// or 24h, or a whole number of days such as 7d, as metrics_lookback takes
+// or 24h, or a whole number of days such as 7d, as metrics.lookback takes
// them. Zero, a negative value and anything else are refused, because a
// retention of nothing is immediately said another way.
func ParseRemove(s string) (Removal, error) {
diff --git a/internal/scenario/plan.go b/internal/scenario/plan.go
index b0a8132..55eb2eb 100644
--- a/internal/scenario/plan.go
+++ b/internal/scenario/plan.go
@@ -83,10 +83,11 @@ const (
// ToolUse is the request a tool_use fragment carries.
type ToolUse struct {
- ID string
- Name string
- Input map[string]any
- Changes []Change
+ ID string
+ Name string
+ Input map[string]any
+ Changes []Change
+ Execution *Execution
}
// Ack is what a launch acknowledgement names.
@@ -147,6 +148,10 @@ type Event struct {
// it changed, for the writers to record each the way its producer does.
ToolName string
Changes []Change
+ // Execution and ToolInput are what the plugin's hook saw of the call,
+ // and the input the call was made with.
+ Execution *Execution
+ ToolInput map[string]any
// A notice.
NoticeTool string
@@ -493,7 +498,7 @@ func (b *planner) call(l *lane, s *Step, id string) error {
if err := checkChanges(c.Tool, id); err != nil {
return err
}
- tool = &ToolUse{ID: tid, Name: c.Tool.Name, Input: c.Tool.Input, Changes: c.Tool.Changes}
+ tool = &ToolUse{ID: tid, Name: c.Tool.Name, Input: c.Tool.Input, Changes: c.Tool.Changes, Execution: c.Tool.Execution}
case c.Agent != nil:
tool = &ToolUse{ID: id + "-tool", Name: "Agent", Input: map[string]any{"description": c.Agent.Name, "prompt": c.Agent.Prompt}}
case c.Skill != nil:
@@ -531,7 +536,7 @@ func (b *planner) call(l *lane, s *Step, id string) error {
rat := b.tick(l, r.After)
e := Event{Kind: EvResult, Stream: l.stream, Batch: l.batch, At: rat, ID: id + "-result", Parent: l.last, Run: run,
Of: tool.ID, Text: r.Text, Failed: r.Failed, StringEnrichment: r.String, Lost: lost || r.Lost,
- ToolName: tool.Name, Changes: tool.Changes}
+ ToolName: tool.Name, Changes: tool.Changes, Execution: tool.Execution, ToolInput: tool.Input}
b.emit(e)
l.last = e.ID
case c.Agent != nil:
diff --git a/internal/scenario/run/push.go b/internal/scenario/run/push.go
index 4b510ba..5c4e103 100644
--- a/internal/scenario/run/push.go
+++ b/internal/scenario/run/push.go
@@ -43,13 +43,14 @@ import (
"github.com/apache/skywalking-ai-sessionizer/internal/scenario/expect"
"github.com/apache/skywalking-ai-sessionizer/internal/storage"
"github.com/apache/skywalking-ai-sessionizer/internal/verify"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/execution"
"github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
"github.com/apache/skywalking-ai-sessionizer/pkg/sessionflow"
)
// The file kinds the export page names.
var wireKinds = map[string]bool{
- "changes": true, "provider_body": true,
+ "changes": true, "execution": true, "provider_body": true,
"transcript": true, "agent_meta": true, "journal": true, "workflow_manifest": true, "workflow_script": true, "round": true,
}
@@ -60,10 +61,10 @@ var wireKinds = map[string]bool{
// receiver bounds a read on, delivery once and at least once, and the
// export path: writing every body back gives a root that verifies and
// folds the same.
-func pushFollowsTheWire(out, session string, f scenario.Format, want *expect.Push, tokens map[string]int64, checkTokens bool) ([]string, error) {
+func pushFollowsTheWire(out, session string, f scenario.Format, want *expect.Push, tokens, mcp map[string]int64, checkTokens bool) ([]string, error) {
var problems []string
for _, protocol := range []string{otlp.ProtocolGRPC, otlp.ProtocolHTTP} {
- found, err := pushOver(protocol, out, session, f, want, tokens, checkTokens)
+ found, err := pushOver(protocol, out, session, f, want, tokens, mcp, checkTokens)
if err != nil {
return nil, fmt.Errorf("over %s: %w", protocol, err)
}
@@ -74,7 +75,7 @@ func pushFollowsTheWire(out, session string, f scenario.Format, want *expect.Pus
return problems, nil
}
-func pushOver(protocol, out, session string, f scenario.Format, want *expect.Push, tokens map[string]int64, checkTokens bool) ([]string, error) {
+func pushOver(protocol, out, session string, f scenario.Format, want *expect.Push, tokens, mcp map[string]int64, checkTokens bool) ([]string, error) {
rcv, err := otlptest.Start()
if err != nil {
return nil, err
@@ -331,6 +332,33 @@ func pushOver(protocol, out, session string, f scenario.Format, want *expect.Pus
if st.Metrics == 0 && len(tokens) > 0 {
bad("metrics: the plan has tokens but the push sent no metrics request")
}
+ // The MCP family says what the plan's execution records say: each
+ // call counted once, and its measured time summed, per series.
+ gotMCP := map[string]int64{}
+ for _, req := range rcv.MetricsRequests() {
+ for _, rm := range req.GetResourceMetrics() {
+ for _, sm := range rm.GetScopeMetrics() {
+ for _, m := range sm.GetMetrics() {
+ if m.GetName() != metrics.MCPCalls && m.GetName() != metrics.MCPDuration {
+ continue
+ }
+ for _, dp := range m.GetSum().GetDataPoints() {
+ gotMCP[m.GetName()+"/"+mcpKey(otlptest.Attrs(dp.GetAttributes()))] += int64(dp.GetAsDouble())
+ }
+ }
+ }
+ }
+ }
+ for k, v := range mcp {
+ if gotMCP[k] != v {
+ bad("metrics: %s on the wire is %d, the plan says %d", k, gotMCP[k], v)
+ }
+ }
+ for k, v := range gotMCP {
+ if _, planned := mcp[k]; !planned {
+ bad("metrics: %s on the wire is %d, the plan has no such calls", k, v)
+ }
+ }
}
// Sent once: a second pass sends nothing.
@@ -381,6 +409,42 @@ func pushOver(protocol, out, session string, f scenario.Format, want *expect.Pus
return out2, nil
}
+// mcpOnTheWire checks one metric of the MCP family: its unit, a monotonic
+// delta sum, and points that name the session, the server, the tool, how the
+// call ended and the stream that made it.
+func mcpOnTheWire(m *metricspb.Metric, i int, session string, bad func(string, ...any), window func(string, uint64, uint64)) {
+ unit := "{call}"
+ if m.GetName() == metrics.MCPDuration {
+ unit = "ms"
+ }
+ if m.GetUnit() != unit {
+ bad("request %d: %s has unit %q, want %q", i, m.GetName(), m.GetUnit(), unit)
+ }
+ sum := m.GetSum()
+ if sum == nil || sum.GetAggregationTemporality() != metricspb.AggregationTemporality_AGGREGATION_TEMPORALITY_DELTA || !sum.GetIsMonotonic() {
+ bad("request %d: %s is not a monotonic delta sum", i, m.GetName())
+ return
+ }
+ for _, dp := range sum.GetDataPoints() {
+ a := otlptest.Attrs(dp.GetAttributes())
+ switch {
+ case a["session.id"] != session, a["mcp_server.name"] == "", a["mcp_tool.name"] == "",
+ a["outcome"] != execution.OutcomeReturned && a["outcome"] != execution.OutcomeFailed && a["outcome"] != execution.OutcomeInterrupted,
+ a["query_source"] != metrics.SourceMain && a["query_source"] != metrics.SourceSubagent:
+ bad("request %d: a %s point carries %v", i, m.GetName(), a)
+ }
+ if _, ok := dp.GetValue().(*metricspb.NumberDataPoint_AsDouble); !ok || dp.GetTimeUnixNano() <= dp.GetStartTimeUnixNano() || dp.GetAsDouble() < 0 {
+ bad("request %d: a %s point is not a positive window of a double count", i, m.GetName())
+ }
+ window(m.GetName()+"/"+mcpKey(a), dp.GetStartTimeUnixNano(), dp.GetTimeUnixNano())
+ }
+}
+
+// mcpKey names one series of the MCP family by its labels.
+func mcpKey(a map[string]string) string {
+ return a["mcp_server.name"] + "/" + a["mcp_server.source"] + "/" + a["mcp_tool.name"] + "/" + a["outcome"] + "/" + a["query_source"]
+}
+
// tokensOnTheWire checks the shape of every metrics request as the export
// page states it: asz's identity on the resource, the runtime's metric with
// its unit, monotonic delta sums, one-minute points that name the session.
@@ -416,6 +480,12 @@ func tokensOnTheWire(rcv *otlptest.Receiver, session string) (out []string) {
bad("request %d scope is %q", i, sm.GetScope().GetName())
}
for _, m := range sm.GetMetrics() {
+ if m.GetName() == metrics.MCPCalls || m.GetName() == metrics.MCPDuration {
+ mcpOnTheWire(m, i, session, bad, func(key string, start, end uint64) {
+ windows[key] = append(windows[key], window{start, end})
+ })
+ continue
+ }
if m.GetName() != metrics.TokenUsage || m.GetUnit() != "tokens" {
bad("request %d carries metric %s %s", i, m.GetName(), m.GetUnit())
continue
diff --git a/internal/scenario/run/run.go b/internal/scenario/run/run.go
index 21e184a..d7d6134 100644
--- a/internal/scenario/run/run.go
+++ b/internal/scenario/run/run.go
@@ -39,6 +39,7 @@ import (
"github.com/apache/skywalking-ai-sessionizer/internal/scenario"
"github.com/apache/skywalking-ai-sessionizer/internal/scenario/expect"
"github.com/apache/skywalking-ai-sessionizer/internal/storage"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/execution"
)
// Options settle a check run.
@@ -245,7 +246,7 @@ func checkFormat(sc *scenario.Scenario, ex *expect.File, f scenario.Format, out
{"records_well_formed", ex.Properties.RecordsWellFormed, func() ([]string, error) { return expect.RecordsWellFormed(out, session) }},
{"repack_keeps_structure", ex.Properties.RepackKeepsStructure, func() ([]string, error) { return repackKeepsStructure(out, session, ex.Parse.MaxRoundBytes) }},
{"push_follows_the_wire", ex.Properties.PushFollowsTheWire, func() ([]string, error) {
- return pushFollowsTheWire(out, session, f, ex.Push, expectedTokens(built.Plan), expect.On(ex.Properties.MetricsMatchThePlan))
+ return pushFollowsTheWire(out, session, f, ex.Push, expectedTokens(built.Plan), expectedMCP(built.Plan), expect.On(ex.Properties.MetricsMatchThePlan))
}},
{"removed_after_sent", ex.Properties.RemovedAfterSent, func() ([]string, error) {
return removedAfterSent(sc, f, out, session, built.Plan.Project, ex.Parse.MaxRoundBytes, opts)
@@ -257,6 +258,12 @@ func checkFormat(sc *scenario.Scenario, ex *expect.File, f scenario.Format, out
}
return changesLeaveTheFold(sc, f, out, session, opts, ex.Parse.MaxRoundBytes)
}},
+ {"executions_leave_the_fold", ex.Properties.ExecutionsLeaveTheFold, func() ([]string, error) {
+ if !scenario.HasExecutions(sc.Steps) {
+ return nil, nil
+ }
+ return leavesTheFold("executions_leave_the_fold", "the execution records", sc.WithoutExecutions(), f, out, session, opts, ex.Parse.MaxRoundBytes)
+ }},
{"provider_bodies_leave_the_fold", ex.Properties.ProviderBodiesLeaveTheFold, func() ([]string, error) {
if !sc.ProviderBodies {
return nil, nil
@@ -504,7 +511,7 @@ func leavesTheFold(property, what string, plainScenario *scenario.Scenario, f sc
return nil, err
}
var lines []string
- for _, d := range expect.Compare(with, without) {
+ for _, d := range append(expect.Compare(with, without), expect.CompareRecords(with, without)...) {
lines = append(lines, property+": the fold differs with and without "+what+": "+d)
}
same, err := expect.SameIdentity(out, session, plain, built.Session)
@@ -700,6 +707,40 @@ func names(p *scenario.Plan) map[string]string {
return out
}
+// expectedMCP is what the plan says the MCP family counts: each call the
+// plugin's hook saw once, and the time it measured, by server, tool, how the
+// call ended and the stream that made it. A replayed result is the same call.
+func expectedMCP(p *scenario.Plan) map[string]int64 {
+ out := map[string]int64{}
+ for i := range p.Events {
+ e := &p.Events[i]
+ if e.Kind != scenario.EvResult || e.Execution == nil || e.Replayed {
+ continue
+ }
+ source := metrics.SourceSubagent
+ if e.Stream == "main" {
+ source = metrics.SourceMain
+ }
+ tool := e.ToolName
+ if _, split, ok := claudecode.MCPName(e.ToolName); ok {
+ tool = split
+ }
+ outcome := e.Execution.Outcome
+ if outcome == "" {
+ outcome = execution.OutcomeReturned
+ if e.Failed != nil && *e.Failed {
+ outcome = execution.OutcomeFailed
+ }
+ }
+ key := e.Execution.Server + "/" + e.Execution.Source + "/" + tool + "/" + outcome + "/" + source
+ out[metrics.MCPCalls+"/"+key]++
+ if e.Execution.Duration > 0 {
+ out[metrics.MCPDuration+"/"+key] += e.Execution.Duration.Milliseconds()
+ }
+ }
+ return out
+}
+
// expectedTokens is what the plan says the runtime's exporter would have
// counted: every call's usage once, whatever its fragments repeat, under
// the stream that made it, main or a child. A lost call left no record, a
diff --git a/internal/scenario/scenario.go b/internal/scenario/scenario.go
index b518bc9..6e8a7a5 100644
--- a/internal/scenario/scenario.go
+++ b/internal/scenario/scenario.go
@@ -189,6 +189,10 @@ type Tool struct {
// change record beside the stream. Both land as changes/1 and both
// join the step, so the same scenario checks the two paths.
Changes []Change `yaml:"changes"`
+ // Execution is what the plugin's hook saw of a call to an MCP server.
+ // A claude-code build writes it the way the plugin does, to the
+ // stream's execution file; an sd build lands it. Both join the step.
+ Execution *Execution `yaml:"execution"`
}
// Change is one file a tool changed. A side that is absent, nil in YAML or
diff --git a/internal/scenario/write_claudecode.go b/internal/scenario/write_claudecode.go
index db3a4be..18b9e25 100644
--- a/internal/scenario/write_claudecode.go
+++ b/internal/scenario/write_claudecode.go
@@ -39,7 +39,7 @@ import (
// and digest of the bytes written, for the session's marker.
func writeClaudeCode(p *Plan, root string) ([]string, []MarkerFile, error) {
proj := filepath.Join(root, p.Project)
- w := &ccWriter{p: p, files: map[string][]string{}, plugin: map[string][]string{}}
+ w := &ccWriter{p: p, files: map[string][]string{}, plugin: map[string][]string{}, execs: map[string][]string{}}
lost := p.lostStreams()
for i := range p.Events {
e := &p.Events[i]
@@ -160,6 +160,11 @@ func writeClaudeCode(p *Plan, root string) ([]string, []MarkerFile, error) {
return nil, nil, err
}
}
+ for stream, lines := range w.execs {
+ if err := putUnder(root, PluginOutputDir+"/"+p.Session+"/"+stream+".execution.jsonl", lines); err != nil {
+ 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.
@@ -198,6 +203,8 @@ type ccWriter struct {
files map[string][]string
// plugin holds the lines the asz plugin would have written, by stream.
plugin map[string][]string
+ // execs are the plugin's execution records, one file per stream.
+ execs map[string][]string
}
func ccTime(t time.Time) string { return t.UTC().Format("2006-01-02T15:04:05.000Z") }
@@ -344,6 +351,14 @@ func (w *ccWriter) event(e *Event) error {
line, _ := w.p.pluginRecord(e).Marshal()
w.plugin[s] = append(w.plugin[s], string(line))
}
+ if e.Execution != nil && !e.Replayed {
+ // The plugin's hook after the call saw which server ran it.
+ line, _ := w.p.executionRecord(e).Marshal()
+ w.execs[s] = append(w.execs[s], string(line))
+ if e.Execution.Twice {
+ w.execs[s] = append(w.execs[s], string(line))
+ }
+ }
}
w.add(s, w.rec(m, s))
case EvNotice:
diff --git a/internal/scenario/write_sd.go b/internal/scenario/write_sd.go
index bb4369d..2307157 100644
--- a/internal/scenario/write_sd.go
+++ b/internal/scenario/write_sd.go
@@ -28,6 +28,7 @@ import (
"sort"
"time"
+ "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/adapters/mock"
"github.com/apache/skywalking-ai-sessionizer/internal/index"
@@ -225,7 +226,45 @@ func (w *sdWriter) stream(stream string, events []*Event) error {
if _, err := w.land(dir, "transcript", "transcript.cursor", hdr, recs); err != nil {
return err
}
- return w.changes(stream, events)
+ if err := w.changes(stream, events); err != nil {
+ return err
+ }
+ return w.executions(stream, events)
+}
+
+// executions lands what the plugin's hook saw of each call to an MCP server
+// on the stream, as the changes adapter lands the plugin's execution lines:
+// the line whole, as one data part.
+func (w *sdWriter) executions(stream string, events []*Event) error {
+ var recs []*sessiondata.Record
+ var off uint64
+ for _, e := range events {
+ if e.Kind != EvResult || e.Execution == nil || e.Replayed {
+ continue
+ }
+ xr := w.p.executionRecord(e)
+ line, err := xr.Marshal()
+ if err != nil {
+ return err
+ }
+ times := 1
+ if e.Execution.Twice {
+ times = 2
+ }
+ for range times {
+ r := &sessiondata.Record{ID: xr.ID, Tool: xr.Tool, Time: xr.Time, Parts: []sessiondata.Part{{
+ Kind: sessiondata.PartData, Data: line, State: model.ContentAvailable, Bytes: len(line)}}}
+ finish(r, uint64(len(recs)+1), &off)
+ recs = append(recs, r)
+ }
+ }
+ if len(recs) == 0 {
+ return nil
+ }
+ dir := w.z.StreamDir(w.p.Session, stream)
+ hdr := sessiondata.Header{Kind: sessiondata.KindExecution, Src: w.source("streams", stream+".execution"), Stream: stream}
+ _, err := w.land(dir, "execution", "execution.cursor", hdr, recs)
+ return err
}
// changes lands what the plugin would have observed on the stream: one
@@ -360,7 +399,11 @@ func (w *sdWriter) record(e *Event) *sessiondata.Record {
input = map[string]any{}
}
data := jsLine(input)
- r.Parts = []sessiondata.Part{{Kind: sessiondata.PartCall, Data: data, ID: e.Tool.ID, Name: e.Tool.Name, State: "available", Bytes: len(data)}}
+ call := sessiondata.Part{Kind: sessiondata.PartCall, Data: data, ID: e.Tool.ID, Name: e.Tool.Name, State: "available", Bytes: len(data)}
+ // The adapter reads the MCP server and tool from the name, and so
+ // does the record written directly, so the two formats land alike.
+ call.Server, call.ServerTool, _ = claudecode.MCPName(e.Tool.Name)
+ r.Parts = []sessiondata.Part{call}
}
case EvResult:
r.From = sessiondata.FromExternal
diff --git a/internal/storage/spool.go b/internal/storage/spool.go
index 06ed3f5..6eb48bb 100644
--- a/internal/storage/spool.go
+++ b/internal/storage/spool.go
@@ -18,29 +18,23 @@
package storage
import (
- "bufio"
- "errors"
- "fmt"
"io"
"os"
"path/filepath"
"sort"
"strconv"
"strings"
- "time"
)
-// SpoolDir is where OTLP metrics requests wait under a storage root: the
-// ones the local adapter derived from landed files, and the ones the
-// runtime's own exporter sent to the receiver adapter. Each is one file,
-// bytes as produced or as received, write-once like every landed file, and
-// asz push sends them in order and records which were sent.
+// SpoolDir is where OTLP metrics requests wait under a storage root, each
+// derived from one landed file. Each is one file, write-once like every
+// landed file, and asz push sends them in order and records which were sent.
const SpoolDir = "_metrics"
// Spool is the metrics spool of one storage root.
type Spool struct{ dir string }
-// NewSpool opens the spool of a root; nothing is created until Put.
+// NewSpool opens the spool of a root; nothing is created until PutNamed.
func NewSpool(z *Zone) *Spool { return &Spool{dir: filepath.Join(z.Root(), SpoolDir)} }
// Dir is the spool directory.
@@ -51,45 +45,6 @@ type SpoolFile struct {
Path string
Seq uint64
Source string
- // At is when a received request was put, read off its name; zero for
- // a derived request, which is named after its landed file instead.
- At time.Time
-}
-
-// spoolName builds a spool filename: sortable by the time it was put and
-// then by a sequence, which keeps two requests of one nanosecond apart.
-func spoolName(stamp string, seq uint64, source string) string {
- return fmt.Sprintf("metrics-%s-%06d-%s.pb", stamp, seq, source)
-}
-
-// Put writes one request to the spool and returns its path. The sequence
-// comes from the spool's own state file under the spool's lock, so two
-// writers in one root, the local derivation and the receiver, never take
-// the same number.
-func (s *Spool) Put(source string, data []byte, now time.Time) (string, error) {
- if err := os.MkdirAll(s.dir, 0o755); err != nil {
- return "", err
- }
- lock, err := lockDir(s.dir)
- if err != nil {
- return "", err
- }
- defer func() { _ = lock.Unlock() }()
- seq, err := s.nextSeq()
- if err != nil {
- return "", err
- }
- path := filepath.Join(s.dir, spoolName(Stamp(now), seq, source))
- if err := WriteAtomic(path, PermLanded, func(w io.Writer) error {
- _, err := w.Write(data)
- return err
- }); err != nil {
- return "", err
- }
- if err := s.saveSeq(seq+1, now); err != nil {
- return "", err
- }
- return path, nil
}
// PutNamed writes one request under a name the caller chose, and reports
@@ -113,45 +68,6 @@ func (s *Spool) PutNamed(name string, data []byte) (string, bool, error) {
return path, true, nil
}
-func (s *Spool) statePath() string { return filepath.Join(s.dir, "spool.state") }
-
-// nextSeq reads the counter, or recovers it from the files when the state
-// is gone: the files are the authority, as they are for a session.
-func (s *Spool) nextSeq() (uint64, error) {
- f, err := os.Open(s.statePath())
- if err == nil {
- defer f.Close()
- sc := bufio.NewScanner(f)
- for sc.Scan() {
- if v, ok := strings.CutPrefix(sc.Text(), "next_seq "); ok {
- return strconv.ParseUint(strings.TrimSpace(v), 10, 64)
- }
- }
- return 0, errors.New("storage: spool.state carries no next_seq")
- }
- if !os.IsNotExist(err) {
- return 0, err
- }
- files, err := s.List()
- if err != nil {
- return 0, err
- }
- var next uint64 = 1
- for _, sf := range files {
- if sf.Seq >= next {
- next = sf.Seq + 1
- }
- }
- return next, nil
-}
-
-func (s *Spool) saveSeq(next uint64, now time.Time) error {
- return WriteAtomic(s.statePath(), PermState, func(w io.Writer) error {
- _, err := fmt.Fprintf(w, "schema 1\nupdated_at %s\nnext_seq %d\n", now.UTC().Format(time.RFC3339Nano), next)
- return err
- })
-}
-
// List returns every request in the spool, in the order they were put.
func (s *Spool) List() ([]SpoolFile, error) {
items, err := os.ReadDir(s.dir)
@@ -167,7 +83,7 @@ func (s *Spool) List() ([]SpoolFile, error) {
if it.IsDir() || !strings.HasPrefix(name, "metrics-") || !strings.HasSuffix(name, ".pb") {
continue
}
- // metrics-<stamp>-<seq>-<source>.pb
+ // metrics-<session>-<seq>-<source>.pb
parts := strings.Split(strings.TrimSuffix(name, ".pb"), "-")
if len(parts) < 4 {
continue
@@ -176,14 +92,9 @@ func (s *Spool) List() ([]SpoolFile, error) {
if err != nil {
continue
}
- sf := SpoolFile{Path: filepath.Join(s.dir, name), Seq: seq, Source: parts[len(parts)-1]}
- if at, err := time.Parse("20060102T150405.000000000Z", parts[1]); err == nil {
- sf.At = at
- }
- out = append(out, sf)
+ out = append(out, SpoolFile{Path: filepath.Join(s.dir, name), Seq: seq, Source: parts[len(parts)-1]})
}
- // Received requests in the order they were put, derived ones by name;
- // the order only has to be the same every time.
+ // The order only has to be the same every time.
sort.Slice(out, func(i, j int) bool {
if out[i].Seq != out[j].Seq {
return out[i].Seq < out[j].Seq
diff --git a/internal/verify/session.go b/internal/verify/session.go
index d9dbcd7..2b3d1c4 100644
--- a/internal/verify/session.go
+++ b/internal/verify/session.go
@@ -79,7 +79,7 @@ func Session(z *storage.Zone, session string) (*SessionReport, error) {
sub string
kinds []string
}{
- {"streams", []string{"transcript", "meta", "changes"}},
+ {"streams", []string{"transcript", "meta", "changes", "execution"}},
{"runs", []string{"journal", "manifest", "script"}},
} {
base := filepath.Join(z.SessionDir(session), group.sub)
diff --git a/internal/view/changes.go b/internal/view/changes.go
index 5ebf6a1..010b8ce 100644
--- a/internal/view/changes.go
+++ b/internal/view/changes.go
@@ -44,21 +44,7 @@ import (
// tool-use id, which the change record names and the step's call part
// carries; nothing is matched by time.
func (c *Conversation) workspaceChanges(landed []storage.LandedFile, recs map[[2]uint64]*sessiondata.Record) []sessionview.WorkspaceChange {
- // Which step carries which tool-use id, from the call part each tool
- // step points at.
- stepOf := map[string]string{}
- for _, n := range c.View.Nodes {
- if (n.Kind != model.KindTool && n.Kind != model.KindAgentCall) || n.Ref == nil {
- continue
- }
- rec := recs[[2]uint64{n.Ref.Seq, n.Ref.Row}]
- if rec == nil {
- continue
- }
- if p := partAt(rec, n.Ref.Block); p != nil && p.Kind == sessiondata.PartCall && p.ID != "" {
- stepOf[p.ID] = n.ID
- }
- }
+ stepOf := c.stepsByToolUse(recs)
var out []sessionview.WorkspaceChange
seen := map[string]bool{}
@@ -134,8 +120,8 @@ func (c *Conversation) workspaceChanges(landed []storage.LandedFile, recs map[[2
// stay: the runtime's is listed first, and a viewer showing one
// prefers it.
sort.SliceStable(out, func(i, j int) bool {
- if out[i].Time != out[j].Time {
- return out[i].Time < out[j].Time
+ if c := compareTimes(out[i].Time, out[j].Time); c != 0 {
+ return c < 0
}
if out[i].CapturedBy != out[j].CapturedBy {
return out[i].CapturedBy == changes.CapturedByClaudeCode
@@ -148,6 +134,25 @@ func (c *Conversation) workspaceChanges(landed []storage.LandedFile, recs map[[2
return out
}
+// stepsByToolUse says which step carries which tool-use id, from the call
+// part each tool step points at.
+func (c *Conversation) stepsByToolUse(recs map[[2]uint64]*sessiondata.Record) map[string]string {
+ stepOf := map[string]string{}
+ for _, n := range c.View.Nodes {
+ if (n.Kind != model.KindTool && n.Kind != model.KindAgentCall) || n.Ref == nil {
+ continue
+ }
+ rec := recs[[2]uint64{n.Ref.Seq, n.Ref.Row}]
+ if rec == nil {
+ continue
+ }
+ if p := partAt(rec, n.Ref.Block); p != nil && p.Kind == sessiondata.PartCall && p.ID != "" {
+ stepOf[p.ID] = n.ID
+ }
+ }
+ return stepOf
+}
+
// partAt returns the part a reference names, or the only part when the
// reference names none.
func partAt(rec *sessiondata.Record, block *int) *sessiondata.Part {
diff --git a/internal/view/conversationview.go b/internal/view/conversationview.go
index f524eeb..13d0fbc 100644
--- a/internal/view/conversationview.go
+++ b/internal/view/conversationview.go
@@ -89,6 +89,7 @@ func (c *Conversation) Build() (*sessionview.Conversation, error) {
Rounds: []sessionview.Round{}, Files: []sessionview.File{}, Talks: []sessionview.Node{}, Loose: []sessionview.Node{},
Relations: []sessionview.Relation{}, Unresolved: []sessionview.Unresolved{},
WorkspaceChanges: []sessionview.WorkspaceChange{},
+ ToolExecutions: []sessionview.ToolExecution{},
}
if sn := c.View.Nodes[sessionflow.NodeID("session", c.Session)]; sn != nil {
v.Summary.From, v.Summary.To = millisOf(attrString(sn, "from_time")), millisOf(attrString(sn, "through_time"))
@@ -175,6 +176,17 @@ func (c *Conversation) Build() (*sessionview.Conversation, error) {
annotateChanges(v.Loose, byStep)
v.Summary.Changes = len(v.WorkspaceChanges)
+ // The tool execution records, joined to their steps the same way.
+ v.ToolExecutions = c.toolExecutions(landed, recs)
+ execsByStep := map[string][]string{}
+ for _, te := range v.ToolExecutions {
+ if te.Step != "" {
+ execsByStep[te.Step] = append(execsByStep[te.Step], te.ID)
+ }
+ }
+ annotateExecutions(v.Talks, execsByStep)
+ annotateExecutions(v.Loose, execsByStep)
+
// The provider bodies each call carries, as the round joined them. A call
// names where its request and its response landed; a reader loads them when
// it wants them, and the document stays the size of the structure.
@@ -369,6 +381,28 @@ func countByte(b []byte, x byte) int {
return n
}
+// compareTimes orders two record times by the instant they name. RFC 3339
+// text does not sort as text when its fractions differ in length, and Go's
+// RFC3339Nano, which the plugin writes, drops trailing zeros: as text, .11
+// sorts before .1, and .5 before the whole second. A time that does not
+// parse sorts after every one that does, by its text, so the order stays
+// total.
+func compareTimes(a, b string) int {
+ ta, ea := time.Parse(time.RFC3339Nano, a)
+ tb, eb := time.Parse(time.RFC3339Nano, b)
+ switch {
+ case ea == nil && eb == nil:
+ if c := ta.Compare(tb); c != 0 {
+ return c
+ }
+ case ea == nil:
+ return -1
+ case eb == nil:
+ return 1
+ }
+ return strings.Compare(a, b)
+}
+
// millisOf renders a record time a round header carries as unix
// milliseconds, or 0 when it carries none.
func millisOf(s string) int64 {
diff --git a/internal/view/executions.go b/internal/view/executions.go
new file mode 100644
index 0000000..b18b127
--- /dev/null
+++ b/internal/view/executions.go
@@ -0,0 +1,104 @@
+// 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 (
+ "errors"
+ "io"
+ "os"
+ "path/filepath"
+ "sort"
+ "strings"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/storage"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/execution"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessionflow"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessionview"
+)
+
+// toolExecutions gathers every execution record the session's execution
+// files carry and joins each to the step of the call it observed.
+//
+// The join is the tool-use id the record names, which the step's call part
+// carries; nothing is matched by name or time. One call can have several
+// records, one per observation, so a step lists them all. A record is kept
+// once by its own id: the same record landed twice, by a hook handed its
+// event again or by an interrupted pass, is one record. A record whose call is not a step of the document keeps no
+// step rather than being dropped.
+func (c *Conversation) toolExecutions(landed []storage.LandedFile, recs map[[2]uint64]*sessiondata.Record) []sessionview.ToolExecution {
+ stepOf := c.stepsByToolUse(recs)
+ var out []sessionview.ToolExecution
+ seen := map[string]bool{}
+ for _, lf := range landed {
+ if !strings.HasPrefix(filepath.Base(lf.Path), string(sessiondata.KindExecution)+"-") {
+ 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
+ }
+ for b, p := range rec.Parts {
+ if p.Kind != sessiondata.PartData {
+ continue
+ }
+ r, ok := execution.Decode(p.Data)
+ if !ok || seen[r.ID] {
+ continue
+ }
+ seen[r.ID] = true
+ block := b
+ out = append(out, sessionview.ToolExecution{Step: stepOf[r.Tool],
+ Ref: sessionflow.Ref{Seq: lf.Seq, Row: row, Block: &block}, Record: *r})
+ }
+ }
+ f.Close()
+ }
+ // By time, then by id, so the same files give the same list.
+ sort.SliceStable(out, func(i, j int) bool {
+ if c := compareTimes(out[i].Time, out[j].Time); c != 0 {
+ return c < 0
+ }
+ return out[i].ID < out[j].ID
+ })
+ if out == nil {
+ out = []sessionview.ToolExecution{}
+ }
+ return out
+}
+
+// annotateExecutions writes each step's execution record ids onto the step,
+// in the order the document lists the records.
+func annotateExecutions(nodes []sessionview.Node, byStep map[string][]string) {
+ for i := range nodes {
+ if ids := byStep[nodes[i].ID]; len(ids) > 0 {
+ nodes[i].Executions = ids
+ }
+ annotateExecutions(nodes[i].Children, byStep)
+ }
+}
diff --git a/internal/view/order_test.go b/internal/view/order_test.go
index cdb2b87..ba2d51c 100644
--- a/internal/view/order_test.go
+++ b/internal/view/order_test.go
@@ -18,6 +18,7 @@
package view
import (
+ "sort"
"strings"
"testing"
@@ -59,3 +60,14 @@ func TestTalkOrderIsATotalOrder(t *testing.T) {
}
}
}
+
+// TestRecordTimesSortByInstant. The plugin writes Go's RFC3339Nano, which
+// drops trailing zeros, so the text of three times in order is not in order.
+func TestRecordTimesSortByInstant(t *testing.T) {
+ in := []string{"2026-09-26T10:00:00.11Z", "not a time", "2026-09-26T10:00:00.5Z", "2026-09-26T10:00:00.1Z", "2026-09-26T10:00:00Z"}
+ sort.SliceStable(in, func(i, j int) bool { return compareTimes(in[i], in[j]) < 0 })
+ want := "2026-09-26T10:00:00Z 2026-09-26T10:00:00.1Z 2026-09-26T10:00:00.11Z 2026-09-26T10:00:00.5Z not a time"
+ if got := strings.Join(in, " "); got != want {
+ t.Fatalf("%s, want %s", got, want)
+ }
+}
diff --git a/pkg/execution/execution.go b/pkg/execution/execution.go
new file mode 100644
index 0000000..f33a778
--- /dev/null
+++ b/pkg/execution/execution.go
@@ -0,0 +1,230 @@
+// 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 execution defines the tool execution record, execution/1: what
+// one observer saw a tool call do after the model asked for it.
+//
+// A transcript holds what the model asked for and what came back. It does
+// not say which server ran the call, how long the call took, or what the
+// server was sent, which can differ from what the model wrote. An observer
+// around the call can say those things. Today the one observer is the Claude
+// Code plugin, whose hooks run after every call to an MCP server.
+//
+// A record says which observer wrote it and where that observer stands. One
+// call can have several records: a retry, a bridge that calls several
+// servers, or two observers of the same call. Each has an id of its own, and
+// all of them name the call by its tool-use id, which is the only join. A
+// record's field names are the model's, never a runtime's.
+package execution
+
+import (
+ "bytes"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "strings"
+)
+
+// Schema is the record's format version, carried in every record.
+const Schema = "execution/1"
+
+// Who observed the call.
+const (
+ // ObservedByASZPlugin is the asz plugin, from the runtime's own hooks.
+ ObservedByASZPlugin = "asz-plugin"
+)
+
+// Where the observer stands. Two observers at different places can see
+// different parameters for one call, so a record says which place it saw.
+const (
+ // BoundaryClientHook is the runtime's hook after the call: what the
+ // runtime sent the server once every hook before the call had run, and
+ // what the runtime made of the answer. Measured on Claude Code 2.1.282:
+ // a hook that rewrote the input before the call changed what the server
+ // received and what this hook sees, while the transcript kept the
+ // model's own input.
+ BoundaryClientHook = "client_hook"
+)
+
+// Protocol is what the call went over.
+const (
+ ProtocolMCP = "mcp"
+)
+
+// Outcome is what the observer saw the call end with.
+const (
+ // OutcomeReturned is an answer the runtime did not treat as an error.
+ OutcomeReturned = "returned"
+ // OutcomeFailed is a call the runtime reported as failed: an error the
+ // server returned, a lost connection, or a timeout. Measured on Claude
+ // Code 2.1.282, an MCP result marked isError reaches the failure hook,
+ // not the success hook.
+ OutcomeFailed = "failed"
+ // OutcomeInterrupted is a call a person stopped.
+ OutcomeInterrupted = "interrupted"
+)
+
+// Content states for what a record says about the parameters and the
+// answer. Only their size is kept today, and the record says so.
+const (
+ ContentSizeOnly = "size_only"
+)
+
+// Record is one observation of one tool call.
+type Record struct {
+ Schema string `json:"schema"`
+ // ID is this observation's own identity, unique among the records of a
+ // session. It is not the call's id: one call can have several records.
+ ID string `json:"id"`
+ ObservedBy string `json:"observed_by"`
+ Boundary string `json:"boundary"`
+ Session string `json:"session"`
+ // Stream is main or the agent id the call ran under.
+ Stream string `json:"stream"`
+ // Tool is the tool-use id of the call this record observed, the join to
+ // the call's step. ToolName is the name the runtime called it by.
+ Tool string `json:"tool"`
+ ToolName string `json:"tool_name,omitempty"`
+ // Cwd is the working directory the observer reported, which the
+ // collector's session filters are judged by.
+ Cwd string `json:"cwd,omitempty"`
+ Protocol string `json:"protocol"`
+ Server *Server `json:"server,omitempty"`
+ // Time is when the observation ended, RFC 3339.
+ Time string `json:"time"`
+ // DurationMS is the time the observer measured, in milliseconds. It is
+ // the runtime's own measure around the call, so it includes any waiting
+ // before the call started, and it is not the server's own time. Absent
+ // when the observer did not say.
+ DurationMS *int64 `json:"duration_ms,omitempty"`
+ Outcome string `json:"outcome"`
+ // Arguments are what the observer saw sent. Result is what came back,
+ // for a call that returned.
+ Arguments *Content `json:"arguments,omitempty"`
+ Result *Content `json:"result,omitempty"`
+}
+
+// Server is the server a call went to, as the observer names it.
+type Server struct {
+ Name string `json:"name"`
+ // Source is where the runtime's configuration of the server came from,
+ // in the runtime's own word: for Claude Code "user", "project", "local",
+ // or "dynamic" for a server given on the command line.
+ Source string `json:"source,omitempty"`
+}
+
+// Content is what a record keeps of the parameters or the answer: only
+// their size and a digest, never their text.
+type Content struct {
+ State string `json:"state"`
+ // Bytes is the size of the value as compact JSON with sorted keys.
+ Bytes int `json:"bytes"`
+ // SHA256 is the digest of the same bytes. Two observations of one value
+ // have one digest, whichever order its keys were written in, so it can
+ // tell whether the parameters were changed on the way without keeping
+ // them.
+ SHA256 string `json:"sha256,omitempty"`
+}
+
+// Measure keeps the size and the digest of a JSON value. A value that is not
+// JSON, or is followed by more than white space, is measured as the bytes it
+// is.
+func Measure(raw []byte) *Content {
+ b := raw
+ dec := json.NewDecoder(bytes.NewReader(raw))
+ // A number keeps its own text. Decoded as a float64, two integers above
+ // 2^53 would become one value, and a changed argument would keep its
+ // digest.
+ dec.UseNumber()
+ var v any
+ if dec.Decode(&v) == nil {
+ if _, err := dec.Token(); errors.Is(err, io.EOF) {
+ if c, err := canonical(v); err == nil {
+ b = c
+ }
+ }
+ }
+ sum := sha256.Sum256(b)
+ return &Content{State: ContentSizeOnly, Bytes: len(b), SHA256: hex.EncodeToString(sum[:])}
+}
+
+// canonical is compact JSON with every object's keys sorted. A decoded
+// object is a map, and encoding/json writes a map's keys in sorted order, so
+// encoding the decoded value is enough. A number is written as its own text.
+func canonical(v any) ([]byte, error) {
+ var sb strings.Builder
+ enc := json.NewEncoder(&sb)
+ enc.SetEscapeHTML(false)
+ if err := enc.Encode(v); err != nil {
+ return nil, err
+ }
+ return []byte(strings.TrimSuffix(sb.String(), "\n")), nil
+}
+
+// Decode reads a record and reports whether the bytes are one: a JSON
+// object whose schema is this package's.
+func Decode(raw []byte) (*Record, bool) {
+ if len(raw) == 0 || raw[0] != '{' {
+ return nil, false
+ }
+ var r Record
+ if err := json.Unmarshal(raw, &r); err != nil || r.Schema != Schema {
+ return nil, false
+ }
+ return &r, true
+}
+
+// Validate reports a record that cannot be used.
+func (r *Record) Validate() error {
+ switch {
+ case r.Schema != Schema:
+ return fmt.Errorf("execution: schema %q, want %q", r.Schema, Schema)
+ case r.ID == "":
+ return errors.New("execution: record has no id")
+ case r.ObservedBy == "":
+ return errors.New("execution: record does not say who observed it")
+ case r.Boundary == "":
+ return errors.New("execution: record does not say where it was observed")
+ case r.Session == "":
+ return errors.New("execution: record has no session")
+ case r.Stream == "":
+ return errors.New("execution: record has no stream")
+ case r.Tool == "":
+ return errors.New("execution: record names no tool call")
+ case r.Protocol == "":
+ return errors.New("execution: record has no protocol")
+ case r.Time == "":
+ return errors.New("execution: record has no time")
+ case r.Outcome == "":
+ return errors.New("execution: record has no outcome")
+ }
+ return nil
+}
+
+// Marshal writes a record as one JSON line without a trailing newline.
+func (r *Record) Marshal() ([]byte, error) {
+ var sb strings.Builder
+ enc := json.NewEncoder(&sb)
+ enc.SetEscapeHTML(false)
+ if err := enc.Encode(r); err != nil {
+ return nil, err
+ }
+ return []byte(strings.TrimSuffix(sb.String(), "\n")), nil
+}
diff --git a/pkg/execution/execution_test.go b/pkg/execution/execution_test.go
new file mode 100644
index 0000000..4af799a
--- /dev/null
+++ b/pkg/execution/execution_test.go
@@ -0,0 +1,47 @@
+// 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 execution_test
+
+import (
+ "testing"
+
+ "github.com/apache/skywalking-ai-sessionizer/pkg/execution"
+)
+
+// The digest tells whether a value was changed on the way to the server, so
+// one value has one digest however its keys and white space were written,
+// and two values never share one.
+func TestMeasureIsOneDigestPerValue(t *testing.T) {
+ digest := func(s string) string { return execution.Measure([]byte(s)).SHA256 }
+ if digest(`{"b": 1, "a": [true, "x"]}`) != digest(`{"a":[true,"x"],"b":1}`) {
+ t.Error("one value written in two key orders has two digests")
+ }
+ // Two integers above 2^53 are one float64.
+ if digest(`{"id":9007199254740992}`) == digest(`{"id":9007199254740993}`) {
+ t.Error("two different integers have one digest")
+ }
+ if got := execution.Measure([]byte(`{"a":"<b>"}`)); got.Bytes != len(`{"a":"<b>"}`) {
+ t.Errorf("%d bytes; the text must not be escaped", got.Bytes)
+ }
+ // Not one JSON value: measured as the bytes it is.
+ for _, s := range []string{`not json`, `{} {}`} {
+ if got := execution.Measure([]byte(s)); got.Bytes != len(s) {
+ t.Errorf("%q: %d bytes, want the bytes as they are", s, got.Bytes)
+ }
+ }
+}
diff --git a/pkg/sessiondata/sessiondata.go b/pkg/sessiondata/sessiondata.go
index 68db884..20d879b 100644
--- a/pkg/sessiondata/sessiondata.go
+++ b/pkg/sessiondata/sessiondata.go
@@ -62,6 +62,12 @@ const (
// tool-use id. The kind is one word because a landed file's name
// prefix is the kind.
KindChanges Kind = "changes"
+ // KindExecution holds tool execution records, execution/1: what an
+ // observer around a tool call saw it do, such as which MCP server ran
+ // it and how long it took. Like changes, they are evidence beside a
+ // stream's transcript, never steps of it, and a view joins them to
+ // steps by tool-use id. One call can have several.
+ KindExecution Kind = "execution"
)
// PartKind names what a piece of content IS.
@@ -130,6 +136,13 @@ type Part struct {
Of string `json:"of,omitempty"`
// Name is what was called.
Name string `json:"name,omitempty"`
+ // Server and ServerTool say which MCP server a call is addressed to and
+ // which of its tools, read from the name the runtime gave the call. They
+ // are set only where the runtime's naming splits the name exactly. The
+ // server is the runtime's name for it, which can be a form of the
+ // configured name with some characters replaced, not the name itself.
+ Server string `json:"server,omitempty"`
+ ServerTool string `json:"server_tool,omitempty"`
// Failed reports whether a call returned an error, when the runtime said so.
//
// It is a pointer because absence and false are different answers: most
diff --git a/pkg/sessionview/sessionview.go b/pkg/sessionview/sessionview.go
index 774f2de..4690bb2 100644
--- a/pkg/sessionview/sessionview.go
+++ b/pkg/sessionview/sessionview.go
@@ -35,6 +35,7 @@ import (
"encoding/json"
"github.com/apache/skywalking-ai-sessionizer/pkg/changes"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/execution"
"github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
"github.com/apache/skywalking-ai-sessionizer/pkg/sessionflow"
)
@@ -82,6 +83,12 @@ type Conversation struct {
// observations of shell commands and of edits inside subagents. A
// step lists its records under Changes.
WorkspaceChanges []WorkspaceChange `json:"workspace_changes"`
+ // ToolExecutions holds every tool execution record the session's landed
+ // files carry, joined to the step of the call it observed: what an
+ // observer around the call saw it do, such as which MCP server ran it.
+ // A step lists its records under Executions. A record whose call is not
+ // a step of the document is kept, with no step.
+ ToolExecutions []ToolExecution `json:"tool_executions"`
}
// ProviderBody names one landed provider body joined to a call: whether it
@@ -105,6 +112,14 @@ type WorkspaceChange struct {
changes.Record
}
+// ToolExecution is one tool execution record, the step it joins, and where
+// it landed.
+type ToolExecution struct {
+ Step string `json:"step"`
+ Ref sessionflow.Ref `json:"ref"`
+ execution.Record
+}
+
// Head identifies the fold the document was built from.
type Head struct {
Round uint64 `json:"round"`
@@ -281,6 +296,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"`
+ // Executions names the tool execution records joined to this step, by
+ // id, in the order ToolExecutions lists them.
+ Executions []string `json:"executions,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"`
diff --git a/plugins/claude-code/internal/hook/hook.go b/plugins/claude-code/internal/hook/hook.go
index 3d7b3fa..a2a28f5 100644
--- a/plugins/claude-code/internal/hook/hook.go
+++ b/plugins/claude-code/internal/hook/hook.go
@@ -85,6 +85,21 @@ type Input struct {
// the exit code and the standard error text for a shell command.
ToolResponse json.RawMessage `json:"tool_response"`
Error string `json:"error"`
+
+ // MCPServer is set on every event of a call to an MCP server: the
+ // server's name and where its configuration came from. DurationMS is on
+ // the events after a call, the runtime's own measure around it, and
+ // IsInterrupt on a failure a person caused. All three measured on Claude
+ // Code 2.1.282.
+ MCPServer *MCPServer `json:"mcp_server"`
+ DurationMS *int64 `json:"duration_ms"`
+ IsInterrupt *bool `json:"is_interrupt"`
+}
+
+// MCPServer is the MCP server a call went to, as the runtime names it.
+type MCPServer struct {
+ Name string `json:"name"`
+ Source string `json:"source"`
}
// Read decodes one invocation.
diff --git a/plugins/claude-code/internal/output/output.go b/plugins/claude-code/internal/output/output.go
index 2d0b004..1b8a12a 100644
--- a/plugins/claude-code/internal/output/output.go
+++ b/plugins/claude-code/internal/output/output.go
@@ -29,17 +29,29 @@ import (
"time"
"github.com/apache/skywalking-ai-sessionizer/pkg/changes"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/execution"
)
// Dir is the output directory under the plugin's data directory.
const Dir = "output"
-// Path is the file a session's stream is written to.
+// Path is the file a session's stream's change records are written to.
func Path(dataDir, session, stream string) string {
return filepath.Join(dataDir, Dir, session, stream+".jsonl")
}
-// Append writes one record as one line.
+// ExecutionSuffix ends the name of the file a stream's execution records go
+// to. They are a file of their own, so each file holds one kind of record
+// and lands as one kind.
+const ExecutionSuffix = ".execution.jsonl"
+
+// ExecutionPath is the file a session's stream's execution records are
+// written to.
+func ExecutionPath(dataDir, session, stream string) string {
+ return filepath.Join(dataDir, Dir, session, stream+ExecutionSuffix)
+}
+
+// Append writes one change record as one line.
func Append(dataDir string, r *changes.Record) error {
if err := r.Validate(); err != nil {
return err
@@ -48,7 +60,22 @@ func Append(dataDir string, r *changes.Record) error {
if err != nil {
return err
}
- path := Path(dataDir, r.Session, r.Stream)
+ return appendLine(Path(dataDir, r.Session, r.Stream), line)
+}
+
+// AppendExecution writes one execution record as one line.
+func AppendExecution(dataDir string, r *execution.Record) error {
+ if err := r.Validate(); err != nil {
+ return err
+ }
+ line, err := r.Marshal()
+ if err != nil {
+ return err
+ }
+ return appendLine(ExecutionPath(dataDir, r.Session, r.Stream), line)
+}
+
+func appendLine(path string, line []byte) error {
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
return err
}
diff --git a/plugins/claude-code/internal/settings/settings.go b/plugins/claude-code/internal/settings/settings.go
index 7223fab..874db7f 100644
--- a/plugins/claude-code/internal/settings/settings.go
+++ b/plugins/claude-code/internal/settings/settings.go
@@ -47,6 +47,8 @@ type Settings struct {
// ReadOnly says whether a shell command the classifier marks read-only
// skips its scans. On by default.
ReadOnly ReadOnly `yaml:"read_only"`
+ // MCP is the recording of calls to MCP servers.
+ MCP MCP `yaml:"mcp"`
// Tools names the tools observed with a scan: the shells. Editing
// tools are observed from their own response and need no entry.
Tools Tools `yaml:"tools"`
@@ -74,6 +76,16 @@ type ReadOnly struct {
// On reports whether the skip is enabled.
func (r ReadOnly) On() bool { return r.Enabled == nil || *r.Enabled }
+// MCP is the recording of calls to MCP servers: which server, how the call
+// ended, how long it took, and the size of what went each way. It keeps no
+// parameter or answer text, so it is on unless turned off.
+type MCP struct {
+ Enabled *bool `yaml:"enabled"`
+}
+
+// On reports whether calls to MCP servers are recorded.
+func (m MCP) On() bool { return m.Enabled == nil || *m.Enabled }
+
// Tools names the tools observed with a scan.
//
// A runtime with a fixed tool set can be listed by name, which is what Claude
@@ -137,6 +149,9 @@ func Load(dataDir string) (*Settings, error) {
if loaded.ReadOnly.Enabled != nil {
s.ReadOnly = loaded.ReadOnly
}
+ if loaded.MCP.Enabled != nil {
+ s.MCP = loaded.MCP
+ }
if len(loaded.Tools.Scope) > 0 {
s.Tools.Scope = loaded.Tools.Scope
}
diff --git a/plugins/claude-code/main.go b/plugins/claude-code/main.go
index 0773559..226ea72 100644
--- a/plugins/claude-code/main.go
+++ b/plugins/claude-code/main.go
@@ -204,6 +204,11 @@ func runHook(stdin io.Reader, dataDir, projectDir string, now time.Time) (code i
}
}
case hook.PostToolUse, hook.PostToolUseFailure:
+ if in.MCPServer != nil && p.st.MCP.On() {
+ if err := p.execution(); err != nil {
+ p.logf("%s %s execution: %v", in.ToolName, in.ToolUseID, err)
+ }
+ }
switch {
case p.st.IsScopeTool(in.ToolName):
for _, root := range p.roots() {
diff --git a/plugins/claude-code/mcp.go b/plugins/claude-code/mcp.go
new file mode 100644
index 0000000..ff08479
--- /dev/null
+++ b/plugins/claude-code/mcp.go
@@ -0,0 +1,80 @@
+// 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 (
+ "time"
+
+ "github.com/apache/skywalking-ai-sessionizer/pkg/execution"
+ "github.com/apache/skywalking-ai-sessionizer/plugins/claude-code/internal/hook"
+ "github.com/apache/skywalking-ai-sessionizer/plugins/claude-code/internal/output"
+)
+
+// execution records one call to an MCP server, from the hook that runs after
+// it.
+//
+// The hook sees what the transcript does not: which server ran the call and
+// where its configuration came from, how long the runtime waited for it, and
+// what was sent once every hook before the call had run. Measured on Claude
+// Code 2.1.282, the input this hook sees is what the server received, which a
+// hook before the call can change while the transcript keeps the model's own.
+// Only the size and a digest of the input and the answer are kept: the text
+// is in the transcript, or, when rewritten, not the plugin's to keep.
+//
+// The record's id is the call's id and where it was observed, so the same
+// event handed over twice is one record, and another observer of the same
+// call is another.
+func (p *plugin) execution() error {
+ in := p.in
+ r := &execution.Record{
+ Schema: execution.Schema,
+ ID: in.ToolUseID + "/" + execution.BoundaryClientHook,
+ ObservedBy: execution.ObservedByASZPlugin,
+ Boundary: execution.BoundaryClientHook,
+ Session: in.SessionID,
+ Stream: in.Stream(),
+ Tool: in.ToolUseID,
+ ToolName: in.ToolName,
+ Cwd: in.Cwd,
+ Protocol: execution.ProtocolMCP,
+ Server: &execution.Server{Name: in.MCPServer.Name, Source: in.MCPServer.Source},
+ Time: p.now.UTC().Format(time.RFC3339Nano),
+ DurationMS: in.DurationMS,
+ Outcome: outcomeOf(in),
+ }
+ if len(in.ToolInput) > 0 {
+ r.Arguments = execution.Measure(in.ToolInput)
+ }
+ if in.Event == hook.PostToolUse && len(in.ToolResponse) > 0 {
+ r.Result = execution.Measure(in.ToolResponse)
+ }
+ return output.AppendExecution(p.dataDir, r)
+}
+
+// outcomeOf reads how the call ended. The runtime runs the failure hook for
+// an error the server returned, a lost connection and a timeout alike, all
+// measured on Claude Code 2.1.282, and says when a person stopped the call.
+func outcomeOf(in *hook.Input) string {
+ switch {
+ case in.Event == hook.PostToolUse:
+ return execution.OutcomeReturned
+ case in.IsInterrupt != nil && *in.IsInterrupt:
+ return execution.OutcomeInterrupted
+ }
+ return execution.OutcomeFailed
+}
diff --git a/plugins/claude-code/mcp_test.go b/plugins/claude-code/mcp_test.go
new file mode 100644
index 0000000..60e4ce2
--- /dev/null
+++ b/plugins/claude-code/mcp_test.go
@@ -0,0 +1,217 @@
+// 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 (
+ "bufio"
+ "encoding/json"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/apache/skywalking-ai-sessionizer/pkg/execution"
+ "github.com/apache/skywalking-ai-sessionizer/plugins/claude-code/internal/output"
+)
+
+// mcpHooks are the events Claude Code 2.1.282 handed a hook after calls to
+// two MCP servers, one over stdio and one over HTTP: a call that returned, two
+// identical calls at once, an error the server returned, a server that died,
+// a timeout, a call whose input a hook rewrote, and a call inside a subagent.
+// Local paths are replaced; nothing else is changed.
+func mcpHooks(t *testing.T) map[string][]string {
+ t.Helper()
+ f, err := os.Open(filepath.Join("testdata", "mcp-hooks-2.1.282.jsonl"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer f.Close()
+ out := map[string][]string{}
+ sc := bufio.NewScanner(f)
+ sc.Buffer(make([]byte, 1<<20), 1<<20)
+ for sc.Scan() {
+ var c struct {
+ Case string `json:"probe_case"`
+ }
+ if err := json.Unmarshal(sc.Bytes(), &c); err != nil {
+ t.Fatal(err)
+ }
+ out[c.Case] = append(out[c.Case], sc.Text())
+ }
+ return out
+}
+
+func executions(t *testing.T, data, session, stream string) []*execution.Record {
+ t.Helper()
+ b, err := os.ReadFile(output.ExecutionPath(data, session, stream))
+ if err != nil {
+ if os.IsNotExist(err) {
+ return nil
+ }
+ t.Fatal(err)
+ }
+ var out []*execution.Record
+ for _, line := range strings.Split(strings.TrimSpace(string(b)), "\n") {
+ r, ok := execution.Decode([]byte(line))
+ if !ok {
+ t.Fatalf("not an execution record: %s", line)
+ }
+ if err := r.Validate(); err != nil {
+ t.Fatal(err)
+ }
+ out = append(out, r)
+ }
+ return out
+}
+
+func sessionOf(t *testing.T, line string) string {
+ t.Helper()
+ var h struct {
+ Session string `json:"session_id"`
+ }
+ if err := json.Unmarshal([]byte(line), &h); err != nil {
+ t.Fatal(err)
+ }
+ return h.Session
+}
+
+// TestEachMCPCallIsRecordedAsTheRuntimeEndedIt runs every captured event
+// through the hook and reads back what it wrote.
+func TestEachMCPCallIsRecordedAsTheRuntimeEndedIt(t *testing.T) {
+ hooks := mcpHooks(t)
+ now := time.Date(2026, 9, 25, 10, 0, 0, 0, time.UTC)
+ for _, tc := range []struct {
+ kase, stream, server, source, outcome string
+ records int
+ duration int64
+ result bool
+ }{
+ {"echo", "main", "probe", "dynamic", execution.OutcomeReturned, 1, 2, true},
+ // Two identical calls at once are two calls: the runtime gave each
+ // its own id, and the server was sent each one.
+ {"parallel", "main", "probe", "dynamic", execution.OutcomeReturned, 2, 1, true},
+ // An error the server returned, a lost connection and a timeout all
+ // reach the failure hook. None of them keeps an answer.
+ {"fail", "main", "probe", "dynamic", execution.OutcomeFailed, 1, 1, false},
+ {"crash", "main", "probe", "dynamic", execution.OutcomeFailed, 1, 1, false},
+ {"slow", "main", "probe", "dynamic", execution.OutcomeFailed, 1, 2004, false},
+ {"http", "main", "web", "dynamic", execution.OutcomeReturned, 1, 3, true},
+ {"rewrite", "main", "probe", "dynamic", execution.OutcomeReturned, 1, 2, true},
+ // A call inside a subagent is recorded in the subagent's stream.
+ {"child", "ad1f9feae632355a5", "probe", "dynamic", execution.OutcomeReturned, 1, 1, true},
+ } {
+ t.Run(tc.kase, func(t *testing.T) {
+ data := t.TempDir()
+ for _, line := range hooks[tc.kase] {
+ runHook(strings.NewReader(line), data, "/work/"+tc.kase, now)
+ }
+ got := executions(t, data, sessionOf(t, hooks[tc.kase][0]), tc.stream)
+ if len(got) != tc.records {
+ t.Fatalf("%d records, want %d", len(got), tc.records)
+ }
+ ids := map[string]bool{}
+ for _, r := range got {
+ ids[r.ID] = true
+ if r.Server == nil || r.Server.Name != tc.server || r.Server.Source != tc.source {
+ t.Errorf("server %+v, want %s from %s", r.Server, tc.server, tc.source)
+ }
+ if r.Outcome != tc.outcome {
+ t.Errorf("outcome %s, want %s", r.Outcome, tc.outcome)
+ }
+ if r.DurationMS == nil || *r.DurationMS != tc.duration {
+ t.Errorf("duration %v, want %d", r.DurationMS, tc.duration)
+ }
+ if (r.Result != nil) != tc.result {
+ t.Errorf("result kept: %v, want %v", r.Result != nil, tc.result)
+ }
+ if r.Arguments == nil || r.Arguments.State != execution.ContentSizeOnly {
+ t.Errorf("arguments %+v, want their size only", r.Arguments)
+ }
+ if r.Protocol != execution.ProtocolMCP || r.Boundary != execution.BoundaryClientHook ||
+ r.ObservedBy != execution.ObservedByASZPlugin || !strings.HasPrefix(r.ID, r.Tool+"/") {
+ t.Errorf("record %+v", r)
+ }
+ }
+ if len(ids) != tc.records {
+ t.Errorf("%d distinct ids for %d records", len(ids), tc.records)
+ }
+ })
+ }
+}
+
+// TestTheInputRecordedIsWhatTheServerWasSent. A hook before the call
+// rewrote {"text":"original"} to {"text":"rewritten by the hook"}; the
+// server received the second, the transcript kept the first. The record's
+// digest is of what the server received, in any key order.
+func TestTheInputRecordedIsWhatTheServerWasSent(t *testing.T) {
+ data := t.TempDir()
+ line := mcpHooks(t)["rewrite"][0]
+ runHook(strings.NewReader(line), data, "/work/rewrite", time.Now())
+ got := executions(t, data, sessionOf(t, line), "main")
+ if len(got) != 1 {
+ t.Fatalf("%d records", len(got))
+ }
+ sent := execution.Measure([]byte(`{ "text" : "rewritten by the hook" }`))
+ model := execution.Measure([]byte(`{"text":"original"}`))
+ if got[0].Arguments.SHA256 != sent.SHA256 || got[0].Arguments.Bytes != sent.Bytes {
+ t.Errorf("arguments %+v, want those of what the server was sent %+v", got[0].Arguments, sent)
+ }
+ if got[0].Arguments.SHA256 == model.SHA256 {
+ t.Error("the record holds the model's input, not what the server was sent")
+ }
+}
+
+// TestTheSameEventTwiceIsOneRecord. A record's id is the call and where it
+// was observed, so an event handed over again reads back as the same record,
+// which the collector keeps once.
+func TestTheSameEventTwiceIsOneRecord(t *testing.T) {
+ data := t.TempDir()
+ line := mcpHooks(t)["echo"][0]
+ runHook(strings.NewReader(line), data, "/work/echo", time.Now())
+ runHook(strings.NewReader(line), data, "/work/echo", time.Now())
+ got := executions(t, data, sessionOf(t, line), "main")
+ if len(got) != 2 || got[0].ID != got[1].ID {
+ t.Fatalf("records %+v, want two lines with one id", got)
+ }
+}
+
+// TestMCPRecordingCanBeTurnedOff, and a tool that is not an MCP call is
+// never recorded as one.
+func TestMCPRecordingCanBeTurnedOff(t *testing.T) {
+ line := mcpHooks(t)["echo"][0]
+ session := sessionOf(t, line)
+ data := t.TempDir()
+ if err := os.WriteFile(filepath.Join(data, "settings.yaml"), []byte("mcp:\n enabled: false\n"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ runHook(strings.NewReader(line), data, "/work/echo", time.Now())
+ if got := executions(t, data, session, "main"); len(got) != 0 {
+ t.Errorf("%d records with recording turned off", len(got))
+ }
+ data = t.TempDir()
+ runHook(strings.NewReader(hookInput("PostToolUse", session, "", "Read", "toolu_read", `{"file_path":"a"}`, `{"file":{}}`, "")), data, "/work/echo", time.Now())
+ if got := executions(t, data, session, "main"); len(got) != 0 {
+ t.Errorf("%d records for a call that went to no MCP server", len(got))
+ }
+ // The hook recovers a panic and logs it, so no record is also what a
+ // hook that crashed on the missing server would leave.
+ if log, _ := os.ReadFile(filepath.Join(data, "log", "plugin.log")); strings.Contains(string(log), "panic") {
+ t.Errorf("the hook failed on a call that went to no MCP server: %s", log)
+ }
+}
diff --git a/plugins/claude-code/plugin/hooks/hooks.json b/plugins/claude-code/plugin/hooks/hooks.json
index 1718780..71f1a6f 100644
--- a/plugins/claude-code/plugin/hooks/hooks.json
+++ b/plugins/claude-code/plugin/hooks/hooks.json
@@ -1,14 +1,16 @@
{
- "description": "file-changes: a scan around each shell command, the runtime's own patch for each edit inside a subagent, and the retention rules at the session's start and end.",
+ "description": "file changes and MCP calls: a scan around each shell command, the runtime's own patch for each edit inside a subagent, which MCP server each call went to with how it ended and how long it took, and the retention rules at the session's start and end.",
"hooks": {
"PreToolUse": [
{"matcher": "Bash|PowerShell|Monitor", "hooks": [{"type": "command", "command": "asz-changes", "args": ["hook"], "timeout": 60}]}
],
"PostToolUse": [
- {"matcher": "Bash|PowerShell|Monitor|Edit|Write|NotebookEdit", "hooks": [{"type": "command", "command": "asz-changes", "args": ["hook"], "timeout": 60}]}
+ {"matcher": "Bash|PowerShell|Monitor|Edit|Write|NotebookEdit", "hooks": [{"type": "command", "command": "asz-changes", "args": ["hook"], "timeout": 60}]},
+ {"matcher": "mcp__.*", "hooks": [{"type": "command", "command": "asz-changes", "args": ["hook"], "timeout": 10}]}
],
"PostToolUseFailure": [
- {"matcher": "Bash|PowerShell|Monitor", "hooks": [{"type": "command", "command": "asz-changes", "args": ["hook"], "timeout": 60}]}
+ {"matcher": "Bash|PowerShell|Monitor", "hooks": [{"type": "command", "command": "asz-changes", "args": ["hook"], "timeout": 60}]},
+ {"matcher": "mcp__.*", "hooks": [{"type": "command", "command": "asz-changes", "args": ["hook"], "timeout": 10}]}
],
"SessionStart": [
{"hooks": [{"type": "command", "command": "asz-changes", "args": ["hook"], "timeout": 30}]}
diff --git a/plugins/claude-code/testdata/mcp-hooks-2.1.282.jsonl b/plugins/claude-code/testdata/mcp-hooks-2.1.282.jsonl
new file mode 100644
index 0000000..199c726
--- /dev/null
+++ b/plugins/claude-code/testdata/mcp-hooks-2.1.282.jsonl
@@ -0,0 +1,9 @@
+{"session_id":"86cb6fec-8454-476f-9cf1-4d625d1869b7","transcript_path":"/claude/projects/-work-echo/86cb6fec-8454-476f-9cf1-4d625d1869b7.jsonl","cwd":"/work/echo","prompt_id":"ef9ba75f-6625-413e-8c42-1dc0b5f84762","permission_mode":"default","effort":{"level":"medium"},"hook_event_name":"PostToolUse","tool_name":"mcp__probe__echo","tool_input":{"text":"hello"},"tool_response":[{"type":"text","text":"echo: {\"text\": \"hello\"}"}],"tool_use_id":"toolu_probe_1_0","duration_ms":2,"mcp_serve [...]
+{"session_id":"a815fa95-05ba-4b32-bba4-e4b2a1bc0499","transcript_path":"/claude/projects/-work-parallel/a815fa95-05ba-4b32-bba4-e4b2a1bc0499.jsonl","cwd":"/work/parallel","prompt_id":"5940a50e-3880-4339-9b74-3308d7239975","permission_mode":"default","effort":{"level":"medium"},"hook_event_name":"PostToolUse","tool_name":"mcp__probe__echo","tool_input":{"text":"same"},"tool_response":[{"type":"text","text":"echo: {\"text\": \"same\"}"}],"tool_use_id":"toolu_probe_3_0","duration_ms":1,"mcp [...]
+{"session_id":"a815fa95-05ba-4b32-bba4-e4b2a1bc0499","transcript_path":"/claude/projects/-work-parallel/a815fa95-05ba-4b32-bba4-e4b2a1bc0499.jsonl","cwd":"/work/parallel","prompt_id":"5940a50e-3880-4339-9b74-3308d7239975","permission_mode":"default","effort":{"level":"medium"},"hook_event_name":"PostToolUse","tool_name":"mcp__probe__echo","tool_input":{"text":"same"},"tool_response":[{"type":"text","text":"echo: {\"text\": \"same\"}"}],"tool_use_id":"toolu_probe_3_1","duration_ms":1,"mcp [...]
+{"session_id":"df3365d7-7d1b-466a-a9b2-86d883ff07a3","transcript_path":"/claude/projects/-work-fail/df3365d7-7d1b-466a-a9b2-86d883ff07a3.jsonl","cwd":"/work/fail","prompt_id":"ae1d5381-adec-4d27-b657-6381871fa5ec","permission_mode":"default","effort":{"level":"medium"},"hook_event_name":"PostToolUseFailure","tool_name":"mcp__probe__fail","tool_input":{"text":"x"},"tool_use_id":"toolu_probe_5_0","error":"the probe failed on purpose","is_interrupt":false,"duration_ms":1,"mcp_server":{"name [...]
+{"session_id":"ba6a3896-db9e-440c-80aa-9b74bf92bcf5","transcript_path":"/claude/projects/-work-crash/ba6a3896-db9e-440c-80aa-9b74bf92bcf5.jsonl","cwd":"/work/crash","prompt_id":"c8564209-9a99-4a04-9531-df11da814815","permission_mode":"default","effort":{"level":"medium"},"hook_event_name":"PostToolUseFailure","tool_name":"mcp__probe__crash","tool_input":{},"tool_use_id":"toolu_probe_7_0","error":"Connection closed","is_interrupt":false,"duration_ms":1,"mcp_server":{"name":"probe","source [...]
+{"session_id":"50a85f95-a309-4fcf-818b-b22ccdddcc90","transcript_path":"/claude/projects/-work-slow/50a85f95-a309-4fcf-818b-b22ccdddcc90.jsonl","cwd":"/work/slow","prompt_id":"afd81f80-a79c-445a-8aee-eaa535920a24","permission_mode":"default","effort":{"level":"medium"},"hook_event_name":"PostToolUseFailure","tool_name":"mcp__probe__slow","tool_input":{"seconds":6},"tool_use_id":"toolu_probe_9_0","error":"MCP server \"probe\" tool \"slow\" timed out after 2s","is_interrupt":false,"duratio [...]
+{"session_id":"c97015a0-fd1e-4eb4-a5d6-b98a94be6e10","transcript_path":"/claude/projects/-work-http/c97015a0-fd1e-4eb4-a5d6-b98a94be6e10.jsonl","cwd":"/work/http","prompt_id":"24a36692-a35d-4082-893f-daba6d5f3cd1","permission_mode":"default","effort":{"level":"medium"},"hook_event_name":"PostToolUse","tool_name":"mcp__web__echo","tool_input":{"text":"over http"},"tool_response":[{"type":"text","text":"echo: {\"text\": \"over http\"}"}],"tool_use_id":"toolu_probe_11_0","duration_ms":3,"mc [...]
+{"session_id":"c785e32d-5772-4a30-b26b-83752b40f26a","transcript_path":"/claude/projects/-work-rewrite/c785e32d-5772-4a30-b26b-83752b40f26a.jsonl","cwd":"/work/rewrite","prompt_id":"0763d511-80a0-4d12-a60a-db1ac2a78b17","permission_mode":"default","effort":{"level":"medium"},"hook_event_name":"PostToolUse","tool_name":"mcp__probe__echo","tool_input":{"text":"rewritten by the hook"},"tool_response":[{"type":"text","text":"echo: {\"text\": \"rewritten by the hook\"}"}],"tool_use_id":"toolu [...]
+{"session_id":"ba183925-5bf0-407b-946c-583bea4a1d57","transcript_path":"/claude/projects/-work-child/ba183925-5bf0-407b-946c-583bea4a1d57.jsonl","cwd":"/work/child","prompt_id":"ef7e638b-8bf0-4b8d-b9e8-ef20b07c097f","permission_mode":"default","agent_id":"ad1f9feae632355a5","agent_type":"general-purpose","effort":{"level":"medium"},"hook_event_name":"PostToolUse","tool_name":"mcp__probe__echo","tool_input":{"text":"from child"},"tool_response":[{"type":"text","text":"echo: {\"text\": \"f [...]
diff --git a/tests/scenarios/all-kinds.expect.yaml b/tests/scenarios/all-kinds.expect.yaml
index f18e9db..48313ce 100644
--- a/tests/scenarios/all-kinds.expect.yaml
+++ b/tests/scenarios/all-kinds.expect.yaml
@@ -25,6 +25,6 @@ checkpoints:
delta: true
kinds: {epoch.boundary: 1, epoch.summary: 1, talk: 5}
unresolved: {open: 0}
- view: {state: verified, rounds: 2, files: 11} # main twice, helper, its meta, two children, journal, manifest, script; and two rounds
+ view: {state: verified, rounds: 2, files: 12} # main twice, helper, its meta, two children, journal, manifest, script, the execution record; and two rounds
push:
- kinds: [transcript, agent_meta, journal, workflow_manifest, workflow_script, round]
+ kinds: [transcript, agent_meta, journal, workflow_manifest, workflow_script, execution, round]
diff --git a/tests/scenarios/all-kinds.yaml b/tests/scenarios/all-kinds.yaml
index 034ebc8..c18a309 100644
--- a/tests/scenarios/all-kinds.yaml
+++ b/tests/scenarios/all-kinds.yaml
@@ -16,9 +16,9 @@
# under the License.
# Every kind of file a session leaves: the main transcript, a child agent's
-# transcript and meta file, a workflow's journal, manifest and script, and
-# the rounds. The push must carry all six kinds, and the export page's
-# rules hold for each.
+# transcript and meta file, a workflow's journal, manifest and script, the
+# plugin's record of a call to an MCP server, and the rounds. The push must
+# carry all seven kinds, and the export page's rules hold for each.
title: every kind of file
steps:
- input: check everything
@@ -27,6 +27,13 @@ steps:
text: Checking.
tool: {id: t1, name: Bash, input: {command: make check}, result: {text: all green, after: 500ms}}
usage: {in: 2, out: 50, cache_read: 900, cache_write: 100}
+ - call:
+ text: Asking the status server.
+ tool:
+ name: mcp__status__lookup
+ input: {service: checkout}
+ result: {text: "checkout: 3 of 3 ready", after: 400ms}
+ execution: {server: status, source: user, duration: 380ms}
- call:
agent: {name: helper, prompt: look at the tests, steps: [{call: {text: tests look fine, usage: {out: 20}}}], notify: true}
- call:
diff --git a/tests/scenarios/all-kinds.expect.yaml b/tests/scenarios/mcp-calls.expect.yaml
similarity index 53%
copy from tests/scenarios/all-kinds.expect.yaml
copy to tests/scenarios/mcp-calls.expect.yaml
index f18e9db..f1f7893 100644
--- a/tests/scenarios/all-kinds.expect.yaml
+++ b/tests/scenarios/mcp-calls.expect.yaml
@@ -16,15 +16,19 @@
# under the License.
checkpoints:
- launched:
- rounds: 1
- kinds: {stream: 4, agent.call: 2, agent.launch_ack: 2, agent.output: 3, runtime.notification: 1}
- relations: {starts: 3, ends_with: 3, reports: 1}
final:
- rounds: 2
- delta: true
- kinds: {epoch.boundary: 1, epoch.summary: 1, talk: 5}
- unresolved: {open: 0}
- view: {state: verified, rounds: 2, files: 11} # main twice, helper, its meta, two children, journal, manifest, script; and two rounds
+ # Five tools and one agent: the records add no node.
+ kinds: {tool: 5, agent.call: 1}
+ nodes:
+ tool/s2-tool: {kind: tool, attrs: {mcp_server: status, mcp_tool: lookup}}
+ tool/s3-tool: {kind: tool, attrs: {mcp_server: status, mcp_tool: lookup}}
+ # A second "__" makes two splits possible, so none is made.
+ tool/s5-tool: {kind: tool, attrs: {name: mcp__odd__name__twice, mcp_server: null, mcp_tool: null}}
+ tool/auditor-s1-tool: {kind: tool, stream: auditor, attrs: {mcp_server: docs, mcp_tool: read}}
+ view:
+ state: verified
+ executions: 4
+ executions_joined: 4
+ outcomes: {returned: 3, failed: 1}
push:
- kinds: [transcript, agent_meta, journal, workflow_manifest, workflow_script, round]
+ kinds: [transcript, agent_meta, execution, round]
diff --git a/tests/scenarios/mcp-calls.yaml b/tests/scenarios/mcp-calls.yaml
new file mode 100644
index 0000000..be58674
--- /dev/null
+++ b/tests/scenarios/mcp-calls.yaml
@@ -0,0 +1,76 @@
+# 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.
+
+# Calls to MCP servers.
+#
+# Claude Code names a call to an MCP server mcp__<server>__<tool>, and the
+# adapter reads the server and the tool from the name where it splits
+# exactly. The plugin's hook after each call saw which server ran it, where
+# its configuration came from, how the call ended and how long it took, and
+# wrote an execution record the view joins to the step by the tool-use id.
+# The shapes are the ones measured on Claude Code 2.1.282: a call that
+# returned, an error the server returned, a call whose input a hook rewrote
+# before it was sent, a name with a second "__" that is left whole, and a
+# call inside a subagent. The hook of the first call was handed its event
+# twice, so its record is written twice, and it is still one call. None of
+# the records is a step: the same session without them folds to the same
+# nodes.
+title: calls to MCP servers
+steps:
+ - input: check the checkout service and then the payment service
+ - call:
+ text: Asking the status server about checkout.
+ tool:
+ name: mcp__status__lookup
+ input: {service: checkout}
+ result: {text: "checkout: 3 of 3 ready", after: 400ms}
+ execution: {server: status, source: user, duration: 380ms, twice: true}
+ - call:
+ text: And about payment.
+ tool:
+ name: mcp__status__lookup
+ input: {service: payment}
+ result: {text: "the status server could not reach payment", failed: true, after: 2s}
+ execution: {server: status, source: user, duration: 2s}
+ - call:
+ text: Asking again, through a hook that names the region.
+ tool:
+ name: mcp__status__lookup
+ input: {service: payment}
+ result: {text: "payment: 1 of 3 ready", after: 500ms}
+ execution: {server: status, source: user, duration: 450ms, arguments: {service: payment, region: us-west-2}}
+ - call:
+ text: A tool whose name the runtime cannot split.
+ tool:
+ name: mcp__odd__name__twice
+ input: {}
+ result: {text: "ok", after: 100ms}
+ - call:
+ agent:
+ name: auditor
+ prompt: read the release notes
+ notify: true
+ steps:
+ - call:
+ text: Reading them.
+ tool:
+ name: mcp__docs__read
+ input: {page: release-notes}
+ result: {text: "0.6.0: ...", after: 300ms}
+ execution: {server: docs, source: project, duration: 280ms}
+ - call: {text: Read.}
+ - call: {text: Checkout is healthy and payment is not.}
diff --git a/tools/test/collector-check/main.go b/tools/test/collector-check/main.go
index 3a2e680..1a10f44 100644
--- a/tools/test/collector-check/main.go
+++ b/tools/test/collector-check/main.go
@@ -137,9 +137,10 @@ func run(root, logs string) error {
// source and type, and how many requests carried it.
tokens := map[string]int64{}
metricRequests := 0
- // Points from the runtime's exporter, received by the claude-code-otlp
- // adapter and pushed on: the exporter marks them with sender.
- received := 0
+ // The calls to MCP servers and their time, as the Collector received
+ // them, by server, tool and outcome.
+ mcpCalls := map[string]int64{}
+ mcpMS := map[string]int64{}
for _, line := range bytes.Split(raw, []byte("\n")) {
if len(bytes.TrimSpace(line)) == 0 {
continue
@@ -157,8 +158,9 @@ func run(root, logs string) error {
}
for _, sm := range rm.ScopeMetrics {
for _, m := range sm.Metrics {
- if m.Name != "claude_code.token.usage" || !m.Sum.IsMonotonic {
- return fmt.Errorf("a metrics request carries %s monotonic=%v", m.Name, m.Sum.IsMonotonic)
+ want, ok := map[string]string{"agent.token.usage": "tokens", "agent.mcp.calls": "{call}", "agent.mcp.duration": "ms"}[m.Name]
+ if !ok || !m.Sum.IsMonotonic || m.Unit != want {
+ return fmt.Errorf("a metrics request carries %s unit %q monotonic=%v", m.Name, m.Unit, m.Sum.IsMonotonic)
}
for _, dp := range m.Sum.DataPoints {
a := attrs(dp.Attributes)
@@ -166,17 +168,19 @@ func run(root, logs string) error {
if dp.AsInt != "" {
n, _ = strconv.ParseInt(dp.AsInt, 10, 64)
}
+ // The receiver accepts the runtime's exporter and
+ // keeps nothing, so none of its points is sent on.
if a["sender"] == "telemetrygen" {
- // Received from the exporter and forwarded as sent:
- // the unit is the exporter's to set, and this one
- // sets none.
- received++
- continue
+ return fmt.Errorf("a point the exporter sent to asz's receiver reached the Collector: %v", a)
}
- if m.Unit != "tokens" {
- return fmt.Errorf("a derived point of %s carries the unit %q, want tokens", m.Name, m.Unit)
+ switch m.Name {
+ case "agent.token.usage":
+ tokens[a["query_source"]+"/"+a["type"]] += n
+ case "agent.mcp.calls":
+ mcpCalls[a["mcp_server.name"]+"/"+a["mcp_tool.name"]+"/"+a["outcome"]] += n
+ case "agent.mcp.duration":
+ mcpMS[a["mcp_server.name"]+"/"+a["mcp_tool.name"]+"/"+a["outcome"]] += n
}
- tokens[a["query_source"]+"/"+a["type"]] += n
}
}
}
@@ -259,7 +263,7 @@ func run(root, logs string) error {
return fmt.Errorf("the root holds %d files of kind %s, the Collector received %d; it received %v", n, k, kinds[k], kinds)
}
}
- for _, k := range []string{"transcript", "agent_meta", "journal", "workflow_manifest", "workflow_script", "round"} {
+ for _, k := range []string{"transcript", "agent_meta", "journal", "workflow_manifest", "workflow_script", "execution", "round"} {
if rootKinds[k] == 0 {
return fmt.Errorf("the root holds no file of kind %s, so the check exercised less than the export page names", k)
}
@@ -327,12 +331,15 @@ func run(root, logs string) error {
return fmt.Errorf("no %s tokens reached the Collector; it received %v", k, tokens)
}
}
- // The other source reached the Collector too: points a real external
- // exporter sent to asz's receiver, landed, and pushed on.
- if received == 0 {
- return fmt.Errorf("no point from the exporter through the receiver reached the Collector; it received %d metrics requests", metricRequests)
+ // Each execution record the root holds is one call. The all-kinds
+ // scenario has one call to the status server, which returned in 380ms.
+ if want := int64(rootKinds["execution"]); mcpCalls["status/lookup/returned"] != want || want == 0 {
+ return fmt.Errorf("the root holds %d execution records, the Collector received calls %v", want, mcpCalls)
}
- fmt.Printf("ok: %d requests, %d records, kinds %v, resources %v; %d metrics requests, derived tokens %v, %d points received from the exporter\n",
- requests, len(recs), kinds, services, metricRequests, tokens, received)
+ if got, want := mcpMS["status/lookup/returned"], 380*int64(rootKinds["execution"]); got != want {
+ return fmt.Errorf("the Collector received %dms for the calls to the status server, want %d", got, want)
+ }
+ fmt.Printf("ok: %d requests, %d records, kinds %v, resources %v; %d metrics requests, derived tokens %v, MCP calls %v\n",
+ requests, len(recs), kinds, services, metricRequests, tokens, mcpCalls)
return nil
}
diff --git a/tools/test/e2e-collector.sh b/tools/test/e2e-collector.sh
index 70183ac..3c36ead 100755
--- a/tools/test/e2e-collector.sh
+++ b/tools/test/e2e-collector.sh
@@ -100,11 +100,12 @@ YAML
sleep 2
done
- # The second source of the same metric: the runtime's own exporter. A
- # second root runs the receiver adapter alone, a real external exporter
- # sends it the token metric over gRPC and over HTTP, and the spool is
- # pushed to the same Collector. The check then wants points from both
- # sources, told apart by the sender attribute the exporter is given.
+ # The runtime's own exporter, pointed at asz's receiver. A second root runs
+ # the receiver adapter alone, and a real external exporter sends it a
+ # token metric over gRPC and over HTTP. The receiver accepts both and keeps
+ # nothing: every metric asz sends is derived from the landed files, so the
+ # second root's spool stays empty and its push sends nothing. The check
+ # then fails on any point marked with the exporter's sender attribute.
received="$WORK/$protocol-received"
mkdir -p "$received"
cat > "$received/asz.yaml" <<YAML
@@ -116,7 +117,6 @@ adapters:
- name: claude-code-otlp
enabled: true
listen: ":$RECEIVE_PORT"
- metrics: true
export:
otlp:
protocol: $protocol
@@ -142,7 +142,10 @@ YAML
kill "$receiver_pid" >/dev/null 2>&1 || true
wait "$receiver_pid" 2>/dev/null || true
head -3 "$received/collect.log"
- ls "$received/_metrics" | grep -c '\.pb$' | sed 's/^/spooled requests from the exporter: /'
+ if ls "$received/_metrics" 2>/dev/null | grep -q '\.pb$'; then
+ echo "the receiver put the exporter's metrics in the spool; it must accept and drop them"
+ exit 1
+ fi
./bin/asz push -once -config "$received/asz.yaml"
# The file exporter flushes on its own schedule; give it a moment.
for i in $(seq 1 20); do