You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2026/10/12 03:28:23 UTC
(skywalking-ai-sessionizer) branch main updated: Carry only what a reader needs in a round, and keep a round inside its watermark (#61)
This is an automated email from the ASF dual-hosted git repository.
wu-sheng pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-ai-sessionizer.git
The following commit(s) were added to refs/heads/main by this push:
new aa67971 Carry only what a reader needs in a round, and keep a round inside its watermark (#61)
aa67971 is described below
commit aa67971da10c82ef5b840524c9349c6a9510fbe3
Author: 吴晟 Wu Sheng <wu...@foxmail.com>
AuthorDate: Mon Oct 12 11:28:18 2026 +0800
Carry only what a reader needs in a round, and keep a round inside its watermark (#61)
Segments are activity windows only. A segment node carries no attributes:
its talks are its in_segment relations, and its time range is theirs. Its
state, gates_unmet and committable go, with the four gates that set them. A
round carries no clock, so the last window of every session stayed open for
ever, and nothing read the state.
A round no longer repeats what it says elsewhere: the session's conversation
and title_from, an epoch's records, a talk's loops, runs and trigger, a run's
trigger, a call's fragments and a tool's name. The header's bash_runs is
counted from the tool call nodes assembly made. A provider body's manifest no
longer repeats a request's run or the reason a body was kept whole.
A round cut at the byte budget reads nothing past its watermark. Record
lookups reached past through_seq, so a round could hold structure its header
and input digest did not cover. Assembly now works on the index as it stood at
the watermark, and the index is refused when its entries are out of landed
order. The Claude Code provider adapter and the scenario writer extend only an
index that agrees with its saved state, as the other adapters do.
An existing chain continues: its next round revises the changed nodes over no
new file. On a long conversation that revision did not fit one round, so
revisions that need no new evidence now go first, in rounds of their own cut
by their frames, before the round of the new evidence. A session whose parse
fails is parsed again on every pass, and no scenario removal takes it until a
parse succeeds.
---
CLAUDE.md | 7 +-
README.md | 2 +-
cmd/asz/pipeline_test.go | 54 ++++
cmd/asz/refresh.go | 26 +-
docs/en/adapters/claude-code-glossary.md | 4 +-
docs/en/adapters/claude-code.md | 2 +-
docs/en/adapters/kiro-glossary.md | 2 +-
docs/en/adapters/langsmith-glossary.md | 2 +-
docs/en/changes/changes.md | 30 +++
.../concepts-and-designs/conversation-assembly.md | 83 +++---
docs/en/concepts-and-designs/glossary.md | 7 +-
.../unified-conversation-model.md | 10 +-
docs/en/formats/asz-view-example.yaml | 41 +--
docs/en/formats/asz-view.md | 8 +-
docs/en/formats/session-data.md | 5 +-
docs/en/formats/session-flow.md | 4 +-
docs/en/setup/configuration.md | 2 +-
internal/adapters/claudecode/glossary.go | 4 +-
internal/adapters/claudecodeprovider/collect.go | 8 +-
.../adapters/claudecodeprovider/collect_test.go | 46 ++++
internal/adapters/kiro/glossary.go | 2 +-
internal/adapters/langsmith/glossary.go | 2 +-
internal/assemble/assemble.go | 43 ++--
internal/assemble/calls.go | 6 +-
internal/assemble/epoch.go | 2 +-
internal/assemble/provider.go | 6 -
internal/assemble/segment.go | 102 +-------
internal/assemble/streams.go | 17 +-
internal/assemble/talk.go | 24 +-
internal/assemble/tools.go | 9 +-
internal/index/codec.go | 15 +-
internal/index/index_test.go | 85 +++++++
internal/index/lookup.go | 42 ++-
internal/parse/parse.go | 254 +++++++++++-------
internal/scenario/remove/remove.go | 4 +-
internal/scenario/write_sd.go | 6 +-
internal/view/api.go | 18 +-
pkg/model/model.go | 6 +-
pkg/providerbody/providerbody.go | 6 +-
pkg/providerbody/providerbody_test.go | 35 ++-
pkg/providerbody/session.go | 10 +-
pkg/sessionview/sessionview.go | 10 +-
tests/chain/round_upgrade_test.go | 283 +++++++++++++++++++++
.../meta-20261012T013727.579645000Z-000004.sd | 3 +
...transcript-20261012T013727.579645000Z-000003.sd | 4 +
.../changes-20261012T013727.579645000Z-000002.sd | 3 +
...transcript-20261012T013727.579645000Z-000001.sd | 18 ++
tests/chain/testdata/round-upgrade/README.md | 12 +
.../rounds/r000001-4388f7a5d811.sf | 45 ++++
tests/chain/watermark_test.go | 94 +++++++
tests/langchain_reset_test.go | 11 +-
tests/scenarios/fixture.expect.yaml | 4 +-
tests/scenarios/mcp-calls.expect.yaml | 2 +-
53 files changed, 1114 insertions(+), 416 deletions(-)
diff --git a/CLAUDE.md b/CLAUDE.md
index 156d957..3dea773 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -138,9 +138,10 @@ Every source file carries the Apache-2.0 header; `make license-fix` inserts miss
is never rewritten; later evidence produces a new revision in a later round.
- **Entity ids come from stable evidence, never from position.** An id that shifts when an earlier
source is backfilled cannot supersede its own earlier revision.
-- **A round is cut at a byte budget** by narrowing its input window, never by splitting its frames.
- A round covering one landed file may exceed the budget, as a landed file holding one oversize
- record may. `Round.More` says the chain has not reached the index yet.
+- **A round is cut at a byte budget** by narrowing its input window. A round covering one landed
+ file may exceed the budget, as a landed file holding one oversize record may. Revisions that need
+ no new evidence, as after an upgrade, go first in rounds over no new file, cut by their frames.
+ `Round.More` says the chain has not reached the index yet, or has revisions left.
- **A round carries no wall-clock time**, so the same evidence reproduces the same bytes and digest.
Anything mutable or temporal lives in `conversation.state`, outside every digest. Record times
are evidence, not the clock: the header carries the window's first and last record time, and the
diff --git a/README.md b/README.md
index 771dc52..b810747 100644
--- a/README.md
+++ b/README.md
@@ -26,7 +26,7 @@ measured, stored and exported as one unit.
```text
Conversation durable identity · ownership boundary
- └ Segment activity window · the COMMIT unit
+ └ Segment activity window
└ Session observed source provenance
├ ExecutionStream main ordered parent-agent lineage
│ └ Context epoch × N model-context lifetime
diff --git a/cmd/asz/pipeline_test.go b/cmd/asz/pipeline_test.go
index b885b35..274daf5 100644
--- a/cmd/asz/pipeline_test.go
+++ b/cmd/asz/pipeline_test.go
@@ -466,6 +466,60 @@ func TestAnotherBuilderHoldingTheChainIsNotAFailure(t *testing.T) {
}
}
+// A session whose parse failed is parsed again on the next pass, even though
+// nothing new landed, and no removal takes it meanwhile. A chain whose
+// watermark reaches the last landed file can still have revisions left to
+// write, so the watermark alone does not say the chain is complete.
+func TestAFailedParseIsParsedAgain(t *testing.T) {
+ root := t.TempDir()
+ sc, err := scenario.Load(filepath.Join("..", "..", "tests", "scenarios", "growth.yaml"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ at := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+ b, err := scenario.Build(sc, scenario.FormatClaudeCode, root, scenario.Options{At: at, Through: "turn1"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ zone := storage.NewZone(root)
+ ref := testRefresher(t, zone, root)
+ if err := ref.pass(); err != nil {
+ t.Fatalf("the first pass failed: %v", err)
+ }
+
+ // A round that does not read makes the next parse of the session fail.
+ bad := filepath.Join(root, "_conversations", b.Session, "rounds", "r000099-000000000000.sf")
+ if err := os.WriteFile(bad, []byte("not a round\n"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := scenario.Build(sc, scenario.FormatClaudeCode, root, scenario.Options{At: at}); err != nil {
+ t.Fatal(err)
+ }
+ if err := ref.pass(); err == nil {
+ t.Fatal("a pass whose parse failed reported no error")
+ }
+ if !ref.unparsed[b.Session] || !ref.removeInput().Busy[b.Session] {
+ t.Fatal("a session whose parse failed is neither carried to the next pass nor held back from removal")
+ }
+
+ if err := os.Remove(bad); err != nil {
+ t.Fatal(err)
+ }
+ if err := ref.pass(); err != nil {
+ t.Fatalf("the pass after the round was removed failed: %v", err)
+ }
+ if ref.unparsed[b.Session] {
+ t.Fatal("the session is still carried after a parse succeeded")
+ }
+ rounds, err := filepath.Glob(filepath.Join(root, "_conversations", b.Session, "rounds", "*"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(rounds) < 2 {
+ t.Fatalf("%d round(s): what landed while the parse failed was never parsed", len(rounds))
+ }
+}
+
// TestOnlyOnePassSendsAtATime.
//
// Two pipelines over one root is a supported setup, and both push. A push
diff --git a/cmd/asz/refresh.go b/cmd/asz/refresh.go
index 2b41312..7dc4ef1 100644
--- a/cmd/asz/refresh.go
+++ b/cmd/asz/refresh.go
@@ -146,6 +146,13 @@ type refresher struct {
// never moves again. So they are carried until one pass gets the lock.
retry map[string]bool
+ // unparsed holds sessions whose parse failed. They are parsed again on
+ // every pass until one succeeds, for the same reason as retry, and no
+ // removal takes them meanwhile: a chain whose watermark reaches the last
+ // landed file can still have revisions left to write, so the watermark
+ // alone does not say the chain is complete.
+ unparsed map[string]bool
+
// langsmith converts what the langsmith-ingest receiver accepted. It is
// not a collector of sources: there is nothing to poll, only an inbox to
// drain.
@@ -181,7 +188,7 @@ type refresher struct {
// 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, mc config.Metrics, maxRound int64, once bool) (*refresher, error) {
- r := &refresher{srv: srv, zone: zone, maxRound: maxRound, full: true, retry: map[string]bool{}}
+ r := &refresher{srv: srv, zone: zone, maxRound: maxRound, full: true, retry: map[string]bool{}, unparsed: map[string]bool{}}
// One pipeline serves every adapter, so it needs one mode and one
// period, and both have to be settled before anything is built: the
@@ -370,7 +377,14 @@ func (r *refresher) close() {
// root does not say: where it sends, what it sends, and what it is holding
// back to parse again.
func (r *refresher) removeInput() remove.Input {
- in := remove.Input{Busy: r.retry}
+ busy := map[string]bool{}
+ for id := range r.retry {
+ busy[id] = true
+ }
+ for id := range r.unparsed {
+ busy[id] = true
+ }
+ in := remove.Input{Busy: busy}
if r.pusher != nil {
in.Endpoint, in.SendLogs, in.SendMetrics = r.pusher.Endpoint, !r.pusher.NoLogs, !r.pusher.NoMetrics
}
@@ -583,10 +597,13 @@ func (r *refresher) pass() error {
}
// What this pass landed, plus whatever an earlier pass could not take
- // the lock for.
+ // the lock for or failed to parse.
for id := range r.retry {
changed[id] = true
}
+ for id := range r.unparsed {
+ changed[id] = true
+ }
r.retry = map[string]bool{}
var sessions []string
for id := range changed {
@@ -641,8 +658,10 @@ func (r *refresher) pass() error {
continue
}
errs = append(errs, fmt.Errorf("%s: %w", id, perr))
+ r.unparsed[id] = true
continue
}
+ delete(r.unparsed, id)
if res.Changed() || written > 0 {
rounds += written
// The server caches a folded conversation. A new round has to
@@ -715,6 +734,7 @@ func (r *refresher) pass() error {
r.srv.Forget(id)
}
delete(r.retry, id)
+ delete(r.unparsed, id)
delete(r.deferred, id)
}
r.derive = map[string]bool{}
diff --git a/docs/en/adapters/claude-code-glossary.md b/docs/en/adapters/claude-code-glossary.md
index 1a2995d..3202c37 100644
--- a/docs/en/adapters/claude-code-glossary.md
+++ b/docs/en/adapters/claude-code-glossary.md
@@ -89,11 +89,11 @@ than a plausible word that would send a reader looking for a field that does not
| `exact_ambiguous` | an identifier matched several; nothing chose one |
| `exact_unique` | an identifier matched exactly one candidate |
| `hash_only` | only a digest of the bytes was kept |
-| `in_segment` | an activity window chosen for commit |
+| `in_segment` | the activity window a Talk falls in |
| `input_of` | unavailable; the serialized provider request is not written locally |
| `omitted` | deliberately not carried |
| `retries` | unavailable; retry identity is not recorded |
-| `segment` | an activity window chosen for commit |
+| `segment` | an activity window |
| `size_only` | only the length was kept |
| `strong_inference` | no identifier, but a well-founded structural argument |
| `talk` | one readable interaction; derived from where a run was triggered |
diff --git a/docs/en/adapters/claude-code.md b/docs/en/adapters/claude-code.md
index 49d8363..245b322 100644
--- a/docs/en/adapters/claude-code.md
+++ b/docs/en/adapters/claude-code.md
@@ -479,7 +479,7 @@ pass). Two consequences:
conversation continuing past them, and 7,311 of them had their tools run. Nothing waits for them,
and nothing is marked open.
-Every `llm.call` node in a round carries `fragments` and `usage_from: last_fragment_in_line_order`. A
+Every `llm.call` node in a round carries `usage_from: last_fragment_in_line_order`, and its `refs` are its fragments in line order. A
finished call adds `usage: observed_replayable` and `usage_at`, the `{seq, row}` of the fragment the
usage is read from. An unfinished call has `usage: unavailable` and `stop_reason: unavailable`.
[asz.view](../formats/asz-view.md#a-node-in-talks) copies the token counts from the one record
diff --git a/docs/en/adapters/kiro-glossary.md b/docs/en/adapters/kiro-glossary.md
index 96f0999..0cc1688 100644
--- a/docs/en/adapters/kiro-glossary.md
+++ b/docs/en/adapters/kiro-glossary.md
@@ -93,7 +93,7 @@ not there.
| `reports` | — | derived here | no equivalent |
| `retries` | — | derived here | no equivalent |
| `runtime.notification` | — | derived here | no equivalent: a subagent's end is the result of its call |
-| `segment` | — | derived here | an activity window chosen for commit |
+| `segment` | — | derived here | an activity window |
| `size_only` | — | derived here | no equivalent |
| `strong_inference` | — | derived here | derived |
| `summarizes` | — | derived here | not read: nothing joins a summary to its tombstone |
diff --git a/docs/en/adapters/langsmith-glossary.md b/docs/en/adapters/langsmith-glossary.md
index a6dd377..c43b2b1 100644
--- a/docs/en/adapters/langsmith-glossary.md
+++ b/docs/en/adapters/langsmith-glossary.md
@@ -91,7 +91,7 @@ not there.
| `reports` | — | derived here | derived |
| `retries` | — | derived here | no equivalent yet: the client does not say a run retried another |
| `runtime.notification` | — | derived here | no equivalent |
-| `segment` | — | derived here | an activity window chosen for commit |
+| `segment` | — | derived here | an activity window |
| `size_only` | — | derived here | no equivalent |
| `starts` | — | derived here | derived from a delegating tool and the runs beneath it |
| `stream` | — | derived here | no equivalent: a nested agent is a run like any other |
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index e691d22..7c44037 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -287,6 +287,36 @@
last time, and a node's, keep a time before 1970, which the last time used to lose, and no
longer depend on the order Go gives a map when a record is at exactly 1970-01-01T00:00:00Z.
+- Segments are activity windows only. A segment node carries no attributes: its talks are its
+ `in_segment` relations, and its time range is theirs. `state`, `gates_unmet` and `committable` are
+ gone, with the four gates that set them. A round carries no clock, so the last window of every
+ session stayed `open` for ever, even in a session quiet for weeks, and nothing committed a
+ segment or read its state. In `asz.view` a segment keeps `id`, `talks`, `from` and `to`; whether
+ the last one is still going is for a reader, who knows the time.
+- A round no longer repeats what it says elsewhere. Gone from the nodes: the session's
+ `conversation`, which the header names, and `title_from`; an epoch's `records`; a talk's `loops`
+ and `runs`, which its runs are; a talk's and a run's `trigger` (on the main stream a talk's first
+ run is external and every later run a notification, and a child's talk is unknown); a call's
+ `fragments`, which its `refs` are; and a tool's `name`, which its call part carries. A provider
+ body's manifest no longer repeats a request's `run`, which is the record's own, or `why`, which is
+ the text of the part that holds the body whole. The fixture conversation's round is 6% smaller.
+ The header's `bash_runs` is counted as before, from the tool calls assembly made.
+- The next round of an existing conversation revises each session, segment, epoch, talk, run, call and
+ tool node once, without the dropped keys. Nothing is landed again, and the parser and policy stay
+ the same.
+- Revisions that need no new evidence go first, cut to the round budget by their frames. A round
+ cannot be narrowed below one landed file, so on a long conversation an upgrade's revisions made one
+ round past `max_round_bytes`: 12.5 MB for 35,434 nodes on a 17 MB chain, more than the 10 MiB one
+ write may be in a BanyanDB cluster. They are now rounds of their own over no new file, six of at most 2 MiB
+ there, and the round of the new evidence follows.
+- The Claude Code provider adapter extends only an index that agrees with its saved state, as the
+ other adapters do. A pass that wrote the index and stopped before saving its state made the next
+ pass index the same files a second time, out of landed order.
+- A round cut by `max_round_bytes` reads nothing past its watermark, not even through a lookup.
+ Finding the record a reset continues from, or the run a record's parents lead to, could reach
+ records past the round's `through_seq`, so the round could hold structure its own header and input
+ digest did not cover. Assembly now works on the index as it stood at the watermark.
+
## Release
- The `pypi` skill says what a release manager sets up before an upload. PyPI requires two-factor
diff --git a/docs/en/concepts-and-designs/conversation-assembly.md b/docs/en/concepts-and-designs/conversation-assembly.md
index fc520aa..d0ac010 100644
--- a/docs/en/concepts-and-designs/conversation-assembly.md
+++ b/docs/en/concepts-and-designs/conversation-assembly.md
@@ -43,7 +43,7 @@ Eight stages. The order is forced: each depends on the one before it.
| 5. Join child agents | which call started which child stream |
| 6. Cut context epochs | only from an explicit reset record |
| 7. Build Talks and Runs | the conversation a person reads |
-| 8. Propose segments | activity windows, and whether each could be committed |
+| 8. Cut segments | activity windows |
Eight rules in here are the opposite of the obvious choice, and each is the opposite because a
measurement said so.
@@ -112,24 +112,27 @@ child, it would start a new Talk on every run.
### Segments
-Stage 8 proposes segments. It never commits one.
+Stage 8 cuts activity windows. A window says where the work happened. It says nothing about whether
+the work is finished: a round carries no clock, so only a reader, who knows the time, can tell
+whether the last window is still going.
Only the main stream is cut. Its Talks are taken in order of their first record time. A new window
opens where the gap between one Talk's last record and the next Talk's first record is longer than
the idle gap. Only records that resolve to a run count toward a Talk's time range. A window's id
comes from where it opens, the landed position of its first Talk, and not from its place in the
-list. So late evidence that adds an earlier window cannot renumber a later one. The last window is
-`open`, and every earlier one is a `candidate`. A child stream's Talk joins the window whose time
-range holds its first record, as `strong_inference`, because delegated work belongs to the window
-of the Talk that asked for it.
+list. So late evidence that adds an earlier window cannot renumber a later one. A child stream's
+Talk joins the window whose time range holds its first record, as `strong_inference`, because
+delegated work belongs to the window of the Talk that asked for it.
**A negative gap is no gap.** Record times can step backward. In the corpus, 0.315% of consecutive
records step back, by up to 4.06 hours. Provider-error records are the main cause: their time is
when the failed request began, not when the record was written. So a negative difference between
two Talks is treated as no gap, never as a large one.
-**The idle gap is 10 minutes, and a gap only proposes.** A long gap often falls inside a turn rather
-than between two, so it cannot decide on its own. In the corpus:
+**The idle gap is 10 minutes, and it is measured between Talks.** A long quiet period between two
+records often falls inside a Talk rather than between two, so a window is never cut there: stage 8
+compares only one Talk's last record with the next Talk's first. Among the gaps between consecutive
+records in the corpus:
| Idle gap | Gaps that long | Share inside a run | Share inside a Talk |
| --- | --- | --- | --- |
@@ -137,43 +140,20 @@ than between two, so it cannot decide on its own. In the corpus:
| 15 minutes | 444 | unavailable | 30.41% |
| 20 minutes | 302 | 9.60% | 21.52% |
-The Talk column is the one that matters. A window cut inside a Talk splits one interaction between
-two windows. Whether provider-error records count on this timeline moves the run column by five
-percentage points. In asz such a record counts only when its containment parents reach a run,
-because only those records set a Talk's time range.
-
-Twenty minutes would propose 57% fewer candidates. The default stays at 10 minutes because the gates
-below decide what could be committed, and a shorter gap only gives them more candidates to judge.
-How often the gates stop a bad candidate has not been measured.
+The Talk column is why the cut is made between Talks: cut on the gap between records, about two
+cuts in five would fall inside one interaction. Whether provider-error records count on this timeline
+moves the run column by five percentage points. In asz such a record counts only when its
+containment parents reach a run, because only those records set a Talk's time range. A gap of
+twenty minutes would find 57% fewer gaps that long.
The gap is fixed in the build. No configuration key and no command flag changes it. It is part of
-the policy every round header carries, `v1+idle=10m0s`, so windows proposed under one gap never
-fold together with windows proposed under another. [Session Flow](../formats/session-flow.md#header)
-says how a chain refuses a change of policy.
-
-**Four gates decide whether a window could be committed.**
+the policy every round header carries, `v1+idle=10m0s`, so windows cut under one gap never fold
+together with windows cut under another. That is why it stays at 10 minutes:
+[Session Flow](../formats/session-flow.md#header) says how a chain refuses a change of policy.
-| Gate | Passes when |
-| --- | --- |
-| `activity_boundary` | the window is not the last one |
-| `no_crossing_open_operation` | every tool use whose request time falls inside the window has exactly one result, also timed inside the window, and the session holds no open `child_stream` or `notified_child` entry. The second test covers the whole session, not the window. |
-| `lateness_watermark` | the window is not the last one, and the main stream's last record is more than one idle gap after the window ends |
-| `conversation_identity` | the conversation id is not empty. Assembly refuses to run without one, so this gate always passes. |
-
-A segment node carries `state`, `talks`, `gates_unmet` and `committable`. `gates_unmet` lists the
-gates that fail, and `committable` is true when none fail. [asz.view](../formats/asz-view.md) shows
-the same windows to a reader.
-
-`no_crossing_open_operation` is the gate that does real work. It does not exist for a request left
-with no result at the end of a file. In the corpus that is 0 of 25,892 tool requests on the main
-stream, because the runtime writes a result even for a failure. It exists for a result that arrives
-after the quiet period, which in the corpus happens on 85 of the 697 gaps longer than 10 minutes,
-12.2%. Committing that window would freeze the request in one window while its result lands in the
-next.
-
-**Nothing commits a segment.** Assembly proposes windows and reports `committable`, and no command
-acts on it. The [model](unified-conversation-model.md) calls a segment the commit unit, but the
-commit step is not implemented.
+A segment node carries no attributes. Its Talks are its `in_segment` relations, and its time range
+is theirs. [asz.view](../formats/asz-view.md) shows each window to a reader with the count of its
+Talks and its time range.
## What comes out
@@ -268,6 +248,13 @@ first copy. A repeated record therefore cannot become a second node in a later r
the byte budget is assembled again over a narrower window, as
[`max_round_bytes`](../setup/configuration.md#parse) describes.
+**An upgrade's revisions go first, cut by their frames.** A newer parser can change what it writes
+for every node, and a round cannot narrow below one landed file. When the round of one landed file
+is still over the budget and the chain changes even with no new evidence, those changes go first, in
+rounds of their own over no new file. Each holds as many frames as fit the budget, and at least one.
+The round of the new evidence follows. On a 17 MB chain, the revision of 35,434 nodes that came with
+dropping repeated attributes was one round of 12.5 MB, and is six rounds of at most 2 MiB this way.
+
**The watermark is fixed before assembly, and assembly reads nothing past it.** Deciding it
afterwards would let a round hold nodes drawn from evidence that its own header and input digest do
not cover. A collector writing at the same time makes that likely. A parse also refuses to run when
@@ -288,11 +275,12 @@ It holds under any round budget, because the budget changes where the chain is c
says. A chain test builds one session as a single round and again under a one-byte budget, one round
per landed file, and the two folds hold the same entities. The rule does not say that the fold of
round N+1 extends the fold of round N. Later evidence replaces entities that earlier rounds wrote.
+It holds once the chain has reached the index: between the rounds of an upgrade's revisions, which
+are cut by their frames, the fold holds part of the revision, and the next rounds write the rest.
-**Rounds and segments are separate.** The round is the unit of appending. The segment is a proposed
-commit window. A segment is a node like any other, so when a later Talk turns the open window into a
-candidate, that change is an ordinary node revision in whatever round the parse writes. There is no
-separate mechanism for it, and nothing freezes a segment today, as [Segments](#segments) says.
+**Rounds and segments are separate.** The round is the unit of appending. A segment is an activity
+window. When a later Talk joins a window, the round the parse writes adds that Talk's `in_segment`
+relation, and the segment node itself does not change. There is no separate mechanism for it.
The page that `asz view` serves folds the same chain in the same way, every round in order.
[asz.view](../formats/asz-view.md) says what it does with a broken chain, and when it folds again.
@@ -335,8 +323,7 @@ The assembler writes seven kinds of unresolved entry:
Every entry is written `open`, and it resolves when the missing evidence lands. `child_stream` and
`notified_child` are routine in a live session, because a child can be announced before its stream
-lands. While either is open, no segment is committable. A child in a batch that never reports a
-result is normal, and it is not recorded.
+lands. A child in a batch that never reports a result is normal, and it is not recorded.
The assembler makes one change of state. An open entry it no longer produces is superseded as
`resolved`, and [the fold](../formats/session-flow.md#the-fold) keeps it. The scenario
diff --git a/docs/en/concepts-and-designs/glossary.md b/docs/en/concepts-and-designs/glossary.md
index e0889fa..747cde8 100644
--- a/docs/en/concepts-and-designs/glossary.md
+++ b/docs/en/concepts-and-designs/glossary.md
@@ -33,8 +33,9 @@ unchanged around it.
## Structure
-**Segment** — an activity window that can be committed on its own. It sits above Session and cuts
-**across** every stream beneath it, so it is a time window rather than a branch of the tree.
+**Segment** — an activity window. It sits above Session and cuts **across** every stream beneath
+it, so it is a time window rather than a branch of the tree. It says where the work happened, not
+whether the work has finished.
*Not:* a turn, a Talk, or a new conversation.
**Session** — the source-runtime context a record came from, carried as provenance. One conversation
@@ -97,7 +98,7 @@ typed relation carrying its own correlation quality. Cross-stream flow is never
| `result_of` | a record → the tool use whose result it carried; and a stream's final step → the call that started it, when that call received it as its result |
| `follows` | an epoch → the epoch it continues after a reset |
| `summarizes` | a summary → the reset that produced it |
-| `in_segment` | a Talk → the window that will commit it |
+| `in_segment` | a Talk → the activity window it falls in |
| `retries` | an attempt → the attempt it replaced |
| `cancels` | an interruption → the operation it ended |
| `input_of` | a record → the provider call whose input it was part of |
diff --git a/docs/en/concepts-and-designs/unified-conversation-model.md b/docs/en/concepts-and-designs/unified-conversation-model.md
index 4e433d1..b27c224 100644
--- a/docs/en/concepts-and-designs/unified-conversation-model.md
+++ b/docs/en/concepts-and-designs/unified-conversation-model.md
@@ -27,7 +27,7 @@ Everything else is defined relative to those two.
```text
Conversation durable identity · ownership boundary
- └ Segment activity window · the COMMIT unit
+ └ Segment activity window
└ Session observed source provenance
├ ExecutionStream main ordered parent-agent lineage
│ └ Context epoch × N model-context lifetime
@@ -50,9 +50,9 @@ does not run across it; the call that opened it stays a tool step, and the calle
continues unchanged around it. What tells the two apart is what ran directly under the call: a
chain of any kind is a program, only model calls are a plain call. When in doubt it is a child.
-`Segment` sits above `Session` and **cuts across every stream beneath it**: it is a time window
-chosen for commit, not a branch of the stream tree. A segment is committed only once all of its
-streams are safe to freeze.
+`Segment` sits above `Session` and **cuts across every stream beneath it**: it is a time window,
+not a branch of the stream tree. It says where the work happened, not whether the work has
+finished.
`Context epoch` exists **only where a runtime can reset model context in place**. A stream with no
compaction always has exactly one epoch, and its message history accumulates without interruption.
@@ -75,7 +75,7 @@ has to say which of them it means.
| Object | Is | Is **not** |
| --- | --- | --- |
| **Conversation** | stable semantic branch; the unit of storage, analysis and export | a person, an account, a session, a trace, or a time range |
-| **Segment** | independently committable activity window | a turn, a Talk, or a new conversation |
+| **Segment** | activity window across every stream beneath it | a turn, a Talk, or a new conversation |
| **Session** | source-runtime context carried as provenance | an agent, a stream, or a conversation |
| **ExecutionStream** | ordered main, child, auxiliary or judge lineage within one session | timestamp order, or a prompt sequence |
| **Context epoch** | the lifetime of one model context within a stream | a token-count threshold |
diff --git a/docs/en/formats/asz-view-example.yaml b/docs/en/formats/asz-view-example.yaml
index bf8e0e2..9dff6d6 100644
--- a/docs/en/formats/asz-view-example.yaml
+++ b/docs/en/formats/asz-view-example.yaml
@@ -5,7 +5,7 @@ sessions:
- 00000001-0000-4000-8000-000000000001
head:
round: 1
- digest: befb6026a5776f481ed79bb59cc7ff0fbabd36db8e4d174156230ac3051d723d
+ digest: 58f95b7fc54f33b50e6ebdfd7a63a8139e54b4b351c9e75093d96cf897c0c28e
parser: v1
policy: v1+idle=10m0s
summary:
@@ -56,7 +56,7 @@ summary:
withheld: {}
rounds:
- round: 1
- digest: befb6026a5776f481ed79bb59cc7ff0fbabd36db8e4d174156230ac3051d723d
+ digest: 58f95b7fc54f33b50e6ebdfd7a63a8139e54b4b351c9e75093d96cf897c0c28e
previous: null
from_seq: 1
through_seq: 4
@@ -113,7 +113,7 @@ files:
digest: 62f019483e4d0c0d4d0f5f296bec63fbfae663479a2a78d83b45481bd8df7185
from_time: null
through_time: null
- - file: _conversations/00000001-0000-4000-8000-000000000001/rounds/r000001-befb6026a577.sf
+ - file: _conversations/00000001-0000-4000-8000-000000000001/rounds/r000001-58f95b7fc54f.sf
format: sf
kind: round
seq: null
@@ -121,8 +121,8 @@ files:
stream: null
run: null
lines: 45
- bytes: 10556
- digest: 3fde4586b26d5f4ac6725b0680648df44ba9d3dc03206aa1c4442d839a9621c6
+ bytes: 9909
+ digest: 1f3da85a5823383ce8cd6b0db2f91a4b514077851b7d4882700c09628fa6903d
from_time: 1767225600000
through_time: 1767225611100
streams:
@@ -152,8 +152,6 @@ streams:
quality: exact_unique
segments:
- id: segment/at_1_2
- state: open
- committable: false
talks: 3
from: 1767225600000
to: 1767225610700
@@ -166,10 +164,6 @@ talks:
ref:
seq: 1
row: 2
- attrs:
- loops: 2
- runs: 2
- trigger: external
label: run the build
reply: Build passed and tests are green.
runs: 2
@@ -187,8 +181,6 @@ talks:
ref:
seq: 1
row: 2
- attrs:
- trigger: external
children:
- id: input/1/2
kind: message.external
@@ -232,7 +224,6 @@ talks:
- seq: 1
row: 6
attrs:
- fragments: 3
usage: observed_replayable
usage_at:
row: 6
@@ -287,7 +278,6 @@ talks:
row: 7
block: 0
attrs:
- name: Bash
result: available
result_join: exact_unique
timing: unavailable
@@ -316,7 +306,6 @@ talks:
- seq: 1
row: 8
attrs:
- fragments: 1
usage: observed_replayable
usage_at:
row: 8
@@ -345,7 +334,6 @@ talks:
row: 9
block: 0
attrs:
- name: WebSearch
result: available
result_join: exact_unique
timing: unavailable
@@ -372,7 +360,6 @@ talks:
- seq: 1
row: 10
attrs:
- fragments: 1
usage: observed_replayable
usage_at:
row: 10
@@ -401,7 +388,6 @@ talks:
row: 11
block: 0
attrs:
- name: Agent
result: available
result_join: exact_unique
timing: unavailable
@@ -443,8 +429,6 @@ talks:
ref:
seq: 1
row: 12
- attrs:
- trigger: notification
children:
- id: notify/1/12
kind: runtime.notification
@@ -480,7 +464,6 @@ talks:
- seq: 1
row: 13
attrs:
- fragments: 1
usage: observed_replayable
usage_at:
row: 13
@@ -518,7 +501,6 @@ talks:
- seq: 1
row: 14
attrs:
- fragments: 1
stop_reason: unavailable
usage: unavailable
usage_from: last_fragment_in_line_order
@@ -551,10 +533,6 @@ talks:
ref:
seq: 3
row: 1
- attrs:
- loops: 1
- runs: 1
- trigger: unknown
reply: Tests pass.
runs: 1
steps: 3
@@ -571,8 +549,6 @@ talks:
ref:
seq: 3
row: 1
- attrs:
- trigger: external
children:
- id: call/checker-s1-call
kind: llm.call
@@ -586,7 +562,6 @@ talks:
- seq: 3
row: 2
attrs:
- fragments: 1
usage: observed_replayable
usage_at:
row: 2
@@ -650,10 +625,6 @@ talks:
ref:
seq: 1
row: 16
- attrs:
- loops: 1
- runs: 1
- trigger: external
runs: 1
from: 1767225610700
to: 1767225610700
@@ -667,8 +638,6 @@ talks:
ref:
seq: 1
row: 16
- attrs:
- trigger: external
edges:
- type: in_segment
other: segment/at_1_2
diff --git a/docs/en/formats/asz-view.md b/docs/en/formats/asz-view.md
index 70cce55..c1b33f5 100644
--- a/docs/en/formats/asz-view.md
+++ b/docs/en/formats/asz-view.md
@@ -61,9 +61,9 @@ producer of them.
"from_time":1786033245531,"through_time":1786322168715}],
"streams":[{"id":"stream/main","name":"main","role":"main","label":"","parent":"","records":10728,
"steps":7310,"talk":"talk/main/95a1…","named_by":"","opened_by":[]}, …],
- "segments":[{"id":"segment/at_2_7","state":"candidate","committable":true,"talks":24,"from":1786033290556,"to":1786039329174}, …],
+ "segments":[{"id":"segment/at_2_7","talks":24,"from":1786033290556,"to":1786039329174}, …],
"talks":[{"id":"talk/main/95a1…","kind":"talk","parent":"epoch/main/0","stream":"main","at":1786033290556,
- "ref":{"seq":2,"row":7},"attrs":{"loops":1,"runs":1,"trigger":"external"},
+ "ref":{"seq":2,"row":7},
"label":"for meter mal, …","reply":"## Short answer…","runs":1,"steps":132,"tools":51,
"from":1786033290556,"to":1786033717089,"segment":"segment/at_2_7",
"children":[…],"edges":[{"type":"in_segment","other":"segment/at_2_7","dir":"out","quality":"exact_unique","via":"activity window"}]}, …],
@@ -82,7 +82,7 @@ producer of them.
| `rounds` | one per round, in order: `round`, `digest`, `previous` (null on round 1), `from_seq`, `through_seq`, `input_digest`, `from_time`, `through_time` (the record time range of the files the round consumed, null when none carries a time), `verified` |
| `files` | one per `.sd` file, then one per round: `file` (its path on the wire), `format` (`sd` or `sf`), `kind`, `seq` or `round`, `stream` or `run`, `lines`, `bytes`, `digest`, `from_time`, `through_time`. Absent values are null. Together with `rounds`, this is exactly what a rebuild needs. |
| `streams` | one per execution stream: `id`, `name`, `role` (`main` or `child`), `label`, `parent`, `records`, `steps`, `talk`, `named_by`, and `opened_by`, every step the assembler could tie to the start of the stream as `{step, stream, talk, quality}`, in the order those steps happened; several means it did not choose, and neither does a view |
-| `segments` | one per activity window: `id`, `state`, `committable`, `talks`, `from`, `to` |
+| `segments` | one per activity window: `id`, `talks`, `from`, `to`. A window says where the work happened, not whether it has finished; a reader with a clock compares `to` with the time |
| `talks` | one tree per talk, in time order. See the node below |
| `loose` | the runs and steps no talk contains, as trees from their highest such ancestor: a child's output the fold parented to the session because the child's stream opened no talk, for instance. Empty for most conversations. With `talks`, it holds every run and step of the fold within twelve levels of its root, so the document covers the whole session |
| `relations` | one per relation of the fold: `id`, `type`, `from`, `to`, `quality`, `via`, `evidence`; in the order they happened, each at the earliest record its evidence names |
@@ -114,7 +114,7 @@ say.
```json
{"id":"tool/toolu_01KH…","kind":"tool","parent":"call/msg_011C…","stream":"main","at":1786033295667,
"ref":{"seq":2,"row":13,"block":0},"refs":[{"seq":2,"row":13,"block":0},{"seq":2,"row":14,"block":0}],
- "attrs":{"name":"Bash","result":"available","result_join":"exact_unique","timing":"unavailable"},
+ "attrs":{"result":"available","result_join":"exact_unique","timing":"unavailable"},
"text":"{\"command\":\"ls …\"}","state":"available","bytes":145,"flags":["finished"],
"name":"Bash","result":"…","result_state":"available","result_bytes":2048,
"request_to_result_ms":1312,"request_to_result_join":"exact_unique",
diff --git a/docs/en/formats/session-data.md b/docs/en/formats/session-data.md
index 633488b..d90ad3e 100644
--- a/docs/en/formats/session-data.md
+++ b/docs/en/formats/session-data.md
@@ -244,8 +244,7 @@ part that asz builds, the manifest, schema `provider_body/1`:
| `depth` | how many copies lie between this body and one with none. It is at most 32; a body whose base is that deep copies nothing |
| `bytes` limit | a body may claim at most 256 MiB, and a rebuild stops as soon as it passes the size its record claims, so a damaged record cannot make a reader hold more than one body |
| `chain` | the first sixteen hexadecimal characters of the SHA-256 of the request's first message, with the cache marker taken out. It chooses which earlier body a copy comes from. It is storage, and a join must not rely on it |
-| `why` | set when the body is kept whole in one `unknown` part, and why it could not be cut |
-| `model`, `session`, `run`, `call`, `request`, `previous_request` | the values the body carries that join it, as its adapter read them, in the model's words: the session it names; on a request, the run it was sent in; on a response, the call it answers and the provider's id for its request; on a request, the call it was sent to when its adapter knew it, and otherwise the provider request id of the call before it in its chain. Empty when the body carries none. The [adapter's glossary](.. [...]
+| `model`, `session`, `call`, `request`, `previous_request` | the values the body carries that join it, as its adapter read them, in the model's words: the session it names; on a response, the call it answers and the provider's id for its request; on a request, the call it was sent to when its adapter knew it, and otherwise the provider request id of the call before it in its chain. Empty when the body carries none. The [adapter's glossary](../adapters/claude-code.md#provider-bodies) say [...]
A reference is a digest or a record id, never a file and a row, because `asz repack` renumbers files
and rows and carries record lines unchanged. Every reference points at a record that landed before
@@ -268,7 +267,7 @@ several storage stages has to read the earlier files from whichever stage holds
The adapter checks each record before it lands: it writes the record with the Session Data writer,
reads it back and rebuilds the body. A body that does not come back whole, or is not one JSON object
-in valid UTF-8, lands whole in one `unknown` part with a `why`. `asz verify` rebuilds every landed
+in valid UTF-8, lands whole in one `unknown` part, whose text says why it was not cut. `asz verify` rebuilds every landed
body and compares it with `sha256`, and reports a reference to something the session does not hold.
On two captures of Claude Code 2.1.260, the landed `provider_body` files were 25.5% and 21.6% of the
diff --git a/docs/en/formats/session-flow.md b/docs/en/formats/session-flow.md
index 9947493..37c7b9d 100644
--- a/docs/en/formats/session-flow.md
+++ b/docs/en/formats/session-flow.md
@@ -37,10 +37,10 @@ one `commit`.
| `previous` | the digest of round N-1. Empty only for round 1. This is a dependency, not audit metadata: a round is built on the fold of the round it names. |
| `from_seq`, `through_seq` | the landed sequence range this round consumed |
| `input_digest` | binds the round to the landed evidence it read. It is chained rather than recomputed, so producing it stays proportional to new data. How it is computed follows the table. |
-| `parser`, `policy` | the interpretation versions. `policy` includes the idle gap, the quiet period after which a segment can be closed. asz writes `v1+idle=10m0s`, and a round made under another gap has another policy. |
+| `parser`, `policy` | the interpretation versions. `policy` includes the idle gap, the quiet period between two Talks that opens a new activity window. asz writes `v1+idle=10m0s`, and a round made under another gap has another policy. |
| `from_time`, `through_time` | the earliest and the latest record time among the landed files this round consumed, as the runtime wrote them, in UTC. Absent when no record in the window carries a time. |
| `session_from_time`, `session_through_time` | the session's own range as of this round: when it began, and its last activity so far. The `session` node carries the same pair as `from_time` and `through_time` in its attributes; the header repeats it so a reader of the header alone learns it without folding. |
-| `title`, `talks`, `steps`, `streams`, `segments`, `unresolved` | what a list of conversations shows, as of this round: the session's title and the fold's counts of talks, steps, streams, segments and open unresolved references. A receiver lists conversations off the newest round's header, or off the attributes a sender copies from it, and never folds. |
+| `title`, `talks`, `steps`, `streams`, `segments`, `unresolved` | what a list of conversations shows, as of this round: the session's title and the fold's counts of talks, steps, streams, segments and open unresolved references. A receiver lists conversations off the newest round's header, or off the attributes a sender copies from it, and never folds. When revisions that need no new evidence are cut by their frames across several rounds, each of those rounds gives the counts the last o [...]
| `changes`, `lines_added`, `lines_removed` | the workspace change records the landed files carry up to `through_seq`. `changes` counts the distinct records: one for each tool call a producer watched for file changes, whether or not the call changed a file. The other two are the lines those records' diffs add and remove. |
| `llm_calls`, `subagents`, `bash_runs` | the fold's `llm.call` nodes, its streams whose role is `child`, and its `tool` and `agent.call` nodes named `Bash`. A call that started a child agent becomes an `agent.call` and keeps its name, so `bash_runs` does not fall when the child's transcript lands. |
diff --git a/docs/en/setup/configuration.md b/docs/en/setup/configuration.md
index cebe2cc..828a15a 100644
--- a/docs/en/setup/configuration.md
+++ b/docs/en/setup/configuration.md
@@ -90,7 +90,7 @@ data later.
| Key | Default | Meaning |
| --- | --- | --- |
-| `max_round_bytes` | `2097152` | The largest round file the parser writes, 2 MiB. |
+| `max_round_bytes` | `2097152` | The largest round file the parser writes, 2 MiB. A round over it covers fewer landed files, and revisions that need no new evidence are cut by their frames. A round covering one landed file may exceed it. |
## metrics
diff --git a/internal/adapters/claudecode/glossary.go b/internal/adapters/claudecode/glossary.go
index ad4cc0a..8e1000c 100644
--- a/internal/adapters/claudecode/glossary.go
+++ b/internal/adapters/claudecode/glossary.go
@@ -60,7 +60,7 @@ var glossary = model.NewGlossary(Dialect,
// ---- structure; the runtime names a session, a stream and a run, and the model derives the rest ----
model.Term{Unified: model.KindConversation, Note: "supplied, never inferred; defaults to the session id"},
- model.Term{Unified: model.KindSegment, Note: "an activity window chosen for commit"},
+ model.Term{Unified: model.KindSegment, Note: "an activity window"},
model.Term{Unified: model.KindSession, Native: "sessionId", Where: "record, and the transcript filename"},
// The file name, not the record: the adapter reads the in-record agentId
// only on a sidecar or a journal record.
@@ -111,7 +111,7 @@ var glossary = model.NewGlossary(Dialect,
Note: "also stated redundantly as sourceToolAssistantUUID"},
model.Term{Unified: model.RelFollows, Native: "logicalParentUuid", Where: "a compact_boundary record"},
model.Term{Unified: model.RelSummarizes, Native: "parentUuid", Where: "an isCompactSummary record"},
- model.Term{Unified: model.RelInSegment, Note: "an activity window chosen for commit"},
+ model.Term{Unified: model.RelInSegment, Note: "the activity window a Talk falls in"},
model.Term{Unified: model.RelRetries, Note: "unavailable; retry identity is not recorded"},
model.Term{Unified: model.RelCancels, Native: "toolUseResult.interrupted", Where: "a tool result",
Note: "not read, so asz writes no cancels relation today"},
diff --git a/internal/adapters/claudecodeprovider/collect.go b/internal/adapters/claudecodeprovider/collect.go
index a266ba4..62e721d 100644
--- a/internal/adapters/claudecodeprovider/collect.go
+++ b/internal/adapters/claudecodeprovider/collect.go
@@ -453,11 +453,11 @@ func (c *Collector) collectSession(session string, files []*file, sn *seen, st *
if err := state.RecoverNextSeq(sessionDir); err != nil {
return err
}
+ // Only an index that agrees with its saved state is extended; one that
+ // does not is built again. See index.LoadFor for the crash it guards.
var ix *index.Index
- if ixState.Schema == index.Schema {
- if loaded, ok, lerr := index.Load(c.Zone.IndexDir(session), session); lerr == nil && ok {
- ix = loaded
- }
+ if loaded, ok, lerr := index.LoadFor(c.Zone.IndexDir(session), session, ixState); lerr == nil && ok {
+ ix = loaded
}
if ix == nil {
ix = index.New(session)
diff --git a/internal/adapters/claudecodeprovider/collect_test.go b/internal/adapters/claudecodeprovider/collect_test.go
index 920b55e..bcadc6e 100644
--- a/internal/adapters/claudecodeprovider/collect_test.go
+++ b/internal/adapters/claudecodeprovider/collect_test.go
@@ -514,3 +514,49 @@ func TestADamagedFileDoesNotStopOtherSessions(t *testing.T) {
t.Fatal("a damaged file in one session kept another session's body from landing")
}
}
+
+// A pass that wrote the index and stopped before saving its state leaves the
+// index ahead of the state. The next pass builds the index again rather than
+// adding those files a second time: each landed record is in it once, in
+// landed order, which a round's watermark relies on.
+func TestAnIndexAheadOfItsStateIsBuiltAgain(t *testing.T) {
+ f := newFixture(t)
+ f.request(sessionA, "", 2)
+ f.pass(all)
+ statePath := f.zone.IndexStatePath(sessionA)
+ first, err := os.ReadFile(statePath)
+ if err != nil {
+ t.Fatal(err)
+ }
+ f.request(sessionA, "", 3)
+ f.pass(all)
+ // The crash: the index holds the second pass, its state the first.
+ if err := os.Remove(statePath); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(statePath, first, 0o644); err != nil {
+ t.Fatal(err)
+ }
+ f.request(sessionA, "", 4)
+ f.pass(all)
+
+ ix, ok, err := index.Load(f.zone.IndexDir(sessionA), sessionA)
+ if err != nil || !ok {
+ t.Fatalf("index: ok=%v %v", ok, err)
+ }
+ seen := map[[2]uint32]bool{}
+ var last uint32
+ for _, e := range ix.Entries {
+ if e.Seq < last {
+ t.Fatalf("entry at sequence %d follows one at %d", e.Seq, last)
+ }
+ last = e.Seq
+ if seen[[2]uint32{e.Seq, e.Row}] {
+ t.Fatalf("sequence %d row %d is indexed twice", e.Seq, e.Row)
+ }
+ seen[[2]uint32{e.Seq, e.Row}] = true
+ }
+ if len(seen) != 3 {
+ t.Fatalf("%d records indexed, want the 3 bodies landed", len(seen))
+ }
+}
diff --git a/internal/adapters/kiro/glossary.go b/internal/adapters/kiro/glossary.go
index b510a51..066017a 100644
--- a/internal/adapters/kiro/glossary.go
+++ b/internal/adapters/kiro/glossary.go
@@ -49,7 +49,7 @@ var glossary = model.NewGlossary(Dialect,
// ---- structure ----
model.Term{Unified: model.KindConversation, Native: "id", Where: "session.json",
Note: "one session is one conversation; Kiro Crew names no chain between the sessions of one chat"},
- model.Term{Unified: model.KindSegment, Note: "an activity window chosen for commit"},
+ model.Term{Unified: model.KindSegment, Note: "an activity window"},
model.Term{Unified: model.KindSession, Native: "session", Where: "sessions/<workspace key>/sess_<id>"},
model.Term{Unified: model.KindStream, Native: "sub-execution", Where: "sub-executions/<subSessionId>.jsonl",
Note: "the main stream is messages.jsonl"},
diff --git a/internal/adapters/langsmith/glossary.go b/internal/adapters/langsmith/glossary.go
index 8a777fd..ab51f5d 100644
--- a/internal/adapters/langsmith/glossary.go
+++ b/internal/adapters/langsmith/glossary.go
@@ -56,7 +56,7 @@ var glossary = model.NewGlossary(Dialect,
// ---- structure ----
model.Term{Unified: model.KindConversation, Native: "extra.metadata.thread_id", Where: "a run",
Note: "with session_id and conversation_id as the other spellings; supplied, never inferred"},
- model.Term{Unified: model.KindSegment, Note: "an activity window chosen for commit"},
+ model.Term{Unified: model.KindSegment, Note: "an activity window"},
model.Term{Unified: model.KindSession, Native: "session_name", Where: "a run",
Note: "the client's word for the PROJECT; a conversation is the thread key, and a session here is the two together"},
model.Term{Unified: model.KindStream, Note: "no equivalent: a nested agent is a run like any other"},
diff --git a/internal/assemble/assemble.go b/internal/assemble/assemble.go
index f3e8547..38d7580 100644
--- a/internal/assemble/assemble.go
+++ b/internal/assemble/assemble.go
@@ -50,9 +50,8 @@ type Options struct {
Conversation string
Session string
- // IdleGap is how long a session must be quiet before a segment becomes a
- // candidate for commit. A candidate is not a commit: about a third of long
- // gaps fall in the middle of a turn, so the gates in stage 8 decide.
+ // IdleGap is how long the main stream must be quiet between two talks for
+ // a new activity window to open.
IdleGap time.Duration
// ThroughSeq bounds assembly to landed sequences at or below it.
@@ -67,14 +66,13 @@ type Options struct {
ThroughSeq uint64
}
-// DefaultIdleGap is the quiet period that makes a segment a close candidate.
+// DefaultIdleGap is the quiet period between two Talks that opens a new activity
+// window.
//
-// Ten minutes is where the measurement puts the worst trade: about 39% of gaps
-// that long fall inside a Talk rather than between two, against 22% at twenty
-// minutes. It stays the default because the gates in stage 8 are what actually
-// decide, and a shorter gap proposes more candidates for them to judge. It is
-// carried in the round's policy version, so a chain built under one gap can
-// never be extended under another.
+// It is measured between Talks, never between records: in the measured corpus
+// about 39% of the gaps this long between two records fall inside a Talk. It
+// stays at ten minutes because it is carried in the round's policy version,
+// and a chain built under one gap is never extended under another.
const DefaultIdleGap = 10 * time.Minute
// Stats records what assembly saw. Every count carries its denominator, because
@@ -109,9 +107,6 @@ type Stats struct {
ChildrenOrphan int
Notifications int
NotifyUnmatched int
-
- SegmentsOpen int
- SegmentsCandidate int
}
// Result is one assembly of a whole session.
@@ -121,6 +116,11 @@ type Result struct {
Unresolved []sessionflow.Unresolved
Stats Stats
+ // ToolCalls counts the tool and agent call nodes by the name of the tool
+ // called. The round header counts Bash runs from it; the name itself is the
+ // call part's, so no node carries it a second time.
+ ToolCalls map[string]int
+
// ThroughSeq is the highest landed sequence the assembly covered. It is what
// the round header records, and what tells the next round where to resume.
ThroughSeq uint64
@@ -142,6 +142,11 @@ type builder struct {
unres map[string]*sessionflow.Unresolved
stats Stats
+ // toolName is the name of the tool each tool or agent call node calls, by
+ // node id. Two tool uses whose ids give one node id are one node, and so
+ // one call.
+ toolName map[string]string
+
// canonical is the session's entries in landed order with replayed copies
// removed. Every stage works from this, never from Entries directly.
canonical []int32
@@ -198,6 +203,9 @@ func Session(ix *index.Index, opt Options) (*Result, error) {
if opt.IdleGap == 0 {
opt.IdleGap = DefaultIdleGap
}
+ // Every stage and every lookup sees only what the watermark covers.
+ all := len(ix.Entries)
+ ix = ix.Through(opt.ThroughSeq)
b := &builder{
ix: ix, opt: opt,
nodes: map[string]*sessionflow.Node{},
@@ -207,8 +215,10 @@ func Session(ix *index.Index, opt Options) (*Result, error) {
talkOfRun: map[runKey]string{},
runNode: map[runKey]string{},
runAt: map[runKey]sessionflow.Ref{},
+ toolName: map[string]string{},
}
ix.Build()
+ b.stats.Beyond = all - len(ix.Entries)
// The order is forced: each stage depends on the one before it.
b.stage1Canonical() // duplicates removed
@@ -218,7 +228,7 @@ func Session(ix *index.Index, opt Options) (*Result, error) {
b.stage5Spawns() // children joined to the calls that started them
b.stage6Epochs() // context resets cut
b.stage7Talks() // talks and runs built, steps placed
- b.stage8Segments() // commit windows proposed
+ b.stage8Segments() // activity windows cut
b.joinProviderBodies()
b.emitSteps() // leaves written, now that their containers are known
@@ -231,7 +241,10 @@ func Session(ix *index.Index, opt Options) (*Result, error) {
// bytes. Map iteration order in Go is deliberately random, so emitting in map
// order would make every round differ from the last for no reason.
func (b *builder) result() *Result {
- r := &Result{Stats: b.stats}
+ r := &Result{Stats: b.stats, ToolCalls: map[string]int{}}
+ for _, name := range b.toolName {
+ r.ToolCalls[name]++
+ }
for _, n := range b.nodes {
r.Nodes = append(r.Nodes, *n)
}
diff --git a/internal/assemble/calls.go b/internal/assemble/calls.go
index 80f2f12..56b4fc6 100644
--- a/internal/assemble/calls.go
+++ b/internal/assemble/calls.go
@@ -141,10 +141,8 @@ func (b *builder) emitCall(c *providerCall, parent string) {
for _, f := range c.Fragments {
refs = append(refs, ref(f))
}
- a := map[string]any{
- "fragments": len(c.Fragments),
- "usage_from": usageRule,
- }
+ // The fragments are the references above; their count is not repeated.
+ a := map[string]any{"usage_from": usageRule}
if last := c.Last(); last != nil && c.Finished {
a["usage"] = model.ObservedReplayable
a["usage_at"] = map[string]any{"seq": last.Seq, "row": last.Row}
diff --git a/internal/assemble/epoch.go b/internal/assemble/epoch.go
index 77b68e5..f56e4eb 100644
--- a/internal/assemble/epoch.go
+++ b/internal/assemble/epoch.go
@@ -78,7 +78,7 @@ func (b *builder) stage6Epochs() {
b.epochAt[[2]uint32{e.Seq, e.Row}] = ep.NodeID
}
- a := map[string]any{"records": end - start}
+ a := map[string]any{}
var refs []sessionflow.Ref
if boundary != nil {
refs = append(refs, ref(boundary))
diff --git a/internal/assemble/provider.go b/internal/assemble/provider.go
index 04125a3..0c4b1d1 100644
--- a/internal/assemble/provider.go
+++ b/internal/assemble/provider.go
@@ -201,9 +201,6 @@ func (b *builder) bodiesInWindow() []body {
for i := range b.ix.Bodies {
y := &b.ix.Bodies[i]
e := &b.ix.Entries[y.Entry]
- if b.opt.ThroughSeq > 0 && uint64(e.Seq) > b.opt.ThroughSeq {
- continue
- }
if e.Record == 0 || seen[e.Record] {
continue
}
@@ -280,9 +277,6 @@ func (b *builder) streamHasGap(stream uint32) bool {
if e.Stream != stream || !transcriptKind(e.Kind) {
continue
}
- if b.opt.ThroughSeq > 0 && uint64(e.Seq) > b.opt.ThroughSeq {
- continue
- }
if e.Ord > prev+1 {
return true
}
diff --git a/internal/assemble/segment.go b/internal/assemble/segment.go
index 66c6fb1..74ae80e 100644
--- a/internal/assemble/segment.go
+++ b/internal/assemble/segment.go
@@ -24,42 +24,22 @@ import (
"github.com/apache/skywalking-ai-sessionizer/pkg/sessionflow"
)
-// segment is an activity window that can be committed on its own.
+// segment is an activity window: the main stream's talks between two quiet
+// periods, and the delegated work that started inside them.
type segment struct {
NodeID string
- Index int
Talks []*talk
From int64
To int64
- Gates map[string]bool
}
-// Gate names. All four must pass before a segment may be committed.
-const (
- // GateActivityBoundary - the quiet period is real and not an artefact of
- // records arriving out of order.
- GateActivityBoundary = "activity_boundary"
- // GateNoCrossingOpenOperation - no unfinished tool call and no child agent
- // still running. This gate does real work rather than being a safety net: it
- // is what makes "a committed segment is never rewritten" safe to promise.
- // Without it a tool request could sit in a committed segment while its
- // result arrives in the next one.
- GateNoCrossingOpenOperation = "no_crossing_open_operation"
- // GateLatenessWatermark - no late evidence is still expected.
- GateLatenessWatermark = "lateness_watermark"
- // GateConversationIdentity - the conversation this belongs to is known.
- GateConversationIdentity = "conversation_identity"
-)
-
-// Stage 8 - propose segment boundaries.
+// Stage 8 - cut the activity windows.
//
-// A long quiet period creates a CANDIDATE, not a commit. About a third of gaps
-// over ten minutes fall in the middle of a turn - a slow tool, a provider retry
-// - so a rule that split on idle time alone would cut roughly one in three
-// boundaries in the wrong place.
-//
-// This stage proposes and reports. Committing is the caller's decision, because
-// only the caller knows whether collection has finished.
+// A window is where the work happened, nothing more. It says neither that the
+// window is finished nor that it may be committed: a round carries no clock, so
+// only a reader, who knows the time, can tell whether the last window is still
+// going. A later talk that starts within the idle gap of a window's end joins it
+// through a new in_segment relation; the window's own node does not change.
func (b *builder) stage8Segments() {
main := b.mainStream()
if main == nil || len(main.talks) == 0 {
@@ -72,7 +52,7 @@ func (b *builder) stage8Segments() {
sort.SliceStable(talks, func(i, j int) bool { return talks[i].First < talks[j].First })
gap := b.opt.IdleGap.Nanoseconds()
- cur := &segment{Index: 1}
+ cur := &segment{}
flush := func() {
if len(cur.Talks) == 0 {
return
@@ -92,7 +72,7 @@ func (b *builder) stage8Segments() {
// is clamped rather than read as a very large gap in the other direction.
if i > 0 && t.First != 0 && talks[i-1].Last != 0 && t.First-talks[i-1].Last > gap {
flush()
- cur = &segment{Index: cur.Index + 1}
+ cur = &segment{}
}
if cur.From == 0 || (t.First != 0 && t.First < cur.From) {
cur.From = t.First
@@ -104,41 +84,13 @@ func (b *builder) stage8Segments() {
}
flush()
- sessionEnd := main.Last
for _, sg := range b.segments {
- last := sg.Index == len(b.segments)
- sg.Gates = map[string]bool{
- GateActivityBoundary: !last,
- GateConversationIdentity: b.opt.Conversation != "",
- // Late evidence is still expected while the window sits inside the
- // quiet period at the end of the session.
- GateLatenessWatermark: !last && sessionEnd-sg.To > gap,
- GateNoCrossingOpenOperation: b.noOpenOperation(sg),
- }
- state := "candidate"
- if last {
- state = "open"
- b.stats.SegmentsOpen++
- } else {
- b.stats.SegmentsCandidate++
- }
- open := []string{}
- for _, g := range []string{GateActivityBoundary, GateNoCrossingOpenOperation,
- GateLatenessWatermark, GateConversationIdentity} {
- if !sg.Gates[g] {
- open = append(open, g)
- }
- }
+ // A window carries no attributes: its talks are its in_segment
+ // relations, and its time range is theirs.
b.node(sessionflow.Node{
Entity: sessionflow.Entity{ID: sg.NodeID}, Kind: model.KindSegment,
Parent: sessionflow.NodeID("session", b.opt.Session),
Ref: refPtr(sg.Talks[0].At),
- Attrs: attrs(map[string]any{
- "state": state,
- "talks": len(sg.Talks),
- "gates_unmet": open,
- "committable": len(open) == 0,
- }),
})
for _, t := range sg.Talks {
b.relate(model.RelInSegment, t.NodeID, sg.NodeID, model.ExactUnique,
@@ -157,33 +109,3 @@ func (b *builder) stage8Segments() {
}
}
}
-
-// noOpenOperation reports whether every operation started in a window also
-// finished inside it.
-//
-// The thing this catches is not a request left dangling at the end of the file -
-// the runtime writes a result even for a failure, so that is close to
-// non-existent. It is a request whose result arrives on the FAR SIDE of the
-// quiet period, which is roughly one long gap in eight. Committing the window
-// then would freeze the request into a segment that is never rewritten while
-// its result lands in the next one.
-func (b *builder) noOpenOperation(sg *segment) bool {
- inWindow := func(ts int64) bool { return ts >= sg.From && ts <= sg.To }
- for _, t := range b.tools {
- if t.Use == nil || !inWindow(t.Use.TS) {
- continue
- }
- if t.Result == nil || t.Ambiguous {
- return false
- }
- if !inWindow(t.Result.TS) {
- return false
- }
- }
- for _, u := range b.unres {
- if u.Kind == "child_stream" || u.Kind == "notified_child" {
- return false
- }
- }
- return true
-}
diff --git a/internal/assemble/streams.go b/internal/assemble/streams.go
index 2f47694..b26d660 100644
--- a/internal/assemble/streams.go
+++ b/internal/assemble/streams.go
@@ -39,12 +39,10 @@ import (
// measurable share of tool joins look ambiguous when they are not.
func (b *builder) stage1Canonical() {
all := b.ix.Canonical()
- // Everything above the watermark is set aside, not read. The index may
- // already hold records a concurrent collector landed after this round's
- // range was fixed, and a round must describe only the evidence its own
- // header and input digest cover.
+ // The index holds nothing above the watermark: Session narrowed it before
+ // any stage ran.
//
- // A workspace change record is set aside too. It is evidence beside a
+ // A workspace change record is set aside. It is evidence beside a
// stream, not a step of it: it names the tool it belongs to and a view
// joins the two. Letting it into a stream would change that stream's
// record count and time range, so a session with such records would
@@ -55,9 +53,6 @@ func (b *builder) stage1Canonical() {
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 || e.Kind == index.KindExecution {
continue
}
@@ -67,7 +62,6 @@ func (b *builder) stage1Canonical() {
b.stats.ProviderBodies = len(b.bodiesInWindow())
b.stats.Entries = len(b.canonical)
b.stats.Duplicates = len(b.ix.Entries) - len(all)
- b.stats.Beyond = len(all) - len(b.canonical)
}
// streamInfo is one execution stream: an ordered lineage inside a session.
@@ -95,13 +89,14 @@ type streamInfo struct {
// identifier - never from visual nesting, agent identity or timestamps.
func (b *builder) stage2Streams() {
sessionNode := sessionflow.NodeID("session", b.opt.Session)
- a := map[string]any{"conversation": b.opt.Conversation}
+ // The conversation is the round header's; the session node does not repeat
+ // it.
+ a := map[string]any{}
// The runtime writes a title and rewrites it as the work goes on - up to
// ninety times in one session - so the last one is the current one. It is
// the only name anything in this data has.
if t := b.latestTitle(); t != "" {
a["title"] = t
- a["title_from"] = model.ObservedReplayable
}
// When the session began and when it was last active, from the record
// times the runtime wrote. A reader learns both without opening a landed
diff --git a/internal/assemble/talk.go b/internal/assemble/talk.go
index 7df71eb..90d96d4 100644
--- a/internal/assemble/talk.go
+++ b/internal/assemble/talk.go
@@ -272,37 +272,25 @@ func (b *builder) emitTalks() {
if t.Epoch != nil {
parent = t.Epoch.NodeID
}
- trigger := model.TriggerExternal
- if t.Stream.Role != model.StreamMain {
- // A delegated prompt is input to the child, but it did not come from
- // outside the agent - the parent wrote it. The same holds for a
- // prompt the agent wrote for a plain call of its own.
- trigger = model.TriggerUnknown
- }
+ // A talk carries no attributes. Its runs are its children. What
+ // triggered it follows from where it is: a talk on the main stream
+ // starts with input from outside the agent, and a talk on any other
+ // stream with a prompt the agent wrote itself. Its first run is that
+ // trigger, and every later run is a notification.
b.node(sessionflow.Node{
Entity: sessionflow.Entity{ID: t.NodeID}, Kind: model.KindTalk,
Parent: parent, Stream: t.Stream.Name,
Ref: refPtr(t.At),
- Attrs: attrs(map[string]any{
- "loops": len(t.Runs),
- "runs": len(t.Emitted),
- "trigger": trigger,
- }),
})
for _, cyc := range t.Emitted {
id, ok := b.runNode[runKey{t.Stream.ID, cyc}]
if !ok {
continue
}
- trigger := model.TriggerNotification
- if len(t.Runs) > 0 && cyc == t.Runs[0] {
- trigger = model.TriggerExternal
- }
b.node(sessionflow.Node{
Entity: sessionflow.Entity{ID: id}, Kind: model.KindRun,
Parent: t.NodeID, Stream: t.Stream.Name,
- Ref: refPtr(b.runAt[runKey{t.Stream.ID, cyc}]),
- Attrs: attrs(map[string]any{"trigger": trigger}),
+ Ref: refPtr(b.runAt[runKey{t.Stream.ID, cyc}]),
})
}
}
diff --git a/internal/assemble/tools.go b/internal/assemble/tools.go
index c6799af..a73e106 100644
--- a/internal/assemble/tools.go
+++ b/internal/assemble/tools.go
@@ -145,11 +145,9 @@ func (b *builder) manifestFor(t *toolUse) *index.Entry {
// from record timestamps, timing is unavailable rather than guessed from the
// gap between two records.
func (b *builder) emitTool(t *toolUse, parent string) {
- a := map[string]any{
- "name": t.Name,
- // The request lives in the tool_use block; the reference below locates it.
- "timing": model.Unavailable,
- }
+ // The tool's name is the call part's, which the request reference below
+ // locates; it is not repeated here.
+ a := map[string]any{"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
@@ -195,5 +193,6 @@ func (b *builder) emitTool(t *toolUse, parent string) {
Parent: parent, Stream: t.Stream.Name,
Ref: refPtr(refs[0]), Refs: refs, Attrs: attrs(a),
})
+ b.toolName[t.NodeID] = t.Name
b.stats.Steps++
}
diff --git a/internal/index/codec.go b/internal/index/codec.go
index c003a0d..1299f97 100644
--- a/internal/index/codec.go
+++ b/internal/index/codec.go
@@ -131,9 +131,9 @@ func (ix *Index) Write(dir string) error {
return os.Rename(tmp, path)
}
-// Load reads an index written by Write. A missing file, a foreign file or a
-// schema mismatch returns ok=false rather than an error: the index is derived,
-// so the caller rebuilds instead of migrating.
+// Load reads an index written by Write. A missing file, a foreign file, a
+// schema mismatch or entries out of landed order return ok=false rather than an
+// error: the index is derived, so the caller rebuilds instead of migrating.
func Load(dir, session string) (ix *Index, ok bool, err error) {
data, err := os.ReadFile(filepath.Join(dir, "entries.bin"))
if err != nil {
@@ -213,6 +213,15 @@ func Load(dir, session string) (ix *Index, ok bool, err error) {
decodeBody(b, &bodies[i])
}
+ // Entries are in landed order, and a round's watermark takes the entries
+ // at or below it as a prefix (Through). An index whose sequences go back
+ // was extended twice over the same files; it is built again rather than
+ // read.
+ for i := 1; i < len(entries); i++ {
+ if entries[i].Seq < entries[i-1].Seq {
+ return nil, false, nil
+ }
+ }
return &Index{
Session: session, Strings: loadInterner(strs),
Entries: entries, Blocks: blocks, Bodies: bodies,
diff --git a/internal/index/index_test.go b/internal/index/index_test.go
index 8576604..d8c1aa5 100644
--- a/internal/index/index_test.go
+++ b/internal/index/index_test.go
@@ -264,3 +264,88 @@ func TestProviderBodiesAreIndexedAndSurviveTheRoundTrip(t *testing.T) {
t.Errorf("response entry: %+v", e)
}
}
+
+// Through keeps the entries at or below a sequence and builds its lookups over
+// them alone, leaving the index it came from as it was.
+func TestThroughBoundsEveryLookup(t *testing.T) {
+ ix := build(t)
+ cut := ix.Through(1)
+ if _, ok := cut.EntryByRecord("u-c1"); ok {
+ t.Error("a record landed at sequence 2 was found through sequence 1")
+ }
+ if got := cut.Streams(); len(got) != 1 {
+ t.Errorf("Streams through 1: %v, want main only", got)
+ }
+ if got := cut.ProviderCall("msg_1"); len(got) != 2 {
+ t.Errorf("ProviderCall through 1: %d, want 2", len(got))
+ }
+ if got := cut.ToolBlocks("toolu_1"); len(got) != 2 {
+ t.Errorf("ToolBlocks through 1: %d, want 2", len(got))
+ }
+ if _, ok := ix.EntryByRecord("u-c1"); !ok {
+ t.Error("Through changed the index it was called on")
+ }
+ if ix.Through(0) != ix || ix.Through(2) != ix || ix.Through(9) != ix {
+ t.Error("a bound at or past the last entry must return the index itself")
+ }
+}
+
+// An index whose sequences go back was extended twice over the same files, by
+// a pass that wrote the index and stopped before saving its state. It is not
+// read: the caller builds it again from the landed files.
+func TestAnIndexOutOfLandedOrderIsNotRead(t *testing.T) {
+ ix := index.New("s")
+ in := ix.Strings
+ for _, seq := range []uint32{1, 2, 1, 2} {
+ ix.Append(index.Entry{Seq: seq, Row: 1, Record: in.ID("r")})
+ }
+ dir := t.TempDir()
+ if err := ix.Write(dir); err != nil {
+ t.Fatal(err)
+ }
+ if _, ok, err := index.Load(dir, "s"); err != nil || ok {
+ t.Fatalf("an index out of landed order was read: ok=%v %v", ok, err)
+ }
+}
+
+// The first lookup on an index builds its maps, whichever lookup it is.
+func TestAFirstLookupBuildsTheMaps(t *testing.T) {
+ fresh := func() *index.Index {
+ ix := index.New("s")
+ in := ix.Strings
+ ix.Append(index.Entry{Seq: 1, Row: 1, Record: in.ID("a"), Run: in.ID("r"), Batch: in.ID("b"),
+ Tool: in.ID("t"), Child: in.ID("c")})
+ ix.Append(index.Entry{Seq: 2, Row: 1, Record: in.ID("z")})
+ return ix.Through(1)
+ }
+ for name, lookup := range map[string]func(*index.Index) []*index.Entry{
+ "Run": func(ix *index.Index) []*index.Entry { return ix.Run("r") },
+ "Batch": func(ix *index.Index) []*index.Entry { return ix.Batch("b") },
+ "AboutTool": func(ix *index.Index) []*index.Entry { return ix.AboutTool("t") },
+ "AboutChild": func(ix *index.Index) []*index.Entry { return ix.AboutChild("c") },
+ } {
+ if got := lookup(fresh()); len(got) != 1 {
+ t.Errorf("%s as the first lookup found %d entries, want 1", name, len(got))
+ }
+ }
+}
+
+// Through keeps the blocks and bodies of the entries it keeps, and only those.
+// A record appended to the prefix does not write over the index it came from.
+func TestThroughKeepsTheBlocksAndBodiesOfItsEntries(t *testing.T) {
+ ix := index.New("s")
+ in := ix.Strings
+ ix.AppendRecord(index.Entry{Seq: 1, Row: 1, Record: in.ID("a"), Kind: index.KindProviderBody},
+ []index.Block{{Kind: index.BlockOther}}, &index.Body{Role: index.BodyRoleRequest})
+ ix.AppendRecord(index.Entry{Seq: 2, Row: 1, Record: in.ID("b"), Kind: index.KindProviderBody},
+ []index.Block{{Kind: index.BlockOther}, {Kind: index.BlockOther}}, &index.Body{Role: index.BodyRoleResponse})
+ cut := ix.Through(1)
+ if len(cut.Entries) != 1 || len(cut.Blocks) != 1 || len(cut.Bodies) != 1 || cut.Bodies[0].Entry != 0 {
+ t.Fatalf("through 1: %d entries, %d blocks, %d bodies", len(cut.Entries), len(cut.Blocks), len(cut.Bodies))
+ }
+ cut.AppendRecord(index.Entry{Seq: 2, Row: 1, Record: in.ID("c")}, []index.Block{{Kind: index.BlockText}},
+ &index.Body{Role: index.BodyRoleRequest})
+ if ix.Entries[1].Record != in.ID("b") || ix.Blocks[1].Kind != index.BlockOther || ix.Bodies[1].Role != index.BodyRoleResponse {
+ t.Fatal("appending to the prefix wrote over the index it came from")
+ }
+}
diff --git a/internal/index/lookup.go b/internal/index/lookup.go
index 9b84591..9715570 100644
--- a/internal/index/lookup.go
+++ b/internal/index/lookup.go
@@ -99,6 +99,34 @@ func (ix *Index) Append(e Entry, blocks ...Block) {
ix.byRecord = nil // invalidate; Build rebuilds on demand
}
+// Through returns the index as it stood when the landed file with sequence
+// seq was the last one: the entries at or below seq, with their blocks and
+// bodies, and lookup maps built over those alone.
+//
+// A round assembled under a watermark must draw on nothing past it, or it could
+// hold structure that its own header and input digest do not cover. Bounding
+// each lookup separately would leave the next new lookup unbounded; bounding
+// the index bounds them all. Entries are kept in landed order, so the entries
+// at or below seq are a prefix. Zero, or a seq at or above the last entry's,
+// returns ix itself.
+func (ix *Index) Through(seq uint64) *Index {
+ n := len(ix.Entries)
+ if seq == 0 || n == 0 || uint64(ix.Entries[n-1].Seq) <= seq {
+ return ix
+ }
+ k := sort.Search(n, func(i int) bool { return uint64(ix.Entries[i].Seq) > seq })
+ blocks := 0
+ if k > 0 {
+ last := ix.Entries[k-1]
+ blocks = int(last.BlockFirst + last.BlockCount)
+ }
+ bodies := sort.Search(len(ix.Bodies), func(i int) bool { return int(ix.Bodies[i].Entry) >= k })
+ return &Index{
+ Session: ix.Session, Strings: ix.Strings,
+ Entries: ix.Entries[:k:k], Blocks: ix.Blocks[:blocks:blocks], Bodies: ix.Bodies[:bodies:bodies],
+ }
+}
+
// Build constructs the lookup maps. It is idempotent.
func (ix *Index) Build() {
if ix.byRecord != nil {
@@ -262,7 +290,7 @@ func (ix *Index) Streams() []string {
// Batch returns every entry the runtime filed under one orchestration.
func (ix *Index) Batch(batchID string) []*Entry {
- return ix.resolve(ix.byBatch, batchID)
+ return ix.resolve(&ix.byBatch, batchID)
}
// IsDuplicate reports whether an entry is a replayed copy of an earlier one.
@@ -290,14 +318,14 @@ func (ix *Index) Canonical() []int32 {
// Run returns every entry of one agent loop, in landed order.
func (ix *Index) Run(runID string) []*Entry {
- return ix.resolve(ix.byRun, runID)
+ return ix.resolve(&ix.byRun, runID)
}
// AboutTool returns every record that names a tool use outside its content
// blocks - a child's sidecar naming the call that created it, a notification
// naming the call it completes.
func (ix *Index) AboutTool(toolUseID string) []*Entry {
- return ix.resolve(ix.byToolRef, toolUseID)
+ return ix.resolve(&ix.byToolRef, toolUseID)
}
// AboutChild returns every record that names a child agent stream, whether it
@@ -307,16 +335,18 @@ func (ix *Index) AboutTool(toolUseID string) []*Entry {
// session: a nested child notifies the main stream while the call that created
// it lives in an agent file, so resolution scoped to one file would miss it.
func (ix *Index) AboutChild(stream string) []*Entry {
- return ix.resolve(ix.byChild, stream)
+ return ix.resolve(&ix.byChild, stream)
}
-func (ix *Index) resolve(m map[uint32][]int32, key string) []*Entry {
+// resolve looks a key up in one of the lookup maps. It takes the map's field,
+// not the map: Build fills the field, and a map read before Build is nil.
+func (ix *Index) resolve(m *map[uint32][]int32, key string) []*Entry {
ix.Build()
id, ok := ix.Strings.Lookup(key)
if !ok || id == 0 {
return nil
}
- idxs := m[id]
+ idxs := (*m)[id]
out := make([]*Entry, 0, len(idxs))
for _, i := range idxs {
out = append(out, &ix.Entries[i])
diff --git a/internal/parse/parse.go b/internal/parse/parse.go
index 879f9af..a005aa7 100644
--- a/internal/parse/parse.go
+++ b/internal/parse/parse.go
@@ -49,13 +49,14 @@ import (
const Parser = "v1"
// policyBase is the version of the choices that are settings rather than
-// interpretation: the segment gates, and how a commit window is proposed.
+// interpretation: how activity windows are cut.
const policyBase = "v1"
-// DefaultIdleGap is the quiet period that makes a segment a close candidate.
+// DefaultIdleGap is the quiet period between two Talks that opens a new
+// activity window.
//
// It is part of the POLICY, not a free parameter, because changing it changes
-// which segments a round proposes. A chain is one interpretation of one body of
+// which windows a round cuts. A chain is one interpretation of one body of
// evidence, so a round produced under a different gap must not fold together
// with rounds produced under this one.
const DefaultIdleGap = assemble.DefaultIdleGap
@@ -121,9 +122,10 @@ type Round struct {
FromSeq uint64
ThroughSeq uint64
- // More reports that the round was cut at the byte budget and evidence
- // past ThroughSeq is indexed already. A caller that wants the chain to
- // reach the index calls Session again until More is false.
+ // More reports that the round was cut at the byte budget: evidence past
+ // ThroughSeq is indexed already, or revisions that need no new evidence
+ // are left. A caller that wants the chain to reach the index calls Session
+ // again until More is false.
More bool
Nodes int
@@ -263,33 +265,40 @@ func Session(z *storage.Zone, opt Options) (*Round, error) {
indexed := through
round := view.Round + 1
var (
+ res *assemble.Result
d *delta
data []byte
digest string
inputDigest string
+ // revising is set when the round carries only what changes with no new
+ // evidence, and frames is how many of the delta's frames it writes.
+ revising bool
+ frames int
)
for {
- res, err := assemble.Session(ix, assemble.Options{
- Conversation: opt.Conversation,
- Session: opt.Session,
- IdleGap: opt.IdleGap,
- ThroughSeq: through,
- })
- if err != nil {
- return nil, err
- }
- out.Stats = res.Stats
- d = diff(view, res)
-
- // A round is written when EVIDENCE advanced, not only when the
- // structure changed. Evidence that produced no new entity - a replayed
- // block, a record type nothing reads - still has to move the chain's
- // watermark, or the next round re-reads it forever and "no round" comes
- // to mean two different things: nothing new arrived, and nothing new
- // mattered.
- if d.empty() && through <= view.ThroughSeq {
- out.ThroughSeq = view.ThroughSeq
- return out, nil
+ if !revising {
+ res, err = assemble.Session(ix, assemble.Options{
+ Conversation: opt.Conversation,
+ Session: opt.Session,
+ IdleGap: opt.IdleGap,
+ ThroughSeq: through,
+ })
+ if err != nil {
+ return nil, err
+ }
+ out.Stats = res.Stats
+ d = diff(view, res)
+
+ // A round is written when EVIDENCE advanced, not only when the
+ // structure changed. Evidence that produced no new entity - a
+ // replayed block, a record type nothing reads - still has to move
+ // the chain's watermark, or the next round re-reads it forever and
+ // "no round" comes to mean two different things: nothing new
+ // arrived, and nothing new mattered.
+ if d.empty() && through <= view.ThroughSeq {
+ out.ThroughSeq = view.ThroughSeq
+ return out, nil
+ }
}
// The input digest chains from the LAST ROUND's own header, not from
@@ -300,55 +309,58 @@ func Session(z *storage.Zone, opt Options) (*Round, error) {
if err != nil {
return nil, err
}
- fromTime, throughTime := windowTimes(ix, view.ThroughSeq, through)
- ch := changeStats(ix, through)
- sessionFrom, sessionThrough, title := sessionRange(res, opt.Session)
- w, err := sessionflow.NewWriter(sessionflow.Header{
- Conversation: opt.Conversation, Session: opt.Session,
- Round: round, Previous: view.Digest,
- FromSeq: view.ThroughSeq + 1, ThroughSeq: through,
- InputDigest: inputDigest, Parser: Parser, Policy: policy,
- FromTime: fromTime, ThroughTime: throughTime,
- SessionFromTime: sessionFrom, SessionThroughTime: sessionThrough,
- Title: title, Talks: countKind(res, model.KindTalk), Steps: countSteps(res), Streams: countKind(res, model.KindStream),
- Segments: countKind(res, model.KindSegment), Unresolved: countOpen(res),
- Changes: &ch.records, LinesAdded: &ch.added, LinesRemoved: &ch.removed,
- LLMCalls: ptr(countKind(res, model.KindLLMCall)), Subagents: ptr(countChildStreams(res)),
- BashRuns: ptr(countTool(res, "Bash")),
- })
+ h := roundHeader(ix, res, opt, view, round, through, inputDigest, policy)
+ frames = d.frames()
+ data, digest, err = encodeRound(h, d, frames)
if err != nil {
return nil, err
}
- for _, n := range d.nodes {
- if err := w.Node(n); err != nil {
- return nil, err
- }
+ if int64(len(data)) <= budget {
+ break
+ }
+ // A round over the budget is built again over a narrower window,
+ // halved each time, until it fits or covers a single landed file.
+ // Nothing is lost: the evidence past the window is still indexed and
+ // the next round starts where this one stops.
+ if through > out.FromSeq {
+ through = out.FromSeq + (through-out.FromSeq)/2
+ continue
}
- for _, r := range d.relations {
- if err := w.Relation(r); err != nil {
+ // One landed file and still over the budget. What differs already at
+ // the chain's own watermark needs no new evidence: an upgrade that
+ // revises what earlier rounds wrote, on every node of a long
+ // conversation. It goes first, in rounds of its own over no new file,
+ // so the round of the evidence carries only what the evidence changed.
+ if !revising && through == out.FromSeq && view.Round > 0 {
+ r0, err := assemble.Session(ix, assemble.Options{
+ Conversation: opt.Conversation,
+ Session: opt.Session,
+ IdleGap: opt.IdleGap,
+ ThroughSeq: view.ThroughSeq,
+ })
+ if err != nil {
return nil, err
}
+ if d0 := diff(view, r0); !d0.empty() {
+ revising, res, d, through = true, r0, d0, view.ThroughSeq
+ out.Stats = r0.Stats
+ continue
+ }
}
- for _, u := range d.unresolved {
- if err := w.Unresolved(u); err != nil {
+ // A round over no new file carries revisions only, and is cut by its
+ // frames: as many as fit, at least one. The next round writes the
+ // rest. A round over one landed file is not cut: it may exceed the
+ // budget, as a landed file holding one oversize record may.
+ if through < out.FromSeq {
+ frames, data, digest, err = fitFrames(h, d, budget)
+ if err != nil {
return nil, err
}
}
- data, digest, err = w.Close()
- if err != nil {
- return nil, err
- }
- // A round over the budget is built again over a narrower window,
- // halved each time, until it fits or covers a single landed file.
- // Nothing is lost: the evidence past the window is still indexed and
- // the next round starts where this one stops.
- if int64(len(data)) <= budget || through <= out.FromSeq {
- break
- }
- through = out.FromSeq + (through-out.FromSeq)/2
+ break
}
out.ThroughSeq = through
- out.More = through < indexed
+ out.More = through < indexed || frames < d.frames()
path, err := chain.Publish(round, digest, data)
if err != nil {
return nil, err
@@ -365,11 +377,76 @@ func Session(z *storage.Zone, opt Options) (*Round, error) {
}
out.Number, out.Digest, out.Path = round, digest, path
- out.Nodes, out.Relations, out.Unresolved = len(d.nodes), len(d.relations), len(d.unresolved)
- out.Tombstones = d.tombstones
+ out.Nodes, out.Relations, out.Unresolved, out.Tombstones = d.written(frames)
return out, nil
}
+// roundHeader is the header of a round through the given watermark. Its counts
+// are the assembly's: when revisions are split across several rounds over no
+// new file, each of them names the conversation those rounds lead to.
+func roundHeader(ix *index.Index, res *assemble.Result, opt Options, view *sessionflow.View,
+ round, through uint64, inputDigest, policy string) sessionflow.Header {
+ fromTime, throughTime := windowTimes(ix, view.ThroughSeq, through)
+ ch := changeStats(ix, through)
+ sessionFrom, sessionThrough, title := sessionRange(res, opt.Session)
+ return sessionflow.Header{
+ Conversation: opt.Conversation, Session: opt.Session,
+ Round: round, Previous: view.Digest,
+ FromSeq: view.ThroughSeq + 1, ThroughSeq: through,
+ InputDigest: inputDigest, Parser: Parser, Policy: policy,
+ FromTime: fromTime, ThroughTime: throughTime,
+ SessionFromTime: sessionFrom, SessionThroughTime: sessionThrough,
+ Title: title, Talks: countKind(res, model.KindTalk), Steps: countSteps(res), Streams: countKind(res, model.KindStream),
+ Segments: countKind(res, model.KindSegment), Unresolved: countOpen(res),
+ Changes: &ch.records, LinesAdded: &ch.added, LinesRemoved: &ch.removed,
+ LLMCalls: ptr(countKind(res, model.KindLLMCall)), Subagents: ptr(countChildStreams(res)),
+ BashRuns: ptr(countTool(res, "Bash")),
+ }
+}
+
+// encodeRound writes a round of the header and the first n frames of the
+// delta: its nodes, then its relations, then its unresolved entries.
+func encodeRound(h sessionflow.Header, d *delta, n int) ([]byte, string, error) {
+ w, err := sessionflow.NewWriter(h)
+ if err != nil {
+ return nil, "", err
+ }
+ for i := 0; i < n; i++ {
+ switch {
+ case i < len(d.nodes):
+ err = w.Node(d.nodes[i])
+ case i < len(d.nodes)+len(d.relations):
+ err = w.Relation(d.relations[i-len(d.nodes)])
+ default:
+ err = w.Unresolved(d.unresolved[i-len(d.nodes)-len(d.relations)])
+ }
+ if err != nil {
+ return nil, "", err
+ }
+ }
+ return w.Close()
+}
+
+// fitFrames finds how many of the delta's frames fit the budget, at least one,
+// and encodes them. A round grows with each frame, so the count is searched.
+func fitFrames(h sessionflow.Header, d *delta, budget int64) (int, []byte, string, error) {
+ lo, hi := 1, d.frames()
+ for lo < hi {
+ mid := (lo + hi + 1) / 2
+ data, _, err := encodeRound(h, d, mid)
+ if err != nil {
+ return 0, nil, "", err
+ }
+ if int64(len(data)) <= budget {
+ lo = mid
+ } else {
+ hi = mid - 1
+ }
+ }
+ data, digest, err := encodeRound(h, d, lo)
+ return lo, data, digest, err
+}
+
// noLandedFiles reports whether a session has no landed file at all.
func noLandedFiles(z *storage.Zone, session string) (bool, error) {
files, err := storage.LandedFiles(z, session)
@@ -486,22 +563,7 @@ func countChildStreams(res *assemble.Result) int {
// that started a child agent is classed as an agent call and keeps its name,
// so it counts too; otherwise the count would fall when a child's transcript
// lands.
-func countTool(res *assemble.Result, name string) int {
- n := 0
- for i := range res.Nodes {
- nd := &res.Nodes[i]
- if (nd.Kind != model.KindTool && nd.Kind != model.KindAgentCall) || len(nd.Attrs) == 0 {
- continue
- }
- var a struct {
- Name string `json:"name"`
- }
- if json.Unmarshal(nd.Attrs, &a) == nil && a.Name == name {
- n++
- }
- }
- return n
-}
+func countTool(res *assemble.Result, name string) int { return res.ToolCalls[name] }
func ptr(n int) *int { return &n }
@@ -592,13 +654,34 @@ type delta struct {
nodes []sessionflow.Node
relations []sessionflow.Relation
unresolved []sessionflow.Unresolved
- tombstones int
}
func (d *delta) empty() bool {
return len(d.nodes) == 0 && len(d.relations) == 0 && len(d.unresolved) == 0
}
+// frames is how many frames the delta holds.
+func (d *delta) frames() int { return len(d.nodes) + len(d.relations) + len(d.unresolved) }
+
+// written counts the nodes, relations, unresolved entries and tombstones among
+// the first n frames.
+func (d *delta) written(n int) (nodes, relations, unresolved, tombstones int) {
+ nodes = min(n, len(d.nodes))
+ relations = min(n-nodes, len(d.relations))
+ unresolved = n - nodes - relations
+ for _, x := range d.nodes[:nodes] {
+ if x.Tombstone {
+ tombstones++
+ }
+ }
+ for _, x := range d.relations[:relations] {
+ if x.Tombstone {
+ tombstones++
+ }
+ }
+ return nodes, relations, unresolved, tombstones
+}
+
// diff compares a fresh assembly against what the chain already says.
//
// Only what changed is written. An entity the assembly produced unchanged is
@@ -618,11 +701,9 @@ func diff(view *sessionflow.View, res *assemble.Result) *delta {
}
d.nodes = append(d.nodes, n)
}
- for id, prev := range view.Nodes {
+ for id := range view.Nodes {
if !seenNode[id] {
d.nodes = append(d.nodes, sessionflow.Node{Entity: sessionflow.Entity{ID: id, Tombstone: true}})
- d.tombstones++
- _ = prev
}
}
@@ -637,7 +718,6 @@ func diff(view *sessionflow.View, res *assemble.Result) *delta {
for id := range view.Relations {
if !seenRel[id] {
d.relations = append(d.relations, sessionflow.Relation{Entity: sessionflow.Entity{ID: id, Tombstone: true}})
- d.tombstones++
}
}
diff --git a/internal/scenario/remove/remove.go b/internal/scenario/remove/remove.go
index 2d12a98..718fdb9 100644
--- a/internal/scenario/remove/remove.go
+++ b/internal/scenario/remove/remove.go
@@ -120,8 +120,8 @@ type Input struct {
Endpoint string
SendLogs bool
SendMetrics bool
- // Busy holds sessions another process held this pass, the pipeline's
- // sessions to parse again.
+ // Busy holds the pipeline's sessions to parse again: another process
+ // held them this pass, or their parse failed. None of them is removed.
Busy map[string]bool
}
diff --git a/internal/scenario/write_sd.go b/internal/scenario/write_sd.go
index a82bcd7..98a6c80 100644
--- a/internal/scenario/write_sd.go
+++ b/internal/scenario/write_sd.go
@@ -125,11 +125,13 @@ func writeSD(p *Plan, root string, now time.Time, maxDelta int64) ([]string, err
// a missing one is built from nothing.
func (w *sdWriter) reindex() error {
dir := w.z.IndexDir(w.p.Session)
- ix, ok, err := index.Load(dir, w.p.Session)
+ st, err := storage.LoadIndexState(w.z.IndexStatePath(w.p.Session), w.p.Session)
if err != nil {
return err
}
- st, err := storage.LoadIndexState(w.z.IndexStatePath(w.p.Session), w.p.Session)
+ // Only an index that agrees with its saved state is extended, as the
+ // adapters do; see index.LoadFor.
+ ix, ok, err := index.LoadFor(dir, w.p.Session, st)
if err != nil {
return err
}
diff --git a/internal/view/api.go b/internal/view/api.go
index 7161c87..8fd1e53 100644
--- a/internal/view/api.go
+++ b/internal/view/api.go
@@ -484,11 +484,7 @@ func segmentRows(c *Conversation, talks []talkRow) []sessionview.Segment {
out := []sessionview.Segment{}
for _, n := range sessionflow.InOrder(segs) {
s := span[n.ID]
- out = append(out, sessionview.Segment{
- ID: n.ID, State: attrString(n, "state"),
- Talks: count[n.ID], From: s[0], To: s[1],
- Committable: attrBool(n, "committable"),
- })
+ out = append(out, sessionview.Segment{ID: n.ID, Talks: count[n.ID], From: s[0], To: s[1]})
}
return out
}
@@ -572,18 +568,6 @@ func streamRows(c *Conversation, talks []talkRow) []sessionview.Stream {
return out
}
-func attrBool(n *sessionflow.Node, key string) bool {
- if len(n.Attrs) == 0 {
- return false
- }
- var m map[string]any
- if json.Unmarshal(n.Attrs, &m) != nil {
- return false
- }
- b, _ := m[key].(bool)
- return b
-}
-
// labelRefs finds where a talk's name should be read from: the first
// external input under it. A talk with no external input at all - work
// delegated to a pool agent arrives outside its own records - is named by
diff --git a/pkg/model/model.go b/pkg/model/model.go
index 828c416..0042cb6 100644
--- a/pkg/model/model.go
+++ b/pkg/model/model.go
@@ -25,8 +25,8 @@ package model
// Structural node kinds - the hierarchy above the leaves.
//
-// Segment sits above Session and cuts ACROSS every stream beneath it: it is a
-// time window chosen for commit, not a branch of the stream tree.
+// Segment sits above Session and cuts ACROSS every stream beneath it: it is an
+// activity window, not a branch of the stream tree.
const (
KindConversation = "conversation"
KindSegment = "segment"
@@ -123,7 +123,7 @@ const (
RelFollows = "follows"
// RelSummarizes: a summary -> the reset that produced it.
RelSummarizes = "summarizes"
- // RelInSegment places a Talk in the activity window that will commit it.
+ // RelInSegment places a Talk in the activity window it falls in.
//
// This is a relation and not containment on purpose. A Segment is a time
// window and a Session outlives many of them, so a Session cannot sit under
diff --git a/pkg/providerbody/providerbody.go b/pkg/providerbody/providerbody.go
index 05f7e20..8b93062 100644
--- a/pkg/providerbody/providerbody.go
+++ b/pkg/providerbody/providerbody.go
@@ -85,15 +85,13 @@ type Manifest struct {
// moving cache marker taken out. It is storage, not structure: it picks
// the body a copy comes from, and says nothing a join may rely on.
Chain string `json:"chain,omitempty"`
- // Why says the body is kept whole, in part 0, and why it was not cut.
- Why string `json:"why,omitempty"`
// The values the body carries that join it, as its adapter read them,
// so a reader can join a body without rebuilding it. Empty when the
- // body does not carry one.
+ // body does not carry one. A request's run is the record's own, not
+ // repeated here.
Model string `json:"model,omitempty"`
Session string `json:"session,omitempty"`
- Run string `json:"run,omitempty"`
Call string `json:"call,omitempty"`
Request string `json:"request,omitempty"`
PreviousRequest string `json:"previous_request,omitempty"`
diff --git a/pkg/providerbody/providerbody_test.go b/pkg/providerbody/providerbody_test.go
index 51c3a2b..061be55 100644
--- a/pkg/providerbody/providerbody_test.go
+++ b/pkg/providerbody/providerbody_test.go
@@ -303,9 +303,17 @@ func TestABodyThatCannotBeCutLandsWholeAndSaysWhy(t *testing.T) {
// Whitespace between tokens is no problem: literals keep it.
continue
}
- if m.Why == "" || rec.Parts[0].Kind != sessiondata.PartUnknown {
+ if rec.Parts[0].Kind != sessiondata.PartUnknown || rec.Parts[0].Text == "" {
t.Fatalf("%s: landed cut, want whole with a reason", name)
}
+ if len(m.Segments) != 1 || m.Segments[0].Part == nil || *m.Segments[0].Part != 0 {
+ t.Fatalf("%s: the manifest does not name the whole part: %+v", name, m.Segments)
+ }
+ // The reason is the whole part's text, and the manifest does not
+ // repeat it.
+ if last := rec.Parts[len(rec.Parts)-1]; bytes.Contains(last.Data, []byte(`"why"`)) {
+ t.Fatalf("%s: the manifest repeats the reason: %s", name, last.Data)
+ }
}
}
@@ -499,3 +507,28 @@ func TestARebuildStopsAtItsClaim(t *testing.T) {
t.Fatalf("RecordBytes is %d", n)
}
}
+
+// A request's run is its record's own. The manifest carries the keys a reader
+// joins a body by, and does not repeat the run.
+func TestARequestKeepsItsRunOnTheRecord(t *testing.T) {
+ s := NewSession()
+ rec, err := s.Encode(Body{ID: "00000000-0000-4000-8000-000000000001.request", Role: RoleRequest,
+ Src: "00000000-0000-4000-8000-000000000001.request.json", Bytes: []byte(`{"model":"m","messages":[]}`),
+ Keys: Keys{Run: "r1", Session: "s1"}})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if rec.Run != "r1" {
+ t.Fatalf("the record's run is %q, want r1", rec.Run)
+ }
+ m, err := ManifestOf(rec)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if m.Session != "s1" {
+ t.Fatalf("the manifest's session is %q, want s1", m.Session)
+ }
+ if last := rec.Parts[len(rec.Parts)-1]; bytes.Contains(last.Data, []byte(`"run"`)) {
+ t.Fatalf("the manifest repeats the run: %s", last.Data)
+ }
+}
diff --git a/pkg/providerbody/session.go b/pkg/providerbody/session.go
index 499eba0..531e3e9 100644
--- a/pkg/providerbody/session.go
+++ b/pkg/providerbody/session.go
@@ -311,8 +311,8 @@ func partBytes(p sessiondata.Part) ([]byte, error) {
// Encode cuts a body against what the session holds, checks that the record
// rebuilds the body after a round trip through the Session Data writer and
// reader, and adds it. A body that cannot be cut, or does not come back whole,
-// lands whole in one unknown part, and the manifest says why. A body the
-// session already holds returns ErrRepeat.
+// lands whole in one unknown part, whose text says why. A body the session
+// already holds returns ErrRepeat.
func (s *Session) Encode(b Body) (*sessiondata.Record, error) {
body := b.Bytes
sum := digest(body)
@@ -325,7 +325,7 @@ func (s *Session) Encode(b Body) (*sessiondata.Record, error) {
k := b.Keys
m := &Manifest{
Schema: Schema, Role: b.Role, Src: b.Src, SHA256: sum, Bytes: len(body),
- Model: k.Model, Session: k.Session, Run: k.Run, Call: k.Call, Request: k.Request, PreviousRequest: k.PreviousRequest,
+ Model: k.Model, Session: k.Session, Call: k.Call, Request: k.Request, PreviousRequest: k.PreviousRequest,
}
rec := &sessiondata.Record{Ord: 1, Sha: sum[:12], Bytes: len(body), ID: b.ID, Model: k.Model}
switch b.Role {
@@ -349,7 +349,9 @@ func (s *Session) Encode(b Body) (*sessiondata.Record, error) {
whole := sessiondata.Part{Kind: sessiondata.PartUnknown, Text: why, State: "available", Bytes: len(body)}
whole.SetRaw(body)
zero := 0
- m.Depth, m.Why, m.Segments = 0, why, []Segment{{Part: &zero}}
+ // Why the body was not cut is the whole part's text; the manifest
+ // does not repeat it.
+ m.Depth, m.Segments = 0, []Segment{{Part: &zero}}
rec.Parts = []sessiondata.Part{whole, manifestPart(m)}
if err := s.roundTrip(rec, m); err != nil {
return nil, err
diff --git a/pkg/sessionview/sessionview.go b/pkg/sessionview/sessionview.go
index 8f9d589..49f2249 100644
--- a/pkg/sessionview/sessionview.go
+++ b/pkg/sessionview/sessionview.go
@@ -244,12 +244,10 @@ type Origin struct {
// Segment is one activity window, with the span of the talks placed in it.
type Segment struct {
- ID string `json:"id"`
- State string `json:"state"`
- Committable bool `json:"committable"`
- Talks int `json:"talks"`
- From int64 `json:"from"`
- To int64 `json:"to"`
+ ID string `json:"id"`
+ Talks int `json:"talks"`
+ From int64 `json:"from"`
+ To int64 `json:"to"`
}
// Node is one entity of the fold with what its record says, and its
diff --git a/tests/chain/round_upgrade_test.go b/tests/chain/round_upgrade_test.go
new file mode 100644
index 0000000..ee16e49
--- /dev/null
+++ b/tests/chain/round_upgrade_test.go
@@ -0,0 +1,283 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package chain
+
+import (
+ "bytes"
+ "encoding/json"
+ "io/fs"
+ "os"
+ "path/filepath"
+ "reflect"
+ "strings"
+ "testing"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/index"
+ "github.com/apache/skywalking-ai-sessionizer/internal/parse"
+ "github.com/apache/skywalking-ai-sessionizer/internal/storage"
+ "github.com/apache/skywalking-ai-sessionizer/internal/verify"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessionflow"
+)
+
+const upgradeSession = "00000001-0000-4000-8000-000000000001"
+
+// A chain the previous parser wrote continues under this one. Its next round
+// revises the nodes whose attributes changed, covers no new landed file, and
+// leaves the conversation as a fresh parse of the same files builds it. A
+// second parse finds nothing to write.
+func TestAnOlderChainContinues(t *testing.T) {
+ old := upgradeRoot(t, true)
+ r := upgradeParse(t, old, 0)
+ if r.Number != 2 || r.FromSeq != r.ThroughSeq+1 || r.More {
+ t.Fatalf("the continuing round is round %d over sequences %d..%d, more=%v; want round 2 over no new file",
+ r.Number, r.FromSeq, r.ThroughSeq, r.More)
+ }
+ if r.Relations != 0 || r.Unresolved != 0 || r.Tombstones != 0 {
+ t.Fatalf("the continuing round writes %d relations, %d unresolved entries and %d tombstones; want node revisions only",
+ r.Relations, r.Unresolved, r.Tombstones)
+ }
+ if again := upgradeParse(t, old, 0); again.Changed() {
+ t.Fatalf("a second parse wrote round %d", again.Number)
+ }
+ continued := sameAsFresh(t, old, upgradeRoot(t, false))
+ for id, n := range continued.Nodes {
+ for _, key := range []string{`"state"`, `"committable"`, `"gates_unmet"`, `"title_from"`, `"loops"`,
+ `"trigger"`, `"fragments"`} {
+ if bytes.Contains(n.Attrs, []byte(key)) {
+ t.Errorf("node %s still carries %s: %s", id, key, n.Attrs)
+ }
+ }
+ }
+}
+
+// upgradeRoot copies the fixture into a new root, with the old round or
+// without it.
+func upgradeRoot(t *testing.T, withRound bool) *storage.Zone {
+ t.Helper()
+ src, dst := filepath.Join("testdata", "round-upgrade"), t.TempDir()
+ err := filepath.WalkDir(src, func(path string, d fs.DirEntry, err error) error {
+ if err != nil || d.IsDir() || filepath.Ext(path) == ".md" {
+ return err
+ }
+ rel, _ := filepath.Rel(src, path)
+ if !withRound && strings.HasPrefix(rel, "_conversations") {
+ return nil
+ }
+ data, err := os.ReadFile(path)
+ if err != nil {
+ return err
+ }
+ if err := os.MkdirAll(filepath.Join(dst, filepath.Dir(rel)), 0o755); err != nil {
+ return err
+ }
+ return os.WriteFile(filepath.Join(dst, rel), data, 0o444)
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ return storage.NewZone(dst)
+}
+
+// A large conversation revises many nodes on an upgrade, and a round over no
+// new file cannot be narrowed. It is cut by its frames: each round of the
+// revision fits the budget, the rounds together leave the conversation as a
+// fresh parse builds it, and a parse after them finds nothing to write.
+func TestAnUpgradeIsCutToTheBudget(t *testing.T) {
+ old := upgradeRoot(t, true)
+ rounds := parseAll(t, old, upgradeBudget)
+ if len(rounds) < 2 {
+ t.Fatalf("the revision took %d round, want it cut into several", len(rounds))
+ }
+ for _, r := range rounds {
+ if r.FromSeq != 5 || r.ThroughSeq != 4 {
+ t.Errorf("round %d covers sequences %d..%d, want no new file", r.Number, r.FromSeq, r.ThroughSeq)
+ }
+ if fi, err := os.Stat(r.Path); err != nil || (fi.Size() > upgradeBudget && r.Nodes+r.Relations+r.Unresolved > 1) {
+ t.Errorf("round %d is over the budget: %v", r.Number, err)
+ }
+ }
+ if again := upgradeParse(t, old, upgradeBudget); again.Changed() {
+ t.Fatalf("a parse after the revision wrote round %d", again.Number)
+ }
+ sameAsFresh(t, old, upgradeRoot(t, false))
+ chainIntact(t, old)
+}
+
+// After an upgrade, the round that lands new evidence would carry every
+// revision as well, and one landed file cannot be narrowed. The revisions go
+// first, in rounds of their own over no new file, and the new evidence follows
+// in a round of its own.
+func TestAnUpgradeGoesBeforeNewEvidence(t *testing.T) {
+ old, fresh := upgradeRoot(t, true), upgradeRoot(t, false)
+ for _, z := range []*storage.Zone{old, fresh} {
+ landNextInput(t, z)
+ }
+ rounds := parseAll(t, old, upgradeBudget)
+ last := rounds[len(rounds)-1]
+ if len(rounds) < 2 || last.FromSeq != 5 || last.ThroughSeq != 5 {
+ t.Fatalf("%d rounds, the last over sequences %d..%d; want revisions, then sequence 5",
+ len(rounds), last.FromSeq, last.ThroughSeq)
+ }
+ for _, r := range rounds[:len(rounds)-1] {
+ if r.FromSeq != 5 || r.ThroughSeq != 4 {
+ t.Errorf("round %d covers sequences %d..%d before the new evidence, want no new file",
+ r.Number, r.FromSeq, r.ThroughSeq)
+ }
+ }
+ sameAsFresh(t, old, fresh)
+ chainIntact(t, old)
+}
+
+// upgradeBudget is small enough that the fixture's revision does not fit one
+// round.
+const upgradeBudget = 4096
+
+func upgradeParse(t *testing.T, z *storage.Zone, budget int64) *parse.Round {
+ t.Helper()
+ r, err := parse.Session(z, parse.Options{Conversation: upgradeSession, Session: upgradeSession,
+ Reindex: index.Rebuild, MaxRoundBytes: budget})
+ if err != nil {
+ t.Fatal(err)
+ }
+ return r
+}
+
+// parseAll parses until the chain reaches the index, and returns the rounds
+// written.
+func parseAll(t *testing.T, z *storage.Zone, budget int64) []*parse.Round {
+ t.Helper()
+ var rounds []*parse.Round
+ for {
+ r := upgradeParse(t, z, budget)
+ if r.Changed() {
+ rounds = append(rounds, r)
+ }
+ if !r.Changed() || !r.More {
+ return rounds
+ }
+ if len(rounds) > 1000 {
+ t.Fatal("the chain does not reach the index")
+ }
+ }
+}
+
+// sameAsFresh parses fresh whole and checks that the continued chain folds to
+// the same conversation: every node without its revision, every relation and
+// every unresolved entry, and the header's Bash count.
+func sameAsFresh(t *testing.T, continuedRoot, fresh *storage.Zone) *sessionflow.View {
+ t.Helper()
+ parseAll(t, fresh, 0)
+ continued, err := parse.View(continuedRoot.Root(), upgradeSession)
+ if err != nil {
+ t.Fatal(err)
+ }
+ built, err := parse.View(fresh.Root(), upgradeSession)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if continued.ThroughSeq != built.ThroughSeq {
+ t.Fatalf("the continued chain reaches sequence %d, a fresh parse %d", continued.ThroughSeq, built.ThroughSeq)
+ }
+ for _, c := range []struct {
+ what string
+ got, want map[string]string
+ }{
+ {"node", withoutRevision(continued.Nodes), withoutRevision(built.Nodes)},
+ {"relation", withoutRevision(continued.Relations), withoutRevision(built.Relations)},
+ {"unresolved entry", withoutRevision(continued.Unresolved), withoutRevision(built.Unresolved)},
+ } {
+ if reflect.DeepEqual(c.got, c.want) {
+ continue
+ }
+ for id := range c.want {
+ if c.got[id] != c.want[id] {
+ t.Errorf("%s %s: continued %s, fresh %s", c.what, id, c.got[id], c.want[id])
+ }
+ }
+ t.Fatalf("the continued chain folds to %d %ss, a fresh parse to %d", len(c.got), c.what, len(c.want))
+ }
+ if continued.BashRuns == nil || built.BashRuns == nil || *continued.BashRuns != 1 || *built.BashRuns != 1 {
+ t.Fatalf("bash_runs: continued %v, fresh %v, want 1 each", continued.BashRuns, built.BashRuns)
+ }
+ return continued
+}
+
+// chainIntact checks the continued chain the way asz verify does.
+func chainIntact(t *testing.T, z *storage.Zone) {
+ t.Helper()
+ rep, err := verify.Chain(z, upgradeSession, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if rep.Problems != 0 {
+ for _, r := range rep.Rounds {
+ for _, p := range r.Problems {
+ t.Error(p)
+ }
+ }
+ t.Fatalf("the continued chain has %d problems", rep.Problems)
+ }
+}
+
+// landNextInput lands one more file in main: a person's next prompt, after the
+// last record the fixture holds.
+func landNextInput(t *testing.T, z *storage.Zone) {
+ t.Helper()
+ dir := z.StreamDir(upgradeSession, "main")
+ path := filepath.Join(dir, "transcript-20261012T013740.000000000Z-000005.sd")
+ f, err := os.Create(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer f.Close()
+ w, err := sessiondata.NewWriter(f, &sessiondata.Header{
+ Seq: 5, At: "2026-10-12T01:37:40Z", Kind: sessiondata.KindTranscript,
+ Adapter: "mock/0.2.0", Dialect: "mock/1", Session: upgradeSession, Stream: "main",
+ Src: "-Users-dev-01-full-conversation/00000001-0000-4000-8000-000000000001/streams/main",
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := w.Write(&sessiondata.Record{
+ Ord: 17, Off: 5374, Sha: "0123456789ab", Bytes: 120, ID: "s9-prompt", Parent: "s8-summary",
+ Run: "s9-cycle", From: sessiondata.FromExternal, Time: "2026-10-12T01:37:40.000Z",
+ Flags: []string{"external_input"},
+ Parts: []sessiondata.Part{{Kind: sessiondata.PartText, Text: "Now run the tests.", State: "available", Bytes: 18}},
+ }); err != nil {
+ t.Fatal(err)
+ }
+ if err := w.Close(); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// withoutRevision renders each entity without its revision, which counts the
+// rounds that wrote it rather than saying what it is.
+func withoutRevision[E any](entities map[string]*E) map[string]string {
+ out := make(map[string]string, len(entities))
+ for id, e := range entities {
+ var m map[string]any
+ b, _ := json.Marshal(e)
+ _ = json.Unmarshal(b, &m)
+ delete(m, "revision")
+ b, _ = json.Marshal(m)
+ out[id] = string(b)
+ }
+ return out
+}
diff --git a/tests/chain/testdata/round-upgrade/00000001-0000-4000-8000-000000000001/streams/a0a10ef0666c4dc7e/meta-20261012T013727.579645000Z-000004.sd b/tests/chain/testdata/round-upgrade/00000001-0000-4000-8000-000000000001/streams/a0a10ef0666c4dc7e/meta-20261012T013727.579645000Z-000004.sd
new file mode 100644
index 0000000..e2e6300
--- /dev/null
+++ b/tests/chain/testdata/round-upgrade/00000001-0000-4000-8000-000000000001/streams/a0a10ef0666c4dc7e/meta-20261012T013727.579645000Z-000004.sd
@@ -0,0 +1,3 @@
+{"h":1,"schema":"sd/1","seq":4,"at":"2026-10-12T01:37:27.579645Z","kind":"agent_meta","adapter":"mock/0.2.0","dialect":"mock/1","src":"-Users-dev-01-full-conversation/00000001-0000-4000-8000-000000000001/streams/a0a10ef0666c4dc7e.meta","session":"00000001-0000-4000-8000-000000000001","stream":"a0a10ef0666c4dc7e"}
+{"ord":1,"off":0,"sha":"c0b42d2ebfe2","bytes":253,"child":"a0a10ef0666c4dc7e","label":"checker","from":"runtime","parts":[{"k":"data","data":{"agentType":"general-purpose","description":"checker","spawnDepth":1,"toolUseId":"s5-tool"},"state":"available","bytes":92}]}
+{"t":"end","records":1,"digest":"f6ff831892da91611bf8594d23b1cf16f2ef6de518c6b3c119db8e41d3923074"}
diff --git a/tests/chain/testdata/round-upgrade/00000001-0000-4000-8000-000000000001/streams/a0a10ef0666c4dc7e/transcript-20261012T013727.579645000Z-000003.sd b/tests/chain/testdata/round-upgrade/00000001-0000-4000-8000-000000000001/streams/a0a10ef0666c4dc7e/transcript-20261012T013727.579645000Z-000003.sd
new file mode 100644
index 0000000..bcd0f0c
--- /dev/null
+++ b/tests/chain/testdata/round-upgrade/00000001-0000-4000-8000-000000000001/streams/a0a10ef0666c4dc7e/transcript-20261012T013727.579645000Z-000003.sd
@@ -0,0 +1,4 @@
+{"h":1,"schema":"sd/1","seq":3,"at":"2026-10-12T01:37:27.579645Z","kind":"transcript","adapter":"mock/0.2.0","dialect":"mock/1","src":"-Users-dev-01-full-conversation/00000001-0000-4000-8000-000000000001/streams/a0a10ef0666c4dc7e","session":"00000001-0000-4000-8000-000000000001","stream":"a0a10ef0666c4dc7e"}
+{"ord":1,"off":0,"sha":"e6be11282954","bytes":231,"id":"a0a10ef0666c4dc7e-prompt","run":"a0a10ef0666c4dc7e-cycle","from":"external","time":"2026-10-12T01:37:33.679Z","parts":[{"k":"text","text":"check the tests","state":"available","bytes":15}]}
+{"ord":2,"off":232,"sha":"4c499e0a27b8","bytes":355,"id":"checker-s1-call-f1","parent":"a0a10ef0666c4dc7e-prompt","call":"checker-s1-call","from":"agent","time":"2026-10-12T01:37:34.679Z","flags":["finished"],"usage":{"in":2,"out":42,"cache_read":900,"cache_write":100},"model":"claude-opus-5","parts":[{"k":"text","text":"Tests pass.","state":"available","bytes":11}]}
+{"t":"end","records":2,"digest":"77162f41d2799100adf4a66709a8b8339a1033fce933a2bcabf845a9e5f3f1d6"}
diff --git a/tests/chain/testdata/round-upgrade/00000001-0000-4000-8000-000000000001/streams/main/changes-20261012T013727.579645000Z-000002.sd b/tests/chain/testdata/round-upgrade/00000001-0000-4000-8000-000000000001/streams/main/changes-20261012T013727.579645000Z-000002.sd
new file mode 100644
index 0000000..e5d9615
--- /dev/null
+++ b/tests/chain/testdata/round-upgrade/00000001-0000-4000-8000-000000000001/streams/main/changes-20261012T013727.579645000Z-000002.sd
@@ -0,0 +1,3 @@
+{"h":1,"schema":"sd/1","seq":2,"at":"2026-10-12T01:37:27.579645Z","kind":"changes","adapter":"mock/0.2.0","dialect":"mock/1","src":"-Users-dev-01-full-conversation/00000001-0000-4000-8000-000000000001/streams/main.changes","session":"00000001-0000-4000-8000-000000000001","stream":"main"}
+{"ord":1,"off":0,"sha":"a1888287cad1","bytes":1361,"id":"tool-run-make-build","tool":"tool-run-make-build","time":"2026-10-12T01:37:29.579Z","parts":[{"k":"data","data":{"schema":"changes/1","id":"tool-run-make-build","captured_by":"asz-plugin","session":"00000001-0000-4000-8000-000000000001","stream":"main","tool":"tool-run-make-build","tool_name":"Bash","time":"2026-10-12T01:37:29.579Z","basis":"tool_window","root":{"path":"/Users/dev/01-full-conversation"},"policy":{"exclusions":"stan [...]
+{"t":"end","records":1,"digest":"a994193aeaaff2d84bc2dc8e105bb24e0daa8bd3cc34fce916f53270a7069bdf"}
diff --git a/tests/chain/testdata/round-upgrade/00000001-0000-4000-8000-000000000001/streams/main/transcript-20261012T013727.579645000Z-000001.sd b/tests/chain/testdata/round-upgrade/00000001-0000-4000-8000-000000000001/streams/main/transcript-20261012T013727.579645000Z-000001.sd
new file mode 100644
index 0000000..85a3de3
--- /dev/null
+++ b/tests/chain/testdata/round-upgrade/00000001-0000-4000-8000-000000000001/streams/main/transcript-20261012T013727.579645000Z-000001.sd
@@ -0,0 +1,18 @@
+{"h":1,"schema":"sd/1","seq":1,"at":"2026-10-12T01:37:27.579645Z","kind":"transcript","adapter":"mock/0.2.0","dialect":"mock/1","src":"-Users-dev-01-full-conversation/00000001-0000-4000-8000-000000000001/streams/main","session":"00000001-0000-4000-8000-000000000001","stream":"main"}
+{"ord":1,"off":0,"sha":"b1ad2a1f1ea1","bytes":188,"label":"build and check","from":"runtime","parts":[{"k":"data","data":{"aiTitle":"build and check","type":"ai-title"},"state":"available","bytes":47}]}
+{"ord":2,"off":189,"sha":"a9fff68ee83a","bytes":248,"id":"s1-input","run":"s1-cycle","from":"external","time":"2026-10-12T01:37:27.579Z","trigger":"external","flags":["external_input"],"parts":[{"k":"text","text":"run the build","state":"available","bytes":13}]}
+{"ord":3,"off":438,"sha":"9d697041cd11","bytes":219,"id":"s2-inject","parent":"s1-input","from":"runtime","time":"2026-10-12T01:37:27.679Z","flags":["injected"],"parts":[{"k":"text","text":"skills: 1","state":"available","bytes":9}]}
+{"ord":4,"off":658,"sha":"2c97773b48b2","bytes":387,"id":"s3-call-f1","parent":"s2-inject","call":"s3-call","from":"agent","time":"2026-10-12T01:37:28.579Z","usage":{"in":2,"out":50,"cache_read":900,"cache_write":100},"model":"claude-opus-5","parts":[{"k":"reasoning","state":"unavailable"}],"dropped":[{"what":"reasoning signature","bytes":3,"why":"a provider verifies it; a reader cannot read it"}]}
+{"ord":5,"off":1046,"sha":"34282b87ea41","bytes":307,"id":"s3-call-f2","parent":"s3-call-f1","call":"s3-call","from":"agent","time":"2026-10-12T01:37:28.679Z","usage":{"in":2,"out":50,"cache_read":900,"cache_write":100},"model":"claude-opus-5","parts":[{"k":"text","text":"Building now.","state":"available","bytes":13}]}
+{"ord":6,"off":1354,"sha":"52b1ed35e2a1","bytes":434,"id":"s3-call-f3","parent":"s3-call-f2","call":"s3-call","from":"agent","time":"2026-10-12T01:37:28.779Z","flags":["finished"],"usage":{"in":2,"out":50,"cache_read":900,"cache_write":100},"model":"claude-opus-5","parts":[{"k":"call","data":{"command":"go generate ./... && make build","description":"generate and build"},"id":"tool-run-make-build","name":"Bash","state":"available","bytes":80}]}
+{"ord":7,"off":1789,"sha":"3163a1d35f7c","bytes":303,"id":"s3-result","parent":"s3-call-f3","run":"s1-cycle","from":"external","time":"2026-10-12T01:37:29.579Z","parts":[{"k":"result","text":"build succeeded","data":{"stderr":"","stdout":"build succeeded"},"of":"tool-run-make-build","state":"available","bytes":15}]}
+{"ord":8,"off":2093,"sha":"7251b96a38ff","bytes":382,"id":"s4-call-f1","parent":"s3-result","call":"s4-call","from":"agent","time":"2026-10-12T01:37:30.579Z","flags":["finished"],"usage":{"in":2,"out":10,"cache_read":900,"cache_write":100},"model":"claude-opus-5","parts":[{"k":"call","data":{"query":"go build cache"},"id":"srvtool-websearch","name":"WebSearch","state":"available","bytes":26}]}
+{"ord":9,"off":2476,"sha":"d81a47729215","bytes":299,"id":"s4-result","parent":"s4-call-f1","run":"s1-cycle","from":"external","time":"2026-10-12T01:37:31.579Z","parts":[{"k":"result","text":"search results","data":{"stderr":"","stdout":"search results"},"of":"srvtool-websearch","state":"available","bytes":14}]}
+{"ord":10,"off":2776,"sha":"98410dad212c","bytes":395,"id":"s5-call-f1","parent":"s4-result","call":"s5-call","from":"agent","time":"2026-10-12T01:37:32.579Z","flags":["finished"],"usage":{"in":2,"out":20,"cache_read":900,"cache_write":100},"model":"claude-opus-5","parts":[{"k":"call","data":{"description":"checker","prompt":"check the tests"},"id":"s5-tool","name":"Agent","state":"available","bytes":52}]}
+{"ord":11,"off":3172,"sha":"a2d082f32181","bytes":415,"id":"s5-ack","parent":"s5-call-f1","run":"s1-cycle","child":"a0a10ef0666c4dc7e","from":"external","time":"2026-10-12T01:37:32.679Z","flags":["launch_ack"],"parts":[{"k":"result","text":"launched","data":{"agentId":"a0a10ef0666c4dc7e","description":"checker","isAsync":true,"prompt":"check the tests","status":"async_launched"},"of":"s5-tool","state":"available","bytes":8}]}
+{"ord":12,"off":3588,"sha":"b9db937e6c4b","bytes":435,"id":"s5-notice","parent":"s5-ack","run":"s5-cycle-notification","tool":"s5-tool","child":"a0a10ef0666c4dc7e","from":"external","time":"2026-10-12T01:37:35.679Z","trigger":"notification","parts":[{"k":"text","text":"<task-notification>\n<task-id>a0a10ef0666c4dc7e</task-id>\n<tool-use-id>s5-tool</tool-use-id>\n<status>completed</status>\n</task-notification>","state":"available","bytes":139}]}
+{"ord":13,"off":4024,"sha":"eaba42f8d3e1","bytes":348,"id":"s6-call-f1","parent":"s5-notice","call":"s6-call","from":"agent","time":"2026-10-12T01:37:36.679Z","flags":["finished"],"usage":{"in":2,"out":30,"cache_read":900,"cache_write":100},"model":"claude-opus-5","parts":[{"k":"text","text":"Build passed and tests are green.","state":"available","bytes":33}]}
+{"ord":14,"off":4373,"sha":"50fd26f1584f","bytes":317,"id":"s7-synthetic","parent":"s6-call-f1","call":"s7-synthetic-call","from":"agent","time":"2026-10-12T01:37:37.679Z","flags":["synthetic"],"usage":{},"model":"<synthetic>","parts":[{"k":"text","text":"API Error: Connection lost mid-response.","state":"available","bytes":40}]}
+{"ord":15,"off":4691,"sha":"193a5ef07ae3","bytes":394,"id":"s8-boundary","continues":"s7-synthetic","from":"runtime","time":"2026-10-12T01:37:38.679Z","flags":["context_reset"],"parts":[{"k":"data","data":{"compactMetadata":{"preservedMessages":{"allUuids":["s7-synthetic"]},"trigger":"auto"},"logicalParentUuid":"s7-synthetic","subtype":"compact_boundary","type":"system"},"state":"available","bytes":168}]}
+{"ord":16,"off":5086,"sha":"e87d62eeafcd","bytes":287,"id":"s8-summary","parent":"s8-boundary","run":"s8-cycle-compact","from":"external","time":"2026-10-12T01:37:38.279Z","flags":["reset_summary"],"parts":[{"k":"text","text":"Summary: the build was run and checked.","state":"available","bytes":39}]}
+{"t":"end","records":16,"digest":"b52abfd565cd480c50004b8eee5d39bc30a9c17a357276903843c29330c27d5a"}
diff --git a/tests/chain/testdata/round-upgrade/README.md b/tests/chain/testdata/round-upgrade/README.md
new file mode 100644
index 0000000..a2b5638
--- /dev/null
+++ b/tests/chain/testdata/round-upgrade/README.md
@@ -0,0 +1,12 @@
+# A chain written by the previous parser
+
+This root was written by asz at commit `6fbf0ae`: `asz scenario build tests/scenarios/fixture.yaml
+--format sd`, then `asz parse`. It holds the four landed files of one session and the round that
+parse wrote, `r000001-4388f7a5d811.sf`. Nothing else is kept: the index is rebuilt from the landed
+files, and the chain head is found by listing the rounds.
+
+That round still carries the node attributes later builds stopped writing: a segment's `state`,
+`gates_unmet` and `committable`, the session's `conversation` and `title_from`, an epoch's
+`records`, a talk's `loops`, `runs` and `trigger`, a run's `trigger`, a call's `fragments` and a
+tool's `name`. Keep these files unchanged. `round_upgrade_test.go` continues the chain with the
+current parser and checks that the conversation comes out as a fresh parse would build it.
diff --git a/tests/chain/testdata/round-upgrade/_conversations/00000001-0000-4000-8000-000000000001/rounds/r000001-4388f7a5d811.sf b/tests/chain/testdata/round-upgrade/_conversations/00000001-0000-4000-8000-000000000001/rounds/r000001-4388f7a5d811.sf
new file mode 100644
index 0000000..f0b8511
--- /dev/null
+++ b/tests/chain/testdata/round-upgrade/_conversations/00000001-0000-4000-8000-000000000001/rounds/r000001-4388f7a5d811.sf
@@ -0,0 +1,45 @@
+{"t":"header","schema":"sf/1","conversation":"00000001-0000-4000-8000-000000000001","session":"00000001-0000-4000-8000-000000000001","round":1,"from_seq":1,"through_seq":4,"input_digest":"9c9a24da744863c6380ad5ac88a7c407446e0cff320292633794ec11a4ca71f4","parser":"v1","policy":"v1+idle=10m0s","from_time":"2026-10-12T01:37:27.579Z","through_time":"2026-10-12T01:37:38.679Z","session_from_time":"2026-10-12T01:37:27.579Z","session_through_time":"2026-10-12T01:37:38.679Z","title":"build and ch [...]
+{"t":"node","id":"ack/1/11","revision":1,"kind":"agent.launch_ack","parent":"run/talk_main_s1-cycle/s1-cycle","stream":"main","ref":{"seq":1,"row":11}}
+{"t":"node","id":"boundary/1/15","revision":1,"kind":"epoch.boundary","parent":"epoch/main/s8-boundary","stream":"main","ref":{"seq":1,"row":15}}
+{"t":"node","id":"call/checker-s1-call","revision":1,"kind":"llm.call","parent":"run/talk_a0a10ef0666c4dc7e/a0a10ef0666c4dc7e-cycle","stream":"a0a10ef0666c4dc7e","ref":{"seq":3,"row":2},"refs":[{"seq":3,"row":2}],"attrs":{"fragments":1,"usage":"observed_replayable","usage_at":{"row":2,"seq":3},"usage_from":"last_fragment_in_line_order"}}
+{"t":"node","id":"call/s3-call","revision":1,"kind":"llm.call","parent":"run/talk_main_s1-cycle/s1-cycle","stream":"main","ref":{"seq":1,"row":4},"refs":[{"seq":1,"row":4},{"seq":1,"row":5},{"seq":1,"row":6}],"attrs":{"fragments":3,"usage":"observed_replayable","usage_at":{"row":6,"seq":1},"usage_from":"last_fragment_in_line_order"}}
+{"t":"node","id":"call/s4-call","revision":1,"kind":"llm.call","parent":"run/talk_main_s1-cycle/s1-cycle","stream":"main","ref":{"seq":1,"row":8},"refs":[{"seq":1,"row":8}],"attrs":{"fragments":1,"usage":"observed_replayable","usage_at":{"row":8,"seq":1},"usage_from":"last_fragment_in_line_order"}}
+{"t":"node","id":"call/s5-call","revision":1,"kind":"llm.call","parent":"run/talk_main_s1-cycle/s1-cycle","stream":"main","ref":{"seq":1,"row":10},"refs":[{"seq":1,"row":10}],"attrs":{"fragments":1,"usage":"observed_replayable","usage_at":{"row":10,"seq":1},"usage_from":"last_fragment_in_line_order"}}
+{"t":"node","id":"call/s6-call","revision":1,"kind":"llm.call","parent":"run/talk_main_s1-cycle/s5-cycle-notification","stream":"main","ref":{"seq":1,"row":13},"refs":[{"seq":1,"row":13}],"attrs":{"fragments":1,"usage":"observed_replayable","usage_at":{"row":13,"seq":1},"usage_from":"last_fragment_in_line_order"}}
+{"t":"node","id":"call/s7-synthetic-call","revision":1,"kind":"llm.call","parent":"run/talk_main_s1-cycle/s5-cycle-notification","stream":"main","ref":{"seq":1,"row":14},"refs":[{"seq":1,"row":14}],"attrs":{"fragments":1,"stop_reason":"unavailable","usage":"unavailable","usage_from":"last_fragment_in_line_order"}}
+{"t":"node","id":"epoch/a0a10ef0666c4dc7e/0","revision":1,"kind":"epoch","parent":"stream/a0a10ef0666c4dc7e","stream":"a0a10ef0666c4dc7e","ref":{"seq":3,"row":1},"attrs":{"records":3,"reset":"none"}}
+{"t":"node","id":"epoch/main/0","revision":1,"kind":"epoch","parent":"stream/main","stream":"main","ref":{"seq":1,"row":1},"attrs":{"records":14,"reset":"none"}}
+{"t":"node","id":"epoch/main/s8-boundary","revision":1,"kind":"epoch","parent":"stream/main","stream":"main","ref":{"seq":1,"row":15},"refs":[{"seq":1,"row":15}],"attrs":{"continues_from":"s7-synthetic","records":2,"reset":"observed_replayable"}}
+{"t":"node","id":"inject/1/3","revision":1,"kind":"context.injection","parent":"run/talk_main_s1-cycle/s1-cycle","stream":"main","ref":{"seq":1,"row":3}}
+{"t":"node","id":"input/1/2","revision":1,"kind":"message.external","parent":"run/talk_main_s1-cycle/s1-cycle","stream":"main","ref":{"seq":1,"row":2}}
+{"t":"node","id":"msg/1/13:0","revision":1,"kind":"message.assistant","parent":"call/s6-call","stream":"main","ref":{"seq":1,"row":13,"block":0}}
+{"t":"node","id":"msg/1/14:0","revision":1,"kind":"message.synthetic","parent":"call/s7-synthetic-call","stream":"main","ref":{"seq":1,"row":14,"block":0}}
+{"t":"node","id":"msg/1/5:0","revision":1,"kind":"message.assistant","parent":"call/s3-call","stream":"main","ref":{"seq":1,"row":5,"block":0}}
+{"t":"node","id":"msg/3/2:0","revision":1,"kind":"message.assistant","parent":"call/checker-s1-call","stream":"a0a10ef0666c4dc7e","ref":{"seq":3,"row":2,"block":0}}
+{"t":"node","id":"notify/1/12","revision":1,"kind":"runtime.notification","parent":"run/talk_main_s1-cycle/s5-cycle-notification","stream":"main","ref":{"seq":1,"row":12}}
+{"t":"node","id":"output/a0a10ef0666c4dc7e","revision":1,"kind":"agent.output","parent":"run/talk_a0a10ef0666c4dc7e/a0a10ef0666c4dc7e-cycle","stream":"a0a10ef0666c4dc7e","ref":{"seq":3,"row":2},"refs":[{"seq":3,"row":2}],"attrs":{"returned_value":"unavailable"}}
+{"t":"node","id":"run/talk_a0a10ef0666c4dc7e/a0a10ef0666c4dc7e-cycle","revision":1,"kind":"run","parent":"talk/a0a10ef0666c4dc7e","stream":"a0a10ef0666c4dc7e","ref":{"seq":3,"row":1},"attrs":{"trigger":"external"}}
+{"t":"node","id":"run/talk_main_s1-cycle/s1-cycle","revision":1,"kind":"run","parent":"talk/main/s1-cycle","stream":"main","ref":{"seq":1,"row":2},"attrs":{"trigger":"external"}}
+{"t":"node","id":"run/talk_main_s1-cycle/s5-cycle-notification","revision":1,"kind":"run","parent":"talk/main/s1-cycle","stream":"main","ref":{"seq":1,"row":12},"attrs":{"trigger":"notification"}}
+{"t":"node","id":"run/talk_main_s8-cycle-compact/s8-cycle-compact","revision":1,"kind":"run","parent":"talk/main/s8-cycle-compact","stream":"main","ref":{"seq":1,"row":16},"attrs":{"trigger":"external"}}
+{"t":"node","id":"segment/at_1_2","revision":1,"kind":"segment","parent":"session/00000001-0000-4000-8000-000000000001","ref":{"seq":1,"row":2},"attrs":{"committable":false,"gates_unmet":["activity_boundary","lateness_watermark"],"state":"open","talks":2}}
+{"t":"node","id":"session/00000001-0000-4000-8000-000000000001","revision":1,"kind":"session","attrs":{"conversation":"00000001-0000-4000-8000-000000000001","from_time":"2026-10-12T01:37:27.579Z","through_time":"2026-10-12T01:37:38.679Z","title":"build and check","title_from":"observed_replayable"}}
+{"t":"node","id":"stream/a0a10ef0666c4dc7e","revision":1,"kind":"stream","parent":"session/00000001-0000-4000-8000-000000000001","stream":"a0a10ef0666c4dc7e","ref":{"seq":3,"row":1},"attrs":{"label":"checker","records":3,"role":"child"}}
+{"t":"node","id":"stream/main","revision":1,"kind":"stream","parent":"session/00000001-0000-4000-8000-000000000001","stream":"main","ref":{"seq":1,"row":1},"attrs":{"records":16,"role":"main"}}
+{"t":"node","id":"summary/1/16","revision":1,"kind":"epoch.summary","parent":"epoch/main/s8-boundary","stream":"main","ref":{"seq":1,"row":16}}
+{"t":"node","id":"talk/a0a10ef0666c4dc7e","revision":1,"kind":"talk","parent":"epoch/a0a10ef0666c4dc7e/0","stream":"a0a10ef0666c4dc7e","ref":{"seq":3,"row":1},"attrs":{"loops":1,"runs":1,"trigger":"unknown"}}
+{"t":"node","id":"talk/main/s1-cycle","revision":1,"kind":"talk","parent":"epoch/main/0","stream":"main","ref":{"seq":1,"row":2},"attrs":{"loops":2,"runs":2,"trigger":"external"}}
+{"t":"node","id":"talk/main/s8-cycle-compact","revision":1,"kind":"talk","parent":"epoch/main/s8-boundary","stream":"main","ref":{"seq":1,"row":16},"attrs":{"loops":1,"runs":1,"trigger":"external"}}
+{"t":"node","id":"think/1/4:0","revision":1,"kind":"thinking","parent":"call/s3-call","stream":"main","ref":{"seq":1,"row":4,"block":0}}
+{"t":"node","id":"tool/s5-tool","revision":1,"kind":"agent.call","parent":"call/s5-call","stream":"main","ref":{"seq":1,"row":10,"block":0},"refs":[{"seq":1,"row":10,"block":0},{"seq":1,"row":11,"block":0}],"attrs":{"name":"Agent","result":"available","result_join":"exact_unique","timing":"unavailable"}}
+{"t":"node","id":"tool/srvtool-websearch","revision":1,"kind":"tool","parent":"call/s4-call","stream":"main","ref":{"seq":1,"row":8,"block":0},"refs":[{"seq":1,"row":8,"block":0},{"seq":1,"row":9,"block":0}],"attrs":{"name":"WebSearch","result":"available","result_join":"exact_unique","timing":"unavailable"}}
+{"t":"node","id":"tool/tool-run-make-build","revision":1,"kind":"tool","parent":"call/s3-call","stream":"main","ref":{"seq":1,"row":6,"block":0},"refs":[{"seq":1,"row":6,"block":0},{"seq":1,"row":7,"block":0}],"attrs":{"name":"Bash","result":"available","result_join":"exact_unique","timing":"unavailable"}}
+{"t":"relation","id":"rel/ends_with/stream_a0a10ef0666c4dc7e/output_a0a10ef0666c4dc7e","revision":1,"type":"ends_with","from":"stream/a0a10ef0666c4dc7e","to":"output/a0a10ef0666c4dc7e","quality":"exact_unique","via":"the last response in the child stream, and what it returned","evidence":[{"seq":3,"row":2}]}
+{"t":"relation","id":"rel/follows/epoch_main_s8-boundary/epoch_main_0","revision":1,"type":"follows","from":"epoch/main/s8-boundary","to":"epoch/main/0","quality":"exact_unique","via":"explicit context reset","evidence":[{"seq":1,"row":15}]}
+{"t":"relation","id":"rel/in_segment/talk_a0a10ef0666c4dc7e/segment_at_1_2","revision":1,"type":"in_segment","from":"talk/a0a10ef0666c4dc7e","to":"segment/at_1_2","quality":"strong_inference","via":"inside the window of the talk that delegated it","evidence":[{"seq":3,"row":1}]}
+{"t":"relation","id":"rel/in_segment/talk_main_s1-cycle/segment_at_1_2","revision":1,"type":"in_segment","from":"talk/main/s1-cycle","to":"segment/at_1_2","quality":"exact_unique","via":"activity window","evidence":[{"seq":1,"row":2}]}
+{"t":"relation","id":"rel/in_segment/talk_main_s8-cycle-compact/segment_at_1_2","revision":1,"type":"in_segment","from":"talk/main/s8-cycle-compact","to":"segment/at_1_2","quality":"exact_unique","via":"activity window","evidence":[{"seq":1,"row":16}]}
+{"t":"relation","id":"rel/reports/notify_1_12/stream_a0a10ef0666c4dc7e","revision":1,"type":"reports","from":"notify/1/12","to":"stream/a0a10ef0666c4dc7e","quality":"exact_unique","via":"task id on the notification","evidence":[{"seq":1,"row":12}]}
+{"t":"relation","id":"rel/starts/tool_s5-tool/stream_a0a10ef0666c4dc7e","revision":1,"type":"starts","from":"tool/s5-tool","to":"stream/a0a10ef0666c4dc7e","quality":"exact_unique","via":"parent tool result","evidence":[{"seq":1,"row":11}]}
+{"t":"relation","id":"rel/summarizes/summary_1_16/boundary_1_15","revision":1,"type":"summarizes","from":"summary/1/16","to":"boundary/1/15","quality":"exact_unique","via":"containment parent","evidence":[{"seq":1,"row":16}]}
+{"t":"commit","digest":"4388f7a5d811657d3b8b2a4b9fc39e123497c4972b2603a026abdd14828b25c3","counts":{"nodes":35,"relations":8,"unresolved":0}}
diff --git a/tests/chain/watermark_test.go b/tests/chain/watermark_test.go
new file mode 100644
index 0000000..02694ed
--- /dev/null
+++ b/tests/chain/watermark_test.go
@@ -0,0 +1,94 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package chain
+
+import (
+ "testing"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/assemble"
+ "github.com/apache/skywalking-ai-sessionizer/internal/index"
+)
+
+// A round narrowed by the byte budget reads nothing past its watermark, not
+// even through a lookup. A reset at sequence 1 names the record it continues
+// from, and that record lands only at sequence 2. Assembled through sequence 1,
+// the round must report the record as missing: nothing it covers holds it, and
+// its header and input digest cover sequence 1 alone.
+func TestARoundReadsNothingPastItsWatermark(t *testing.T) {
+ ix := index.New("s")
+ in := ix.Strings
+ main := in.ID("main")
+ ix.Append(index.Entry{Seq: 1, Row: 1, Stream: main, Kind: index.KindUser, TS: 1000, Record: in.ID("u1"),
+ Flags: index.FlagExternalInput})
+ ix.Append(index.Entry{Seq: 1, Row: 2, Stream: main, Kind: index.KindSystem, TS: 1001, Record: in.ID("b1"),
+ Parent: in.ID("u1"), Continues: in.ID("x1"), Flags: index.FlagEpochBoundary})
+ ix.Append(index.Entry{Seq: 2, Row: 1, Stream: main, Kind: index.KindUser, TS: 1002, Record: in.ID("x1"),
+ Parent: in.ID("b1")})
+
+ open := func(through uint64) bool {
+ res, err := assemble.Session(ix, assemble.Options{Conversation: "s", Session: "s", ThroughSeq: through})
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, u := range res.Unresolved {
+ if u.Kind == "epoch_continuation" && u.RefID == "x1" {
+ return true
+ }
+ }
+ return false
+ }
+ if !open(1) {
+ t.Error("through sequence 1, the continuation landed at sequence 2 was found")
+ }
+ if open(2) {
+ t.Error("through sequence 2, the continuation is reported missing")
+ }
+ if _, ok := ix.EntryByRecord("x1"); !ok {
+ t.Error("narrowing a round changed the index it was given")
+ }
+}
+
+// The header counts Bash runs by the tool and agent call nodes the round
+// holds. Two tool uses whose ids give one node id are one node, so they count
+// once, as the fold shows them.
+func TestBashRunsCountTheNodes(t *testing.T) {
+ ix := index.New("s")
+ in := ix.Strings
+ main := in.ID("main")
+ for row, id := range []string{"a/b", "a_b"} {
+ ix.Append(index.Entry{Seq: 1, Row: uint32(2*row + 1), Stream: main, Kind: index.KindAssistant,
+ TS: int64(1000 + 2*row), Record: in.ID("u" + id), Call: in.ID("m" + id)},
+ index.Block{Kind: index.BlockToolUse, ToolID: in.ID(id), Name: in.ID("Bash")})
+ ix.Append(index.Entry{Seq: 1, Row: uint32(2*row + 2), Stream: main, Kind: index.KindUser,
+ TS: int64(1001 + 2*row), Record: in.ID("r" + id)},
+ index.Block{Kind: index.BlockToolResult, ToolID: in.ID(id)})
+ }
+ res, err := assemble.Session(ix, assemble.Options{Conversation: "s", Session: "s"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ nodes := 0
+ for _, n := range res.Nodes {
+ if n.Kind == "tool" {
+ nodes++
+ }
+ }
+ if nodes != 1 || res.ToolCalls["Bash"] != nodes {
+ t.Fatalf("%d tool nodes, and Bash counted %d times", nodes, res.ToolCalls["Bash"])
+ }
+}
diff --git a/tests/langchain_reset_test.go b/tests/langchain_reset_test.go
index c297076..d509c31 100644
--- a/tests/langchain_reset_test.go
+++ b/tests/langchain_reset_test.go
@@ -362,17 +362,16 @@ func TestABodyNamesThePromptItsCallWasPlacedUnder(t *testing.T) {
for _, kase := range []string{"subagent", "summarized"} {
zone, sessions := land(t, kase)
calls, bodies := map[string]string{}, map[string]string{}
+ // A response joins by its call and its request, and names no prompt.
+ // A request names it as its record's own run.
eachRecord(t, zone, sessions[0], func(_, _ uint64, rec sessiondata.Record) {
if rec.Call != "" && rec.Call == rec.ID {
calls[rec.Call] = rec.Run
}
- })
- // A response joins by its call and its request, and names no prompt.
- for id, m := range manifestsOf(t, zone, sessions[0]) {
- if strings.HasSuffix(id, ":request") {
- bodies[id] = m.Run
+ if strings.HasSuffix(rec.ID, ":request") {
+ bodies[rec.ID] = rec.Run
}
- }
+ })
checked := 0
for id, run := range bodies {
call := id[:strings.LastIndex(id, ":")]
diff --git a/tests/scenarios/fixture.expect.yaml b/tests/scenarios/fixture.expect.yaml
index 51eaa3b..bfcf434 100644
--- a/tests/scenarios/fixture.expect.yaml
+++ b/tests/scenarios/fixture.expect.yaml
@@ -22,8 +22,8 @@ checkpoints:
talks_on: {main: 1, checker: 1}
relations: {starts: 1, reports: 1, ends_with: 1}
nodes:
- tool/tool-run-make-build: {refs: 2, attrs: {name: Bash, result: available}}
- tool/srvtool-websearch: {refs: 2, attrs: {name: WebSearch, result: available}}
+ tool/tool-run-make-build: {refs: 2, attrs: {name: null, result: available}}
+ tool/srvtool-websearch: {refs: 2, attrs: {name: null, result: available}}
final:
rounds: 2
delta: true
diff --git a/tests/scenarios/mcp-calls.expect.yaml b/tests/scenarios/mcp-calls.expect.yaml
index f1f7893..8f4b5fe 100644
--- a/tests/scenarios/mcp-calls.expect.yaml
+++ b/tests/scenarios/mcp-calls.expect.yaml
@@ -23,7 +23,7 @@ checkpoints:
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/s5-tool: {kind: tool, attrs: {name: null, mcp_server: null, mcp_tool: null}}
tool/auditor-s1-tool: {kind: tool, stream: auditor, attrs: {mcp_server: docs, mcp_tool: read}}
view:
state: verified