You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2026/10/18 06:30:25 UTC
(skywalking-ai-sessionizer) 01/01: Show where a model call's history changed
This is an automated email from the ASF dual-hosted git repository.
wu-sheng pushed a commit to branch history-change-server
in repository https://gitbox.apache.org/repos/asf/skywalking-ai-sessionizer.git
commit febf719676dab0ffe9078a1b36351d9c714dd593
Author: Wu Sheng <wu...@foxmail.com>
AuthorDate: Sun Oct 18 14:29:48 2026 +0800
Show where a model call's history changed
Each model call whose request records its message list is compared with
the call before it in its stream and epoch, from the index alone. A
request that continues a provider thread holds only its new messages; its
history is the history of the call it continues, then that call's reply,
then its own messages. A call the framework made for its own purpose is
not compared and is skipped.
A call now says what it found: history is kept, changed, first,
not_compared, or unavailable with the reason, with the history's length,
the call it was compared with, and what became of that call's reply. A
Claude Code call also says whether the system prompt changed. A changed
history draws a history.change step just before the call, under the same
parent, naming the two requests' records, with what was kept, changed,
dropped and added, whether the order changed, and its cause unavailable.
It is no context reset: the calls stay in their epoch, and a reset is
still only an explicit record. The card carries no content.
Every walk is bounded and measured once a call: a thread's history is
measured without building it, a list over a base copies only the front it
keeps, and each base is checked on its own.
Two scenarios hold a trimmed history and a provider thread, and new tests
cover the comparison rules, the round chain publishing, revising and
removing a card, and a chain with no history getting it in rounds over no
new file. Scenario comparisons now compare every history value per call
and card. On every LangChain capture and on the Claude Code 2.1.302
capture, no history changes without a mark, so no card is drawn.
---
docs/en/adapters/claude-code-glossary.md | 3 +-
docs/en/adapters/claude-code.md | 2 +-
docs/en/adapters/kiro-glossary.md | 3 +-
docs/en/adapters/langsmith-glossary.md | 3 +-
docs/en/changes/changes.md | 13 +
docs/en/concepts-and-designs/glossary.md | 1 +
.../unified-conversation-model.md | 13 +
docs/en/formats/asz-view.md | 2 +
docs/en/formats/session-flow.md | 62 ++
docs/en/guides/scenario.md | 19 +-
internal/adapters/claudecode/glossary.go | 2 +
internal/adapters/kiro/glossary.go | 2 +
internal/adapters/langsmith/glossary.go | 2 +
internal/assemble/assemble.go | 10 +-
internal/assemble/calls.go | 16 +
internal/assemble/history.go | 764 +++++++++++++++++++++
internal/assemble/history_assembly_test.go | 580 ++++++++++++++++
internal/assemble/history_test.go | 110 +++
internal/assemble/provider.go | 7 +-
internal/scenario/expect/expect.go | 158 ++++-
internal/scenario/plan.go | 11 +-
internal/scenario/provider.go | 72 +-
internal/scenario/provider_test.go | 57 ++
internal/scenario/run/run.go | 19 +-
internal/scenario/scenario.go | 57 +-
internal/view/step.go | 5 +-
pkg/model/model.go | 5 +
pkg/model/term.go | 2 +-
pkg/sessionflow/before_test.go | 25 +
pkg/sessionflow/fold.go | 24 +
tests/chain/history_round_test.go | 395 +++++++++++
tests/claudecode_threads_test.go | 20 +
tests/history_view_test.go | 151 ++++
tests/langchain_history_test.go | 148 ++++
tests/scenarios/history-forget.expect.yaml | 38 +
tests/scenarios/history-forget.yaml | 34 +
tests/scenarios/history-threads.expect.yaml | 34 +
tests/scenarios/history-threads.yaml | 47 ++
38 files changed, 2885 insertions(+), 31 deletions(-)
diff --git a/docs/en/adapters/claude-code-glossary.md b/docs/en/adapters/claude-code-glossary.md
index 78b88cc..c59497a 100644
--- a/docs/en/adapters/claude-code-glossary.md
+++ b/docs/en/adapters/claude-code-glossary.md
@@ -19,7 +19,7 @@ asz conversation -terms native ID talk · promptId · message.id · tool_use +
asz conversation -terms both ID run (promptId)
```
-**63 terms: 39 that Claude Code names, 24 this project derives.** That ratio is worth watching. If
+**64 terms: 39 that Claude Code names, 25 this project derives.** That ratio is worth watching. If
everything had a native name, the model would be a rename of one product rather than a model of
any. If nothing did, it would have no evidence under it.
@@ -89,6 +89,7 @@ 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 |
+| `history.change` | derived by asz from the request bodies, when provider bodies are collected: Claude Code writes no such record |
| `in_segment` | the activity window a Talk falls in |
| `input_of` | unavailable; the serialized provider request is not written locally |
| `internal_call` | no equivalent: a request Claude Code sends for its own purpose, such as a compaction's, joins no call |
diff --git a/docs/en/adapters/claude-code.md b/docs/en/adapters/claude-code.md
index b724da6..3fbe6b6 100644
--- a/docs/en/adapters/claude-code.md
+++ b/docs/en/adapters/claude-code.md
@@ -862,7 +862,7 @@ fed one turn at a time:
| | Why |
| --- | --- |
| **Reasoning text** | Claude Code asks the provider not to return it, so most thinking blocks carry only a signature. `unavailable` where the runtime wrote none. See below. |
-| **Serialized request** | in the corpus this page was measured on, 2.1.220 – 2.1.251, the system prompt, tool schemas and cache annotations are absent from transcripts: 0 files contain `"tools":[`, and `compactMetadata.preservedMessages.allUuids` names 5 ids that exist nowhere on disk. asz produces no input manifest and no `input_of` relation, and reports model-context coverage as `unavailable`, as the model's [adapter contract](../concepts-and-designs/unified-conversation-model.md#adap [...]
+| **Serialized request** | in the corpus this page was measured on, 2.1.220 – 2.1.251, the system prompt, tool schemas and cache annotations are absent from transcripts: 0 files contain `"tools":[`, and `compactMetadata.preservedMessages.allUuids` names 5 ids that exist nowhere on disk. asz produces no input manifest and no `input_of` relation, and reports model-context coverage as `unavailable`, as the model's [adapter contract](../concepts-and-designs/unified-conversation-model.md#adap [...]
| **Injected preamble** | the instruction block prepended to the first user message has no transcript record. Its *data* survives as `attachment` records; its rendered form does not. |
| **Per-call duration and cost** | not written to transcripts. Available via OTLP. |
| **`tool.execution`** | no local record; the call and result are observable, the execution is not. |
diff --git a/docs/en/adapters/kiro-glossary.md b/docs/en/adapters/kiro-glossary.md
index c6f1bf5..a5aa733 100644
--- a/docs/en/adapters/kiro-glossary.md
+++ b/docs/en/adapters/kiro-glossary.md
@@ -21,7 +21,7 @@ asz conversation -terms native ID talk · executionId · tool_call and tool_re
asz conversation -terms both ID run (executionId)
```
-**63 terms: 24 that Kiro names, 39 this project derives.**
+**64 terms: 24 that Kiro names, 40 this project derives.**
---
@@ -82,6 +82,7 @@ not there.
| `exact_unique` | — | derived here | one toolCallId or subSessionId on both records |
| `follows` | — | derived here | derived from line order |
| `hash_only` | — | derived here | no equivalent |
+| `history.change` | — | derived here | never: every history change Kiro makes is a tombstone, a reset, and it records no model request |
| `in_segment` | — | derived here | derived |
| `input_of` | — | derived here | no equivalent |
| `internal_call` | — | derived here | no equivalent: Kiro records no model request, so no call's history is compared |
diff --git a/docs/en/adapters/langsmith-glossary.md b/docs/en/adapters/langsmith-glossary.md
index 86f00db..ba3bdae 100644
--- a/docs/en/adapters/langsmith-glossary.md
+++ b/docs/en/adapters/langsmith-glossary.md
@@ -21,7 +21,7 @@ asz conversation -terms native ID talk · trace_id · id · run_type: tool
asz conversation -terms both ID run (trace_id)
```
-**63 terms: 18 that LangChain names, 45 this project derives.** That ratio is worth watching against
+**64 terms: 18 that LangChain names, 46 this project derives.** That ratio is worth watching against
the Claude Code one. A framework that records its own graph in detail still has no word for a
conversation boundary, a context reset or a delegated agent — which is the whole reason this model
exists separately from any product's.
@@ -84,6 +84,7 @@ not there.
| `external` | — | derived here | the root run of a trace carries the message that opened it |
| `follows` | — | derived here | derived from order within a trace |
| `hash_only` | — | derived here | no equivalent |
+| `history.change` | — | derived here | derived by asz from the inputs of two runs of type llm: trim_messages, langmem and an edited message under one id change the history and write no mark |
| `in_segment` | — | derived here | derived |
| `message.synthetic` | — | derived here | no equivalent: nothing here fabricates a message on the model's behalf |
| `notification` | — | derived here | no equivalent |
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index 91e217c..c9367d0 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -484,6 +484,19 @@
## Conversation model
+- A call says what its history was compared with, and a card stands where it changed. Where a
+ runtime records its requests, Claude Code with provider bodies and LangChain, each call's history
+ is the message list its request recorded, built from the provider thread where a request
+ continues one, and is compared with the call before it in its stream and epoch. A call carries
+ `history`: `kept`, `changed`, `first`, `not_compared`, or `unavailable` with the reason, and the
+ earlier reply's status. A changed history draws a `history.change` step just before the call, with
+ what was kept, changed, dropped and added, and its cause `unavailable`: it is no context reset, and
+ the calls stay in their epoch. The rules are on the
+ [Session Flow](../formats/session-flow.md#where-a-calls-history-changed) page. On every LangChain
+ capture and on the Claude Code 2.1.302 capture, no history changes without a mark, so no card is
+ drawn; the scenarios `history-forget` and `history-threads` hold a trimmed history and a thread.
+ `provider_bodies_leave_the_fold` now lets provider bodies add the card, and nothing else.
+
- Model output whose record names no provider call is a step of its run. It was in no node at all,
because only a call held text and reasoning. No Claude Code or LangChain record lacks the call;
every Kiro record does, since Kiro ties no record to a request. Grouping such output into calls
diff --git a/docs/en/concepts-and-designs/glossary.md b/docs/en/concepts-and-designs/glossary.md
index 103c283..8ac2fe4 100644
--- a/docs/en/concepts-and-designs/glossary.md
+++ b/docs/en/concepts-and-designs/glossary.md
@@ -75,6 +75,7 @@ Run is triggered from outside the loop, by input or by the runtime reporting a c
| `runtime.notification` | the runtime telling the parent a child completed |
| `epoch.boundary` | an explicit model-context reset |
| `epoch.summary` | the carried-forward summary that reset produced |
+| `history.change` | where a call's history is not the one before it with more after it. **Not** a reset, and it names no cause |
| `error.api` | a provider error |
| `control.interrupt` | an externally initiated interruption |
| `control.permission` | a permission decision or mode change |
diff --git a/docs/en/concepts-and-designs/unified-conversation-model.md b/docs/en/concepts-and-designs/unified-conversation-model.md
index 42a877b..073bdba 100644
--- a/docs/en/concepts-and-designs/unified-conversation-model.md
+++ b/docs/en/concepts-and-designs/unified-conversation-model.md
@@ -116,6 +116,7 @@ rather than synthesised.
| | `runtime.notification` | the runtime informing the parent that a child completed |
| **Epoch** | `epoch.boundary` | an explicit model-context reset |
| | `epoch.summary` | the carried-forward summary produced by that reset |
+| **History** | `history.change` | where a call's history is not the history before it with more after it. It names no cause and carries no content |
| **Control** | `error.api` | a provider error, with retry state where available |
| | `control.interrupt` | an externally initiated interruption |
| | `control.permission` | a permission decision or mode change |
@@ -184,6 +185,18 @@ Four properties of this check matter:
evaluated over a reconstructed message list is a **structural cross-check on the reconstruction**,
not a statement about the wire. Adapters must report which of the two applies.
+asz evaluates a weaker form of the check where an adapter records each request's message list. A
+call is compared with the eligible call before it in its stream and epoch: one that is a turn, not
+work the framework did for itself, and whose list is readable. When the earlier history is an
+ordered prefix of the later one, the call's history is **kept**; otherwise it **changed**, and a
+`history.change` step stands before the call, with what was kept, changed, dropped and added. The
+earlier response's reply is reported apart, as found in the later history, sent back altered, not
+found, or unavailable, so an identical retry is kept with its reply not found. A call that cannot
+be compared says why, and is neither kept nor changed. A changed history names no cause and is not
+a reset, so the calls stay in their epoch. The rules are on the
+[Session Flow](../formats/session-flow.md#where-a-calls-history-changed) page; which runtimes
+record the lists is on each adapter's page.
+
## Evidence and qualification
Every material claim, statistic, diagram edge and rendered annotation carries or inherits a
diff --git a/docs/en/formats/asz-view.md b/docs/en/formats/asz-view.md
index 988685d..f3850cd 100644
--- a/docs/en/formats/asz-view.md
+++ b/docs/en/formats/asz-view.md
@@ -133,6 +133,8 @@ say.
| a tool or agent call adds | `name`, `failed`, `result`, `result_state`, `result_bytes`, `request_to_result_ms` and `request_to_result_join`, the time from the request record to the result record where the assembler joined them exactly; `changes` and `executions`, the records joined to it |
| a call to an MCP server has in `attrs` | `mcp_server` and `mcp_tool`, the server and the tool the runtime's name for the call addresses, where the name splits exactly (see [Parts](session-data.md#parts)). They are the runtime's names; which server ran the call, by its configured name, is in the call's execution record |
| a `turn.duration` step adds | `duration_ms`, `duration_measured_by` |
+| a `history.change` step | carries no `text`, `state`, `bytes`, `flags` or `dropped`: what it says is in its `attrs`, and the record it stands at is its call's. It comes just before the call it names in `to_call`. See [Where a call's history changed](session-flow.md#where-a-calls-history-changed) |
+| an `llm.call` has in `attrs` | `history` and the keys beside it, when its request was landed with them, as the same section says |
| `children` | containment, in record order: by landed position, and on one record a node on the whole record before the nodes on its parts, which follow their `block`, then by id: a talk holds runs, a run holds steps, a call holds what it produced. A tree is written down to twelve levels below its root, a talk or an entry of `loose`, and a node deeper than that is left out. No talk measured has more than three |
| `edges` | every relation touching the node, in both directions, as `{type, other, dir, quality, via}`, in the order the relations happened, each at the earliest record its evidence names, so a viewer draws cross-stream flow without searching `relations`; a workflow launch lists the streams it started in the order its run names them |
diff --git a/docs/en/formats/session-flow.md b/docs/en/formats/session-flow.md
index 37c7b9d..5aa5620 100644
--- a/docs/en/formats/session-flow.md
+++ b/docs/en/formats/session-flow.md
@@ -123,6 +123,7 @@ These are the prefixes asz writes:
| `runtime.notification` | `notify/<seq>/<row>` |
| `epoch.boundary` | `boundary/<seq>/<row>` |
| `epoch.summary` | `summary/<seq>/<row>` |
+| `history.change` | `history/<provider call id>`: the same key as the call whose history changed, `call/<provider call id>`, so a reader finds the call from the card |
| `error.api` | `error/<seq>/<row>` |
| `turn.duration` | `turn/<seq>/<row>` |
| `control.command` | `command/<seq>/<row>` |
@@ -159,6 +160,67 @@ The `session` node states `provider_bodies_landed`, how many bodies the session
round, joined or not, so a reader can say what was captured without opening a landed file. It is a
count, named apart from the list a call carries so that one name never means two things.
+#### Where a call's history changed
+
+A call's history is the message list its request records, as the request body's manifest writes it
+([Session Data, History](session-data.md#history)). A request that continues a provider thread
+holds only what follows the message it names: its history is that message's call's history, then
+that call's reply, then its own messages. The call it names must be in its stream, of its form, and
+have a reply. A history is built through at most 65,536 calls and holds at most 65,536 entries.
+
+Each call is compared with its predecessor: the nearest earlier call of its stream and epoch,
+skipping only an `internal_call`. A call of unknown class, or with no request or no list, is never
+skipped, since skipping a turn would compare a call with the one before it and show a change that
+did not happen. Calls are ordered by their first fragment's landed position, as the body join
+orders them. Every call whose request carries `history_form` has `history` in its `attrs`, the
+first row of this table that applies:
+
+| `history` | when | `history_reason` |
+| --- | --- | --- |
+| `not_compared` | the call is in an `auxiliary` stream, or is an `internal_call` | `auxiliary_stream`, `internal_call` |
+| `unavailable` | the call's class is unknown | `class_unknown` |
+| `unavailable` | its own history cannot be built | `messages:<reason>`, `thread_unresolved`, `thread_reply:<reason>` |
+| `first` | no call comes before it in its stream and epoch | |
+| `unavailable` | the predecessor has no request, or none written with `history_form`, or is of unknown class, or another form, or a history that cannot be built | `earlier_no_request`, `earlier_messages:<reason>`, `earlier_class_unknown`, `form_mismatch`, `earlier_<reason>` |
+| `kept` | the predecessor's history is an ordered prefix of the call's | |
+| `changed` | it is not | |
+
+Two entries are equal when both have an id or neither does, the ids are equal where they are, and
+the digests are equal. Beside `history`:
+
+- `history_messages`: how many entries the call's history holds, whenever it can be built, so a
+ page can say how many messages a continue request did not send.
+- `history_from`: on `kept` and `changed`, the predecessor's node id.
+- `history_reply`, and `history_reply_reason` when it is `unavailable`: what became of the
+ predecessor's reply. `unavailable` with `no_reply` when the predecessor's response names none,
+ and with `form_mismatch` when its form is not the requests'. On `kept` it is looked for after the
+ predecessor's history, where no entry is one of the predecessor's: `back` when an entry there is
+ equal to it, `back_changed` when one carries its id, else `not_found`; a request that continues
+ the predecessor's own reply is `back` by construction. On `changed` the reply is `unavailable`
+ with `ambiguous` when it is equal to an entry of the predecessor's history or shares its id with
+ one, and is otherwise looked for in the whole history.
+- `system_changed`, between two Claude Code calls: `yes`, `no`, or `unavailable`. A request that
+ creates a thread or carries none has its own system prompt; a continue request has the thread's,
+ unless it carries a system block other than the billing header, which makes it, and every call
+ continuing from it, unknown. `unavailable` when either call's is.
+
+A `changed` call has a `history.change` node beside it, under the same parent. It stands at the
+call's first fragment, its `refs` are the two requests' records, and it carries no content. It sorts
+just before the call: among nodes at one position, a `history.change` node is ordered by the call
+it names, and before it. Its `attrs`:
+
+| Key | Value |
+| --- | --- |
+| `from_call`, `to_call` | the predecessor and the call |
+| `kept`, `changed`, `dropped`, `added` | counts by multiset, so repeated messages need no unique ids: for each distinct entry the smaller of its numbers of occurrences in the two histories is `kept`; of what is left, for each id the smaller of its numbers of occurrences on the two sides is `changed`; the rest of the predecessor's is `dropped`, and of the call's `added`. A runtime that sends no message id never reports `changed`: an edited message is one dropped and one added |
+| `order_changed`, `order_changed_reason` | `yes` when the messages kept or changed are no longer in their earlier order, else `no`. An entry's identity is its id, else its digest; with no id repeated on a side, the order changed exactly when the longest common subsequence of the two identity sequences is shorter than `kept` and `changed` together. `unavailable` with `repeated_id` when an id occurs twice on a side, and with `too_large` past 4,194,304 pairs of equal identities |
+| `reply`, `reply_reason`, `system_changed` | as on the call |
+| `cause` | `unavailable`. A changed history names no cause: a context reset is an explicit `epoch.boundary`, and a changed history is not one. The calls stay in their epoch |
+
+All of it is derived from the index, carries no wall-clock time, and is assembled again each round,
+so a body or a response that lands later revises the call and its card, and a card whose pair is no
+longer `changed` is tombstoned.
+
A reference into the landed data is a landed sequence, a row, and optionally a content block. That
is the whole address. A reader takes both the content and the time from the `.sd` record it names.
The record's `time` field is when the runtime says it happened. The page that `asz view` serves
diff --git a/docs/en/guides/scenario.md b/docs/en/guides/scenario.md
index 59c15ba..7601ef9 100644
--- a/docs/en/guides/scenario.md
+++ b/docs/en/guides/scenario.md
@@ -210,6 +210,7 @@ session: mock-build-and-check # optional; default derived from the steps
title: build and check # optional
interval: 1s # the gap between steps, in every stream
provider_bodies: false # optional; true writes every call's request and response body
+provider_threads: false # optional, with provider_bodies; writes provider threads as Claude Code 2.1.302 does
system_prompt: | # optional; the system prompt this stream's calls send
You build and check a repository.
# optional, but all or nothing with system_prompt, and a stream must advertise
@@ -269,6 +270,8 @@ Every step is exactly one of these, plus an optional `after`, `checkpoint` and `
| `reset: {summary}` | a context reset and its summary, the summary timestamped before the boundary as the runtime writes it |
| `replay: N` | the last N main-stream records re-emitted with their run rewritten |
| `system: {subtype, fields}` | a system record of any subtype |
+| `forget: N` | the stream's next request leaves out the N oldest messages after the first, as an application trimming its history does. It writes no record: only the request shows it. It needs `provider_bodies`, and with threads the next request starts a new one |
+| `resume: true` | a new process takes the session up again, so the stream's next request starts a new provider thread with the whole list. It needs `provider_threads` |
A `tool` without a `result` is an unfinished tool. A `skill: {name, agent, steps}` is a fork whose
child is announced only in the parent's result. A `workflow: {name, children: [{name, prompt,
@@ -360,6 +363,8 @@ checkpoints:
unresolved_kinds: {tool_result: none}
session: {from: +0s, to: +11.1s} # the session node's range, as deltas from --at; title: its title
view: {state: verified, problems: 0, talks: 3, files: 6, first_talk: {label: run the build, runs: 2}}
+ # with provider_bodies: the model calls by their history attribute, none for a call with none
+ # history: {first: 1, kept: 2, changed: 1}
# with provider_bodies: the landed bodies, and the calls whose request is captured
# view: {provider_bodies: 14, captured_prompts: 7, provider_files: 5}
# the document's steps by each flag they carry
@@ -407,7 +412,8 @@ the same rounds; folding every round equals one full parse; rounds verify, link
writable; the landed files and rounds are self-sufficient without index and state; the head
round's header says what the fold holds; a parse with no new evidence writes nothing; every landed
record carries only the fields the format states a purpose for; a repack under the smallest budget
-keeps every record and the whole structure; and, for a runtime format, a second collect lands
+keeps every record and the whole structure, down to what each model call's history was and what
+each card says; and, for a runtime format, a second collect lands
nothing, every source line becomes one landed record, parts carrying source JSON keep its bytes,
discovery passes over the noise the writer plants beside the session, and a pass after Claude Code
prunes the session's files sets their active cursors to `source_gone` and lands nothing. Across
@@ -447,7 +453,11 @@ six.
Claude Code would write when `OTEL_LOG_RAW_API_BODIES` names a directory: the whole message list of
its stream again, the cache marker on the newest message, a billing header naming the prompt and
the previous call's request, and the session in `metadata.user_id`. A compaction starts the list
-again.
+again. With `provider_threads: true` they are written as Claude Code 2.1.302 writes them: a
+process's first request starts a provider thread, `{"type":"create"}`, and holds the whole list, the
+system prompt and the tools; every later request continues it from the response before,
+`{"type":"continue","previous_message_id":…}`, and holds only what is new, the billing header alone
+and no tools. A `resume` step, a `forget` step and a compaction each start a new thread.
The system prompt and the tools a request advertises come from `system_prompt` and `tools`, written
per stream: a scenario writes its own, and shares none with another scenario. A request body is read
@@ -467,7 +477,10 @@ sessions. `parts_keep_source_bytes` rebuilds every landed body and compares it w
`provider_bodies_leave_the_fold` builds the scenario again without bodies, with the same ids, and
compares the folds node by node, parent by parent and relation by relation. A node whose id is built
from a landed position is compared by the record it stands on, since files of another kind landing
-between shift positions. `changes_leave_the_fold` compares the same way.
+between shift positions. The one node bodies may add is a `history.change` card, which only a
+request can show, and the calls' history attributes are not compared, since only the build with
+bodies has them. `changes_leave_the_fold` compares the same way, with nothing left out. Every
+other comparison of two folds counts the calls by their history attribute too.
`provider_bodies_rebuild` reads every body a call lists the way a reader that loads bodies on demand
does: only the session's `provider_body` files up to the one the body's `ref` names, in order. Each
body must rebuild to its manifest's digest. It runs again on the root `repack_keeps_structure` re-cuts,
diff --git a/internal/adapters/claudecode/glossary.go b/internal/adapters/claudecode/glossary.go
index 129b833..0c74d1f 100644
--- a/internal/adapters/claudecode/glossary.go
+++ b/internal/adapters/claudecode/glossary.go
@@ -90,6 +90,8 @@ var glossary = model.NewGlossary(Dialect,
model.Term{Unified: model.KindRuntimeNotification, Native: "<task-notification>", Where: "a user record's content"},
model.Term{Unified: model.KindEpochBoundary, Native: "system/compact_boundary", Where: "a system record"},
model.Term{Unified: model.KindEpochSummary, Native: "isCompactSummary", Where: "a user record"},
+ model.Term{Unified: model.KindHistoryChange,
+ Note: "derived by asz from the request bodies, when provider bodies are collected: Claude Code writes no such record"},
model.Term{Unified: model.KindErrorAPI, Native: "system/api_error", Where: "a system record"},
// The runtime has a word for these two, so the Native value stays, but
// nothing reads it yet.
diff --git a/internal/adapters/kiro/glossary.go b/internal/adapters/kiro/glossary.go
index 537d6f9..89ab1cd 100644
--- a/internal/adapters/kiro/glossary.go
+++ b/internal/adapters/kiro/glossary.go
@@ -81,6 +81,8 @@ var glossary = model.NewGlossary(Dialect,
Note: "kind summarization or checkpoint_revert"},
model.Term{Unified: model.KindEpochSummary, Native: "assistant, operationType Summary", Where: "messages.jsonl",
Note: "nothing names the tombstone it follows"},
+ model.Term{Unified: model.KindHistoryChange,
+ Note: "never: every history change Kiro makes is a tombstone, a reset, and it records no model request"},
model.Term{Unified: model.KindErrorAPI, Native: "session_metadata, key displayError", Where: "messages.jsonl"},
model.Term{Unified: model.KindControlInterrupt, Note: "not read: a turn the person cancelled ends with stopReason cancelled, which lands as a notice"},
model.Term{Unified: model.KindControlPermission, Note: "not read: pending_interaction and interaction_resolved are landed and are no step"},
diff --git a/internal/adapters/langsmith/glossary.go b/internal/adapters/langsmith/glossary.go
index 58e8df6..f2fa39f 100644
--- a/internal/adapters/langsmith/glossary.go
+++ b/internal/adapters/langsmith/glossary.go
@@ -85,6 +85,8 @@ var glossary = model.NewGlossary(Dialect,
Note: "the first request sent a marked summary; trim_messages and langmem write no mark and make none"},
model.Term{Unified: model.KindEpochSummary, Native: "kwargs.content",
Where: "the marked message in the inputs of a run of type llm"},
+ model.Term{Unified: model.KindHistoryChange,
+ Note: "derived by asz from the inputs of two runs of type llm: trim_messages, langmem and an edited message under one id change the history and write no mark"},
model.Term{Unified: model.KindErrorAPI, Native: "error", Where: "a run",
Note: "a string with the exception and its traceback"},
model.Term{Unified: model.KindControlInterrupt, Note: "no equivalent"},
diff --git a/internal/assemble/assemble.go b/internal/assemble/assemble.go
index 5612b4a..cfac06a 100644
--- a/internal/assemble/assemble.go
+++ b/internal/assemble/assemble.go
@@ -163,8 +163,13 @@ type builder struct {
spawnEdges []spawnEdge
// bodiesOfCall is the provider bodies joined to each call node, which the
- // call carries into the round.
- bodiesOfCall map[string][]sessionflow.ProviderBody
+ // call carries into the round, and requestRow and responseRow the two
+ // bodies' places in the index's Bodies.
+ bodiesOfCall map[string][]sessionflow.ProviderBody
+ requestRow, responseRow map[string]int
+ // history is what comparing each call's history with its predecessor's
+ // found, by call node, which emitCall writes on the call and its card.
+ history map[string]*callHistory
// byPosition finds an entry by its landed position, built when first asked.
byPosition map[[2]uint32]int32
@@ -230,6 +235,7 @@ func Session(ix *index.Index, opt Options) (*Result, error) {
b.stage7Talks() // talks and runs built, steps placed
b.stage8Segments() // activity windows cut
b.joinProviderBodies()
+ b.compareHistories()
b.emitSteps() // leaves written, now that their containers are known
return b.result(), nil
diff --git a/internal/assemble/calls.go b/internal/assemble/calls.go
index 56b4fc6..1b3b10c 100644
--- a/internal/assemble/calls.go
+++ b/internal/assemble/calls.go
@@ -157,11 +157,27 @@ func (b *builder) emitCall(c *providerCall, parent string) {
if bodies := b.bodiesOfCall[c.NodeID]; len(bodies) > 0 {
a[sessionflow.ProviderBodiesAttr] = bodies
}
+ h := b.history[c.NodeID]
+ if h != nil {
+ for k, v := range h.attrs {
+ a[k] = v
+ }
+ }
b.node(sessionflow.Node{
Entity: sessionflow.Entity{ID: c.NodeID}, Kind: model.KindLLMCall,
Parent: parent, Stream: c.Stream.Name,
Ref: &refs[0], Refs: refs, Attrs: attrs(a),
})
+ // Where the call's history changed, a card of its own just before it,
+ // at the call's first fragment, naming the two requests it compares.
+ if h != nil && h.card != nil {
+ b.node(sessionflow.Node{
+ Entity: sessionflow.Entity{ID: sessionflow.NodeID("history", b.str(c.CallID))}, Kind: model.KindHistoryChange,
+ Parent: parent, Stream: c.Stream.Name,
+ Ref: &refs[0], Refs: h.refs, Attrs: attrs(h.card),
+ })
+ b.stats.Steps++
+ }
// The response's own content becomes steps inside the call. Their position
// is (record, block), which is stable because a landed record never changes.
diff --git a/internal/assemble/history.go b/internal/assemble/history.go
new file mode 100644
index 0000000..a9c0022
--- /dev/null
+++ b/internal/assemble/history.go
@@ -0,0 +1,764 @@
+// 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 assemble
+
+import (
+ "sort"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/index"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/model"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/providerbody"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessionflow"
+)
+
+// Where a call's history changed.
+//
+// A call's history is the message list its request records, as its adapter
+// wrote it in the body's manifest, read here from the index alone. A request
+// that continues a provider thread holds only what follows the message it
+// names; its history is that message's call's history, then that call's
+// reply, then its own messages. Each call is compared with the call before it
+// in its stream and epoch, skipping only the calls the framework made for its
+// own purpose. A history that is the one before it with more after it is kept;
+// anything else changed, and draws a history.change card before the call. The
+// cause is never inferred: a context reset is an explicit record, and a
+// changed history is not one. docs/en/formats/asz-view.md gives every rule.
+
+// maxHistoryLinks bounds the calls a thread's history is built through, and
+// maxHistoryEntries the entries a history holds; past either the history is
+// too large to compare. Variables only so a test can reach them without
+// building tens of thousands of calls.
+var (
+ maxHistoryLinks = 65536
+ maxHistoryEntries = 65536
+)
+
+// maxOrderPairs bounds the pairs of equal identities the order test visits.
+// Measured: 65,536 entries a side at this many pairs took about half a second.
+const maxOrderPairs = 4194304
+
+// The values a call's history attribute takes.
+const (
+ historyKept = "kept"
+ historyChanged = "changed"
+ historyFirst = "first"
+ historyNotCompared = "not_compared"
+)
+
+// The earlier reply's status.
+const (
+ replyBack = "back"
+ replyBackChanged = "back_changed"
+ replyNotFound = "not_found"
+)
+
+// callHistory is what comparing one call with its predecessor found.
+type callHistory struct {
+ // attrs go on the call's node.
+ attrs map[string]any
+ // card is the card's attributes, nil when the history is not changed,
+ // and refs the two requests' records the card names.
+ card map[string]any
+ refs []sessionflow.Ref
+}
+
+// hcall is one call as the comparison sees it.
+type hcall struct {
+ *call
+ pc *providerCall
+ req, resp int // rows in the index's Bodies, -1 for none
+ epoch string
+ aux bool
+ at int // place among its stream's calls
+}
+
+// historian holds what building histories needs and what it has built.
+type historian struct {
+ b *builder
+ // bodyOf is the row of each body's record, by its interned record id.
+ bodyOf map[uint32]int
+ // byMessage is each stream's calls by their message id.
+ byMessage map[uint32]map[uint32]*hcall
+ // measured is each call's history as far as it is known without building
+ // it: its length, how many continues it goes back through, or why it has
+ // none. Every entry is the call's own, never a limit another call's walk
+ // ran into.
+ measured map[string]measure
+ // systems is each Claude Code call's system prompt, once resolved.
+ systems map[string]system
+ // lists is each request body's own list as measured, by its row.
+ lists map[int]ownList
+ classes map[string]string
+}
+
+type measure struct {
+ length, links int
+ why string
+}
+
+type system struct {
+ digest uint32
+ known bool
+}
+
+// compareHistories fills b.history, for emitCall to write.
+func (b *builder) compareHistories() {
+ b.history = map[string]*callHistory{}
+ calls := b.callsInLineOrder()
+ if len(calls) == 0 {
+ return
+ }
+ pcs := map[string]*providerCall{}
+ for _, c := range b.calls {
+ pcs[c.NodeID] = c
+ }
+ h := &historian{b: b, bodyOf: map[uint32]int{}, byMessage: map[uint32]map[uint32]*hcall{},
+ measured: map[string]measure{}, systems: map[string]system{}, lists: map[int]ownList{},
+ classes: map[string]string{}}
+ for i := range b.ix.Bodies {
+ rec := b.ix.Entries[b.ix.Bodies[i].Entry].Record
+ if _, ok := h.bodyOf[rec]; !ok && rec != 0 {
+ h.bodyOf[rec] = i
+ }
+ }
+ byStream := map[uint32][]*hcall{}
+ for _, k := range calls {
+ pc := pcs[k.node]
+ c := &hcall{call: k, pc: pc, req: -1, resp: -1, epoch: b.epochOf(pc.Fragments[0])}
+ if r, ok := b.requestRow[k.node]; ok {
+ c.req = r
+ }
+ if r, ok := b.responseRow[k.node]; ok {
+ c.resp = r
+ }
+ if s, ok := b.byStream[k.stream]; ok {
+ c.aux = s.Role == model.StreamAuxiliary
+ }
+ c.at = len(byStream[k.stream])
+ byStream[k.stream] = append(byStream[k.stream], c)
+ if h.byMessage[k.stream] == nil {
+ h.byMessage[k.stream] = map[uint32]*hcall{}
+ }
+ if k.msg != 0 {
+ h.byMessage[k.stream][k.msg] = c
+ }
+ }
+ for _, list := range byStream {
+ // A call's predecessor is the nearest earlier call of its stream and
+ // epoch, skipping only the calls the framework made for its own
+ // purpose. A call of unknown class, or with no request or no list, is
+ // never skipped: had it been a turn, skipping it would compare the
+ // call with the one before it and show a change that did not happen.
+ // One walk forward keeps the nearest, so a run of internal calls costs
+ // nothing more.
+ var last *hcall
+ for _, n := range list {
+ p := last
+ if p != nil && p.epoch != n.epoch {
+ p = nil
+ }
+ if n.req >= 0 && b.ix.Bodies[n.req].Form != 0 {
+ b.history[n.node] = h.compare(n, p)
+ }
+ if h.class(n) != model.ClassInternalCall {
+ last = n
+ }
+ }
+ }
+}
+
+// class is a call's class: what its records say, and for Claude Code, whose
+// every call its transcript records is a turn, the form of its request.
+func (h *historian) class(c *hcall) string {
+ if class, ok := h.classes[c.node]; ok {
+ return class
+ }
+ class := ""
+ turn := false
+ for _, f := range c.pc.Fragments {
+ if f.Flags.Has(index.FlagInternalCall) {
+ class = model.ClassInternalCall
+ break
+ }
+ turn = turn || f.Flags.Has(index.FlagTurnCall)
+ }
+ if class == "" && (turn || h.formOf(c.req) == providerbody.FormClaudeCode) {
+ class = model.ClassTurnCall
+ }
+ h.classes[c.node] = class
+ return class
+}
+
+func (h *historian) formOf(row int) string {
+ if row < 0 {
+ return ""
+ }
+ return h.b.str(h.b.ix.Bodies[row].Form)
+}
+
+// compare decides one call's history attributes, in the order the format
+// page's table gives: the first row that applies.
+func (h *historian) compare(n, p *hcall) *callHistory {
+ out := &callHistory{attrs: map[string]any{}}
+ set := func(history, reason string) *callHistory {
+ out.attrs["history"] = history
+ if reason != "" {
+ out.attrs["history_reason"] = reason
+ }
+ return out
+ }
+ nm := h.length(n)
+ nWhy := nm.why
+ if nWhy == "" {
+ out.attrs["history_messages"] = nm.length
+ }
+ switch class := h.class(n); {
+ case n.aux:
+ return set(historyNotCompared, "auxiliary_stream")
+ case class == model.ClassInternalCall:
+ return set(historyNotCompared, model.ClassInternalCall)
+ case class == "":
+ return set(model.Unavailable, "class_unknown")
+ case nWhy != "":
+ return set(model.Unavailable, nWhy)
+ case p == nil:
+ return set(historyFirst, "")
+ }
+ switch {
+ case p.req < 0:
+ return set(model.Unavailable, "earlier_no_request")
+ case h.b.ix.Bodies[p.req].Form == 0:
+ return set(model.Unavailable, "earlier_messages:absent")
+ case h.class(p) == "":
+ return set(model.Unavailable, "earlier_class_unknown")
+ case h.formOf(p.req) != h.formOf(n.req):
+ return set(model.Unavailable, "form_mismatch")
+ }
+ if why := h.length(p).why; why != "" {
+ return set(model.Unavailable, "earlier_"+why)
+ }
+ out.attrs["history_from"] = p.node
+
+ // A request that continues its predecessor's reply keeps its history by
+ // construction, and its reply sits right after it; nothing is walked.
+ if y := &h.b.ix.Bodies[n.req]; y.Thread == index.ThreadContinue && y.ThreadPrevious == p.msg && p.msg != 0 {
+ set(historyKept, "")
+ out.attrs["history_reply"] = replyBack
+ h.system(out, p, n)
+ return out
+ }
+ P, _ := h.list(p)
+ N, _ := h.list(n)
+ reply, replyWhy := h.reply(p, n)
+ if isPrefix(P, N) {
+ set(historyKept, "")
+ h.replyAfter(out, reply, replyWhy, N[len(P):])
+ h.system(out, p, n)
+ return out
+ }
+ set(historyChanged, "")
+ h.replyIn(out, reply, replyWhy, P, N)
+ h.system(out, p, n)
+ card := counts(P, N)
+ card["from_call"], card["to_call"] = p.node, n.node
+ card["cause"] = model.Unavailable
+ card["reply"] = out.attrs["history_reply"]
+ if why, ok := out.attrs["history_reply_reason"]; ok {
+ card["reply_reason"] = why
+ }
+ if sc, ok := out.attrs["system_changed"]; ok {
+ card["system_changed"] = sc
+ }
+ out.card = card
+ out.refs = []sessionflow.Ref{h.refOf(p.req), h.refOf(n.req)}
+ return out
+}
+
+func (h *historian) refOf(row int) sessionflow.Ref {
+ e := &h.b.ix.Entries[h.b.ix.Bodies[row].Entry]
+ return sessionflow.Ref{Seq: uint64(e.Seq), Row: uint64(e.Row)}
+}
+
+// length is how many entries a call's history holds, how many continues it
+// goes back through, or why it has none. A thread's history is measured
+// through the calls it continues without building it: one walk back to where
+// the measure is known, then forward, keeping each call's own result, so a
+// long session costs one walk and no call's result depends on where a walk
+// began.
+func (h *historian) length(c *hcall) measure {
+ if m, ok := h.measured[c.node]; ok {
+ return m
+ }
+ // Back through the thread to a call whose measure is known or that holds
+ // its whole list, noting what each continues.
+ var chain []*hcall
+ next := map[string]*hcall{}
+ onChain := map[string]bool{}
+ for x := c; ; {
+ if _, ok := h.measured[x.node]; ok {
+ break
+ }
+ if onChain[x.node] {
+ // A thread that comes back to itself has no start.
+ for _, y := range chain {
+ h.measured[y.node] = measure{why: "thread_unresolved"}
+ }
+ return h.measured[c.node]
+ }
+ onChain[x.node] = true
+ chain = append(chain, x)
+ if x.req < 0 {
+ break
+ }
+ y, why := h.continued(x)
+ if why != "" || y == nil {
+ break
+ }
+ next[x.node] = y
+ x = y
+ }
+ for i := len(chain) - 1; i >= 0; i-- {
+ h.measured[chain[i].node] = h.measureOne(chain[i], next[chain[i].node])
+ }
+ return h.measured[c.node]
+}
+
+// measureOne is one call's measure, from the measure of the call it
+// continues, which is known.
+func (h *historian) measureOne(c, x *hcall) measure {
+ if c.req < 0 {
+ return measure{why: "no_request"}
+ }
+ own := h.ownOf(c.req)
+ switch {
+ case own.why != "":
+ return measure{why: own.why}
+ case own.length > maxHistoryEntries:
+ return measure{why: "messages:" + providerbody.TooLarge}
+ }
+ if x == nil {
+ if _, why := h.continued(c); why != "" {
+ return measure{why: why}
+ }
+ return measure{length: own.length}
+ }
+ before := h.measured[x.node]
+ switch {
+ case before.why == "messages:"+providerbody.TooLarge:
+ return measure{why: before.why}
+ case before.why != "":
+ return measure{why: "thread_unresolved"}
+ case before.links+1 > maxHistoryLinks || before.length+1+own.length > maxHistoryEntries:
+ return measure{why: "messages:" + providerbody.TooLarge}
+ }
+ return measure{length: before.length + 1 + own.length, links: before.links + 1}
+}
+
+// continued is the call a continue request goes on from, nil for a request
+// that holds its whole list, or why the thread cannot be followed: the call it
+// names is not in its stream, or is of another form, or gave no reply.
+func (h *historian) continued(c *hcall) (*hcall, string) {
+ y := &h.b.ix.Bodies[c.req]
+ if y.Thread != index.ThreadContinue {
+ return nil, ""
+ }
+ x := h.byMessage[c.stream][y.ThreadPrevious]
+ if x == nil || x.req < 0 || h.formOf(x.req) != h.formOf(c.req) {
+ return nil, "thread_unresolved"
+ }
+ if _, why := h.reply(x, c); why != "" {
+ return nil, "thread_reply:" + why
+ }
+ return x, ""
+}
+
+// list builds a call's whole history, walking a thread back to its start.
+func (h *historian) list(c *hcall) ([]index.Message, string) {
+ m := h.length(c)
+ if m.why != "" {
+ return nil, m.why
+ }
+ var parts [][]index.Message
+ for x := c; x != nil; {
+ own, _ := h.own(x.req)
+ parts = append(parts, own)
+ next, _ := h.continued(x)
+ if next != nil {
+ r, _ := h.reply(next, x)
+ parts = append(parts, []index.Message{r})
+ }
+ x = next
+ }
+ out := make([]index.Message, 0, m.length)
+ for i := len(parts) - 1; i >= 0; i-- {
+ out = append(out, parts[i]...)
+ }
+ return out, ""
+}
+
+// ownList is what a request's own body says about its list, without building
+// it: how many entries it holds and how many bases deep its copies go, or why
+// it has none.
+type ownList struct {
+ length, depth int
+ why string
+}
+
+// ownOf measures a request's own list: whole, or against the body its copy
+// came from. A base must be a request landed earlier, of the same form, whose
+// own list holds, at most a body's copies deep, at least the entries kept from
+// it. A list over a base that does not hold is not read, however few of its
+// entries a later list keeps. One walk back to a measured or whole body, then
+// forward, so a long run of copies costs one walk.
+func (h *historian) ownOf(row int) ownList {
+ var chain []int
+ for r := row; ; {
+ if _, ok := h.lists[r]; ok {
+ break
+ }
+ chain = append(chain, r)
+ y := &h.b.ix.Bodies[r]
+ if y.List != index.ListRelative {
+ break
+ }
+ base, ok := h.bodyOf[y.ListBase]
+ if !ok || base >= r {
+ break
+ }
+ r = base
+ }
+ for i := len(chain) - 1; i >= 0; i-- {
+ h.lists[chain[i]] = h.ownOne(chain[i])
+ }
+ return h.lists[row]
+}
+
+// ownOne is one body's measure, from its base's, which is known.
+func (h *historian) ownOne(row int) ownList {
+ y := &h.b.ix.Bodies[row]
+ switch y.List {
+ case index.ListUnavailable:
+ return ownList{why: "messages:" + h.b.str(y.ListWhy)}
+ case index.ListAbsent:
+ return ownList{why: "messages:absent"}
+ case index.ListWhole:
+ return ownList{length: int(y.MsgCount)}
+ }
+ bad := ownList{why: "messages:" + providerbody.UnrecognisedShape}
+ base, ok := h.bodyOf[y.ListBase]
+ if !ok || base >= row {
+ return bad
+ }
+ if b := &h.b.ix.Bodies[base]; b.Role != index.BodyRoleRequest || b.Form != y.Form {
+ return bad
+ }
+ prior := h.lists[base]
+ if prior.why != "" || prior.depth >= providerbody.MaxDepth || prior.length < int(y.ListKeep) {
+ return bad
+ }
+ return ownList{length: int(y.ListKeep) + int(y.MsgCount), depth: prior.depth + 1}
+}
+
+// own is a request's whole own list, or why there is none to compare: a list
+// over the entry bound is too large, though a later list may keep its front.
+func (h *historian) own(row int) ([]index.Message, string) {
+ l := h.ownOf(row)
+ switch {
+ case l.why != "":
+ return nil, l.why
+ case l.length > maxHistoryEntries:
+ return nil, "messages:" + providerbody.TooLarge
+ }
+ y := &h.b.ix.Bodies[row]
+ if y.List == index.ListWhole {
+ return h.b.ix.Messages[y.MsgFirst : y.MsgFirst+y.MsgCount], ""
+ }
+ out := make([]index.Message, l.length)
+ h.fill(row, out)
+ return out, ""
+}
+
+// fill writes the first len(out) entries of a request's measured list into
+// out: what each body adds goes in its place, and the walk goes on to its
+// base only for the front it keeps, so nothing past the front is copied.
+func (h *historian) fill(row int, out []index.Message) {
+ for {
+ y := &h.b.ix.Bodies[row]
+ mine := h.b.ix.Messages[y.MsgFirst : y.MsgFirst+y.MsgCount]
+ if y.List == index.ListWhole {
+ copy(out, mine)
+ return
+ }
+ if keep := int(y.ListKeep); len(out) > keep {
+ copy(out[keep:], mine)
+ out = out[:keep]
+ }
+ row = h.bodyOf[y.ListBase]
+ }
+}
+
+// reply is the message p's response produced, for comparing with n, or why
+// there is none to compare: no response or none it names, or one of another
+// form than n's request.
+func (h *historian) reply(p, n *hcall) (index.Message, string) {
+ if p.resp < 0 {
+ return index.Message{}, "no_reply"
+ }
+ y := &h.b.ix.Bodies[p.resp]
+ if y.Reply != index.ReplyEntry {
+ return index.Message{}, "no_reply"
+ }
+ if h.formOf(p.resp) != h.formOf(n.req) {
+ return index.Message{}, "form_mismatch"
+ }
+ return index.Message{HasID: y.ReplyHasID, ID: y.ReplyID, Digest: y.ReplyDigest}, ""
+}
+
+// replyAfter finds the earlier reply in what a kept history added after the
+// earlier one, where no entry is one of the earlier history's.
+func (h *historian) replyAfter(out *callHistory, r index.Message, why string, after []index.Message) {
+ if why != "" {
+ out.attrs["history_reply"], out.attrs["history_reply_reason"] = model.Unavailable, why
+ return
+ }
+ out.attrs["history_reply"] = find(r, after)
+}
+
+// replyIn finds the earlier reply in a changed history. The counts match
+// occurrences by multiset, so a reply equal to an entry of the earlier
+// history, or sharing its id, could be either, and none is claimed.
+func (h *historian) replyIn(out *callHistory, r index.Message, why string, P, N []index.Message) {
+ if why == "" {
+ for _, e := range P {
+ if e == r || (r.HasID && e.HasID && e.ID == r.ID) {
+ why = "ambiguous"
+ break
+ }
+ }
+ }
+ h.replyAfter(out, r, why, N)
+}
+
+func find(r index.Message, in []index.Message) string {
+ result := replyNotFound
+ for _, e := range in {
+ switch {
+ case e == r:
+ return replyBack
+ case r.HasID && e.HasID && e.ID == r.ID:
+ result = replyBackChanged
+ }
+ }
+ return result
+}
+
+// system says whether the system prompt changed between two Claude Code
+// calls. A continue request's system is the thread's, unless it carries a
+// block of its own, which may add to the thread's or replace it: that is not
+// measured, so it, and every call continuing from it, is not known.
+func (h *historian) system(out *callHistory, p, n *hcall) {
+ if h.formOf(n.req) != providerbody.FormClaudeCode {
+ return
+ }
+ a, aok := h.systemOf(p)
+ b, bok := h.systemOf(n)
+ switch {
+ case !aok || !bok:
+ out.attrs["system_changed"] = model.Unavailable
+ case a != b:
+ out.attrs["system_changed"] = "yes"
+ default:
+ out.attrs["system_changed"] = "no"
+ }
+}
+
+// systemOf is a call's system prompt, kept once resolved: a continue
+// request's comes from the call it continues, walked back iteratively to
+// where it is known. The walk has no bound of its own: what it walks a
+// comparison has measured within the link bound, and every answer kept is
+// the call's own, never where a bound stopped some other call's walk.
+func (h *historian) systemOf(c *hcall) (uint32, bool) {
+ var chain []*hcall
+ onChain := map[string]bool{}
+ got := system{}
+ for x := c; ; {
+ if s, ok := h.systems[x.node]; ok {
+ got = s
+ break
+ }
+ if onChain[x.node] {
+ // A thread that comes back to itself has no start.
+ got = system{}
+ break
+ }
+ onChain[x.node] = true
+ chain = append(chain, x)
+ if x.req < 0 {
+ got = system{}
+ break
+ }
+ y := &h.b.ix.Bodies[x.req]
+ if y.SystemUnknown {
+ got = system{}
+ break
+ }
+ if y.Thread != index.ThreadContinue {
+ got = system{digest: y.System, known: true}
+ break
+ }
+ if y.System != 0 {
+ got = system{}
+ break
+ }
+ next, why := h.continued(x)
+ if why != "" || next == nil {
+ got = system{}
+ break
+ }
+ x = next
+ }
+ // Every call walked shares the answer: a continue with no system block
+ // of its own has the thread's, and one whose walk found no start has none
+ // known. The call that decided it has its own answer, which is the same.
+ for _, x := range chain {
+ h.systems[x.node] = got
+ }
+ return got.digest, got.known
+}
+
+func isPrefix(p, n []index.Message) bool {
+ if len(p) > len(n) {
+ return false
+ }
+ for i := range p {
+ if p[i] != n[i] {
+ return false
+ }
+ }
+ return true
+}
+
+// identity is an entry's id where it has one, else its digest; the two kinds
+// are never equal.
+type identity struct {
+ hasID bool
+ v uint64
+}
+
+func identityOf(m index.Message) identity {
+ if m.HasID {
+ return identity{true, m.ID}
+ }
+ return identity{false, m.Digest}
+}
+
+// counts explains a changed history by multiset counts, which need no unique
+// keys: repeated messages are common, such as a person typing "continue"
+// twice.
+func counts(P, N []index.Message) map[string]any {
+ cp, cn := map[index.Message]int{}, map[index.Message]int{}
+ for _, e := range P {
+ cp[e]++
+ }
+ for _, e := range N {
+ cn[e]++
+ }
+ kept := 0
+ leftP, leftN := map[uint64]int{}, map[uint64]int{}
+ for e, n := range cp {
+ k := min(n, cn[e])
+ kept += k
+ if e.HasID {
+ leftP[e.ID] += n - k
+ }
+ }
+ for e, n := range cn {
+ if e.HasID {
+ leftN[e.ID] += n - min(n, cp[e])
+ }
+ }
+ changed := 0
+ for id, n := range leftP {
+ changed += min(n, leftN[id])
+ }
+ out := map[string]any{
+ "kept": kept, "changed": changed,
+ "dropped": len(P) - kept - changed, "added": len(N) - kept - changed,
+ }
+ switch order, why := orderChanged(P, N, kept+changed); {
+ case why != "":
+ out["order_changed"], out["order_changed_reason"] = model.Unavailable, why
+ case order:
+ out["order_changed"] = "yes"
+ default:
+ out["order_changed"] = "no"
+ }
+ return out
+}
+
+// orderChanged says whether the messages kept or changed are no longer in
+// their earlier order. An id that occurs twice on a side leaves which earlier
+// message became which unknown. Otherwise an id names one message on each
+// side and a repeated digest names identical messages, so the order changed
+// exactly when the longest common subsequence of the two identity sequences
+// is shorter than the messages matched.
+func orderChanged(P, N []index.Message, matched int) (bool, string) {
+ if repeatedID(P) || repeatedID(N) {
+ return false, "repeated_id"
+ }
+ at := map[identity][]int{}
+ for j := len(N) - 1; j >= 0; j-- {
+ at[identityOf(N[j])] = append(at[identityOf(N[j])], j)
+ }
+ pairs := 0
+ for _, e := range P {
+ pairs += len(at[identityOf(e)])
+ }
+ if pairs > maxOrderPairs {
+ return false, providerbody.TooLarge
+ }
+ // Hunt and Szymanski: the matching positions of N, each taken in
+ // descending order, are a longest increasing subsequence away from the
+ // longest common subsequence.
+ var tails []int
+ for _, e := range P {
+ for _, j := range at[identityOf(e)] {
+ k := sort.SearchInts(tails, j)
+ if k == len(tails) {
+ tails = append(tails, j)
+ } else {
+ tails[k] = j
+ }
+ }
+ }
+ return len(tails) < matched, ""
+}
+
+func repeatedID(list []index.Message) bool {
+ seen := map[uint64]bool{}
+ for _, e := range list {
+ if !e.HasID {
+ continue
+ }
+ if seen[e.ID] {
+ return true
+ }
+ seen[e.ID] = true
+ }
+ return false
+}
diff --git a/internal/assemble/history_assembly_test.go b/internal/assemble/history_assembly_test.go
new file mode 100644
index 0000000..aeaaaf2
--- /dev/null
+++ b/internal/assemble/history_assembly_test.go
@@ -0,0 +1,580 @@
+// 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 assemble
+
+import (
+ "encoding/json"
+ "fmt"
+ "runtime"
+ "testing"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/index"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/model"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/providerbody"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessionflow"
+)
+
+// A history fixture: a session built record by record, each landing in a file
+// of its own, so landing order is the order a test writes them in.
+type hfix struct {
+ t *testing.T
+ ix *index.Index
+ seq uint32
+}
+
+func newHfix(t *testing.T) *hfix { return &hfix{t: t, ix: index.New("s")} }
+
+func (f *hfix) land(hdr *sessiondata.Header, rec *sessiondata.Record) {
+ f.seq++
+ e, blocks, body := index.FromRecord(f.ix, hdr, rec, f.seq, 1)
+ f.ix.AppendRecord(e, blocks, body)
+}
+
+func transcriptOf(stream string) *sessiondata.Header {
+ return &sessiondata.Header{Kind: sessiondata.KindTranscript, Session: "s", Stream: stream}
+}
+
+var bodies = &sessiondata.Header{Kind: sessiondata.KindProviderBody, Session: "s", Src: "."}
+
+func (f *hfix) input(stream, run string) {
+ f.land(transcriptOf(stream), &sessiondata.Record{ID: run + "-in", From: sessiondata.FromExternal, Run: run,
+ Flags: []string{"external_input"}, Parts: []sessiondata.Part{{Kind: sessiondata.PartText, Text: "q", State: "available"}}})
+}
+
+// call lands a model call's record; flags add to "finished".
+func (f *hfix) call(stream, run, call string, flags ...string) {
+ f.land(transcriptOf(stream), &sessiondata.Record{ID: call + "-r", From: sessiondata.FromAgent, Call: call, Run: run,
+ Flags: append([]string{"finished"}, flags...), Parts: []sessiondata.Part{{Kind: sessiondata.PartText, Text: "a", State: "available"}}})
+}
+
+func (f *hfix) manifest(id, call, role string, fields map[string]any) {
+ m := map[string]any{"schema": providerbody.Schema, "role": role, "sha256": "d", "bytes": 2, "depth": 0,
+ "segments": []any{map[string]any{"lit": "{}"}}}
+ for k, v := range fields {
+ m[k] = v
+ }
+ data, err := json.Marshal(m)
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ f.land(bodies, &sessiondata.Record{ID: id, Call: call, Parts: []sessiondata.Part{{Kind: sessiondata.PartData, Data: data, State: "available"}}})
+}
+
+// request lands a call's request with its history fields; a nil list leaves
+// the body as one landed before asz wrote any.
+func (f *hfix) request(call, form string, fields map[string]any) {
+ all := map[string]any{}
+ if form != "" {
+ all["history_form"] = form
+ }
+ for k, v := range fields {
+ all[k] = v
+ }
+ f.manifest(call+":request", call, providerbody.RoleRequest, all)
+}
+
+func (f *hfix) response(call, form string, reply map[string]any) {
+ f.manifest(call+":response", call, providerbody.RoleResponse, map[string]any{"history_form": form, "reply": reply})
+}
+
+// e is an entry: an id when id is not 0, and its digest.
+func e(id, digest uint64) map[string]any {
+ out := map[string]any{"digest": fmt.Sprintf("%016x", digest)}
+ if id != 0 {
+ out["id"] = fmt.Sprintf("%016x", id)
+ }
+ return out
+}
+
+func whole(entries ...map[string]any) map[string]any {
+ list := make([]any, len(entries))
+ for i, x := range entries {
+ list[i] = x
+ }
+ return map[string]any{"messages": map[string]any{"entries": list}}
+}
+
+// turn lands one LangChain turn: a turn call with its request and its reply.
+func (f *hfix) turn(stream, call string, reply map[string]any, entries ...map[string]any) {
+ f.call(stream, "run", call, model.ClassTurnCall)
+ f.request(call, providerbody.FormLangChain, whole(entries...))
+ f.response(call, providerbody.FormLangChain, reply)
+}
+
+// assembled is every node's attributes by id, and the order of the nodes
+// under each parent.
+func (f *hfix) assembled(through uint64) (map[string]map[string]any, map[string][]string) {
+ f.t.Helper()
+ res, err := Session(f.ix, Options{Conversation: "s", Session: "s", ThroughSeq: through})
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ attrs := map[string]map[string]any{}
+ view := sessionflow.NewView("s")
+ for i := range res.Nodes {
+ n := &res.Nodes[i]
+ var a map[string]any
+ _ = json.Unmarshal(n.Attrs, &a)
+ if a == nil {
+ a = map[string]any{}
+ }
+ a["kind"], a["parent"] = n.Kind, n.Parent
+ attrs[n.ID] = a
+ view.Nodes[n.ID] = n
+ }
+ order := map[string][]string{}
+ for _, n := range res.Nodes {
+ for _, k := range view.Children(n.ID) {
+ order[n.ID] = append(order[n.ID], k.ID)
+ }
+ }
+ return attrs, order
+}
+
+func check(t *testing.T, attrs map[string]map[string]any, id string, want map[string]any) {
+ t.Helper()
+ got, ok := attrs[id]
+ if !ok {
+ t.Fatalf("no node %s", id)
+ }
+ for k, v := range want {
+ if fmt.Sprint(got[k]) != fmt.Sprint(v) {
+ t.Errorf("%s: %s is %v, want %v", id, k, got[k], v)
+ }
+ }
+}
+
+var (
+ sys = e(0, 0x5)
+ q1 = e(1, 0x10)
+ a1 = e(2, 0x20)
+ q2 = e(3, 0x30)
+ a2 = e(4, 0x40)
+ q3 = e(5, 0x50)
+ a1x = e(2, 0x21) // a1 sent back altered
+ none = map[string]any{"unavailable": "no_single_reply"}
+)
+
+// TestAHistoryThatGrowsIsKeptAndOneTrimmedChanges: the two verdicts, the
+// reply's status on each, and the card with what it counts, where it stands
+// and what it names.
+func TestAHistoryThatGrowsIsKeptAndOneTrimmedChanges(t *testing.T) {
+ f := newHfix(t)
+ f.input("main", "run")
+ f.turn("main", "c1", a1, sys, q1)
+ f.turn("main", "c2", a2, sys, q1, a1, q2)
+ f.turn("main", "c3", none, sys, q2, a2, q3) // q1 and a1 trimmed
+ attrs, order := f.assembled(0)
+ check(t, attrs, "call/c1", map[string]any{"history": "first", "history_messages": 2})
+ check(t, attrs, "call/c2", map[string]any{"history": "kept", "history_from": "call/c1", "history_reply": "back", "history_messages": 4})
+ check(t, attrs, "call/c3", map[string]any{"history": "changed", "history_from": "call/c2", "history_reply": "back"})
+ check(t, attrs, "history/c3", map[string]any{"kind": model.KindHistoryChange, "from_call": "call/c2", "to_call": "call/c3",
+ "kept": 2, "changed": 0, "dropped": 2, "added": 2, "order_changed": "no", "reply": "back", "cause": "unavailable",
+ "parent": attrs["call/c3"]["parent"]})
+ if _, ok := attrs["history/c2"]; ok {
+ t.Fatal("a kept history drew a card")
+ }
+ // The card comes just before its call among the call's siblings.
+ siblings := order[fmt.Sprint(attrs["call/c3"]["parent"])]
+ at := map[string]int{}
+ for i, id := range siblings {
+ at[id] = i
+ }
+ if at["history/c3"]+1 != at["call/c3"] {
+ t.Fatalf("siblings %v: the card is not just before its call", siblings)
+ }
+}
+
+// TestAShorterHistoryChanges: a history that is the one before it with its
+// end taken off is not kept, though every message it holds was there.
+func TestAShorterHistoryChanges(t *testing.T) {
+ f := newHfix(t)
+ f.input("main", "run")
+ f.turn("main", "c1", a1, sys, q1, q2)
+ f.turn("main", "c2", none, sys, q1)
+ attrs, _ := f.assembled(0)
+ check(t, attrs, "call/c2", map[string]any{"history": "changed", "history_from": "call/c1", "history_reply": "not_found"})
+ check(t, attrs, "history/c2", map[string]any{"kept": 2, "changed": 0, "dropped": 1, "added": 0, "order_changed": "no"})
+}
+
+// TestAListIsReadOnlyOverABaseThatHolds: a base must be a request, and one
+// whose own list holds what is kept from it, however little a later list
+// takes of it.
+func TestAListIsReadOnlyOverABaseThatHolds(t *testing.T) {
+ t.Run("a base that is a response", func(t *testing.T) {
+ f := newHfix(t)
+ f.input("main", "run")
+ f.call("main", "run", "c1", model.ClassTurnCall)
+ f.request("c1", providerbody.FormLangChain, whole(sys, q1))
+ f.manifest("c1:response", "c1", providerbody.RoleResponse, map[string]any{"history_form": providerbody.FormLangChain,
+ "reply": a1, "messages": map[string]any{"entries": []any{sys, q1, a1}}})
+ f.call("main", "run", "c2", model.ClassTurnCall)
+ f.request("c2", providerbody.FormLangChain, map[string]any{"messages": map[string]any{"base": "c1:response", "keep": 3, "add": []any{q2}}})
+ attrs, _ := f.assembled(0)
+ check(t, attrs, "call/c2", map[string]any{"history": "unavailable", "history_reason": "messages:unrecognised_shape"})
+ })
+ t.Run("a base that keeps more than its own base holds", func(t *testing.T) {
+ f := newHfix(t)
+ f.input("main", "run")
+ f.turn("main", "c1", none, sys)
+ f.call("main", "run", "c2", model.ClassTurnCall)
+ f.request("c2", providerbody.FormLangChain, map[string]any{"messages": map[string]any{"base": "c1:request", "keep": 2, "add": []any{q1}}})
+ f.call("main", "run", "c3", model.ClassTurnCall)
+ f.request("c3", providerbody.FormLangChain, map[string]any{"messages": map[string]any{"base": "c2:request", "keep": 1, "add": []any{q2}}})
+ attrs, _ := f.assembled(0)
+ check(t, attrs, "call/c2", map[string]any{"history": "unavailable", "history_reason": "messages:unrecognised_shape"})
+ // c3 keeps only the entry c1 does hold, but c2's list does not.
+ check(t, attrs, "call/c3", map[string]any{"history": "unavailable", "history_reason": "messages:unrecognised_shape"})
+ })
+}
+
+// TestWhatACallIsComparedWith holds the table's precedence and the calls a
+// predecessor skips or stops at.
+func TestWhatACallIsComparedWith(t *testing.T) {
+ t.Run("an internal call is not compared and is skipped", func(t *testing.T) {
+ f := newHfix(t)
+ f.input("main", "run")
+ f.turn("main", "c1", a1, sys, q1)
+ f.call("main", "run", "s1", model.ClassInternalCall)
+ f.request("s1", providerbody.FormLangChain, whole(q1))
+ f.turn("main", "c2", a2, sys, q1, a1, q2)
+ attrs, _ := f.assembled(0)
+ check(t, attrs, "call/s1", map[string]any{"history": "not_compared", "history_reason": "internal_call"})
+ check(t, attrs, "call/c2", map[string]any{"history": "kept", "history_from": "call/c1"})
+ })
+ t.Run("a synthetic call is no predecessor", func(t *testing.T) {
+ f := newHfix(t)
+ f.input("main", "run")
+ f.turn("main", "c1", a1, sys, q1)
+ f.call("main", "run", "x1", "synthetic")
+ f.turn("main", "c2", a2, sys, q1, a1, q2)
+ attrs, _ := f.assembled(0)
+ check(t, attrs, "call/c2", map[string]any{"history": "kept", "history_from": "call/c1"})
+ })
+ t.Run("a call of unknown class, as one collected before the class", func(t *testing.T) {
+ f := newHfix(t)
+ f.input("main", "run")
+ f.turn("main", "c1", a1, sys, q1)
+ f.call("main", "run", "u1")
+ f.request("u1", providerbody.FormLangChain, whole(sys, q1, a1, q2))
+ f.turn("main", "c2", a2, sys, q1, a1, q2)
+ attrs, _ := f.assembled(0)
+ check(t, attrs, "call/u1", map[string]any{"history": "unavailable", "history_reason": "class_unknown", "history_messages": 4})
+ // Never skipped: had it been a turn, c2 would be compared with c1.
+ check(t, attrs, "call/c2", map[string]any{"history": "unavailable", "history_reason": "earlier_class_unknown"})
+ })
+ t.Run("a predecessor with no request, or none written with the fields", func(t *testing.T) {
+ f := newHfix(t)
+ f.input("main", "run")
+ f.call("main", "run", "n1", model.ClassTurnCall)
+ f.turn("main", "c1", a1, sys, q1)
+ f.call("main", "run", "o1", model.ClassTurnCall)
+ f.request("o1", "", nil) // landed before asz wrote any history field
+ f.turn("main", "c2", a2, sys, q1, a1, q2)
+ f.turn("main", "c3", none, sys, q1, a1, q2, a2, q3)
+ attrs, _ := f.assembled(0)
+ check(t, attrs, "call/c1", map[string]any{"history": "unavailable", "history_reason": "earlier_no_request"})
+ if _, ok := attrs["call/o1"]["history"]; ok {
+ t.Fatal("a request landed before the fields got a history")
+ }
+ // The old call, then a new one that cannot be compared with it, then
+ // one kept against that: the middle one still says how long it is.
+ check(t, attrs, "call/c2", map[string]any{"history": "unavailable", "history_reason": "earlier_messages:absent", "history_messages": 4})
+ check(t, attrs, "call/c3", map[string]any{"history": "kept", "history_from": "call/c2", "history_reply": "back"})
+ })
+ t.Run("a predecessor's list that is unavailable", func(t *testing.T) {
+ f := newHfix(t)
+ f.input("main", "run")
+ f.call("main", "run", "p1", model.ClassTurnCall)
+ f.request("p1", providerbody.FormLangChain, map[string]any{"messages": map[string]any{"unavailable": "plural_prompts"}})
+ f.turn("main", "c1", a1, sys, q1)
+ attrs, _ := f.assembled(0)
+ check(t, attrs, "call/p1", map[string]any{"history": "unavailable", "history_reason": "messages:plural_prompts"})
+ check(t, attrs, "call/c1", map[string]any{"history": "unavailable", "history_reason": "earlier_messages:plural_prompts"})
+ })
+ t.Run("a predecessor of another form", func(t *testing.T) {
+ f := newHfix(t)
+ f.input("main", "run")
+ f.call("main", "run", "k1")
+ f.request("k1", providerbody.FormClaudeCode, whole(q1))
+ f.turn("main", "c1", a1, sys, q1)
+ attrs, _ := f.assembled(0)
+ check(t, attrs, "call/c1", map[string]any{"history": "unavailable", "history_reason": "form_mismatch"})
+ })
+}
+
+// TestTheEarlierReplysStatus: a retry finds no reply, a reply of another form
+// is not compared, a kept history's reply is looked for only after the
+// earlier history, and on a changed one a reply that could be an earlier
+// message is ambiguous.
+func TestTheEarlierReplysStatus(t *testing.T) {
+ t.Run("a retry sends the same request again", func(t *testing.T) {
+ f := newHfix(t)
+ f.input("main", "run")
+ f.turn("main", "c1", a1, sys, q1)
+ f.turn("main", "c2", none, sys, q1)
+ attrs, _ := f.assembled(0)
+ check(t, attrs, "call/c2", map[string]any{"history": "kept", "history_reply": "not_found"})
+ })
+ t.Run("a reply of another form", func(t *testing.T) {
+ f := newHfix(t)
+ f.input("main", "run")
+ f.call("main", "run", "c1", model.ClassTurnCall)
+ f.request("c1", providerbody.FormLangChain, whole(sys, q1))
+ f.response("c1", providerbody.FormClaudeCode, a1)
+ f.turn("main", "c2", none, sys, q1, a1, q2)
+ attrs, _ := f.assembled(0)
+ check(t, attrs, "call/c2", map[string]any{"history": "kept", "history_reply": "unavailable", "history_reply_reason": "form_mismatch"})
+ })
+ t.Run("a kept history's reply is looked for after the earlier one", func(t *testing.T) {
+ f := newHfix(t)
+ f.input("main", "run")
+ f.turn("main", "c1", a1, sys, a1) // the earlier history already holds a message equal to the reply
+ f.turn("main", "c2", none, sys, a1, q2)
+ attrs, _ := f.assembled(0)
+ check(t, attrs, "call/c2", map[string]any{"history": "kept", "history_reply": "not_found"})
+ })
+ t.Run("a reply sent back altered", func(t *testing.T) {
+ f := newHfix(t)
+ f.input("main", "run")
+ f.turn("main", "c1", a1, sys, q1)
+ f.turn("main", "c2", none, sys, q1, a1x, q2)
+ attrs, _ := f.assembled(0)
+ check(t, attrs, "call/c2", map[string]any{"history": "kept", "history_reply": "back_changed"})
+ })
+ t.Run("a changed history's reply equal to an earlier message", func(t *testing.T) {
+ f := newHfix(t)
+ f.input("main", "run")
+ f.turn("main", "c1", a1, sys, q1, a1)
+ f.turn("main", "c2", none, sys, a1, q2)
+ attrs, _ := f.assembled(0)
+ check(t, attrs, "call/c2", map[string]any{"history": "changed", "history_reply": "unavailable", "history_reply_reason": "ambiguous"})
+ check(t, attrs, "history/c2", map[string]any{"reply": "unavailable", "reply_reason": "ambiguous"})
+ })
+}
+
+// TestAHistoryOverTheBoundIsNotCompared: a whole list over the entry bound,
+// and a list against a base whose whole list is over it, which keeps less.
+func TestAHistoryOverTheBoundIsNotCompared(t *testing.T) {
+ defer func(entries, links int) { maxHistoryEntries, maxHistoryLinks = entries, links }(maxHistoryEntries, maxHistoryLinks)
+ maxHistoryEntries = 3
+ f := newHfix(t)
+ f.input("main", "run")
+ f.turn("main", "c1", a1, sys, q1, a1, q2) // four entries, one over
+ f.call("main", "run", "c2", model.ClassTurnCall)
+ f.request("c2", providerbody.FormLangChain, map[string]any{"messages": map[string]any{"base": "c1:request", "keep": 1, "add": []any{q3}}})
+ f.response("c2", providerbody.FormLangChain, none)
+ attrs, _ := f.assembled(0)
+ check(t, attrs, "call/c1", map[string]any{"history": "unavailable", "history_reason": "messages:too_large"})
+ // c2 keeps one entry of c1's list and adds one: its own history fits.
+ check(t, attrs, "call/c2", map[string]any{"history_messages": 2, "history": "unavailable", "history_reason": "earlier_messages:too_large"})
+}
+
+// A Claude Code turn: entries carry no id, and a request may start or
+// continue a provider thread.
+func (f *hfix) ccTurn(stream, call string, thread map[string]any, system string, reply map[string]any, entries ...map[string]any) {
+ f.call(stream, "run", call)
+ fields := whole(entries...)
+ if thread != nil {
+ fields["thread"] = thread
+ }
+ if system != "" {
+ fields["system"] = system
+ }
+ f.request(call, providerbody.FormClaudeCode, fields)
+ if reply != nil {
+ f.response(call, providerbody.FormClaudeCode, reply)
+ }
+}
+
+var (
+ u1, r1, u2, r2, u3 = e(0, 0x101), e(0, 0x102), e(0, 0x103), e(0, 0x104), e(0, 0x105)
+ create = map[string]any{"type": "create"}
+ sysA = "000000000000000a"
+)
+
+func cont(call string) map[string]any { return map[string]any{"type": "continue", "previous": call} }
+
+// TestAThreadsHistoryIsBuiltFromWhatItContinues covers the thread cases: a
+// continue from the call before, a resumed create, a rewind to an older
+// call, a kept continue of an older call, and the threads that cannot be
+// followed.
+func TestAThreadsHistoryIsBuiltFromWhatItContinues(t *testing.T) {
+ t.Run("continued, then resumed", func(t *testing.T) {
+ f := newHfix(t)
+ f.input("main", "run")
+ f.ccTurn("main", "c1", create, sysA, r1, u1)
+ f.ccTurn("main", "c2", cont("c1"), "", r2, u2)
+ f.ccTurn("main", "c3", create, sysA, none, u1, r1, u2, r2, u3)
+ attrs, _ := f.assembled(0)
+ check(t, attrs, "call/c2", map[string]any{"history": "kept", "history_messages": 3, "history_reply": "back", "system_changed": "no"})
+ check(t, attrs, "call/c3", map[string]any{"history": "kept", "history_messages": 5, "history_reply": "back", "system_changed": "no"})
+ })
+ t.Run("a rewind to an older call", func(t *testing.T) {
+ f := newHfix(t)
+ f.input("main", "run")
+ f.ccTurn("main", "c1", create, sysA, r1, u1)
+ f.ccTurn("main", "c2", cont("c1"), "", r2, u2)
+ f.ccTurn("main", "c3", cont("c1"), "", none, u3) // c2's message and reply are gone
+ attrs, _ := f.assembled(0)
+ check(t, attrs, "call/c3", map[string]any{"history": "changed", "history_from": "call/c2", "history_messages": 3})
+ check(t, attrs, "history/c3", map[string]any{"kept": 2, "dropped": 1, "added": 1})
+ })
+ t.Run("a kept continue of a call older than its predecessor", func(t *testing.T) {
+ f := newHfix(t)
+ f.input("main", "run")
+ f.ccTurn("main", "c1", create, sysA, r1, u1)
+ f.ccTurn("main", "c2", cont("c1"), "", r2, u2)
+ f.ccTurn("main", "c3", cont("c1"), "", none, u2, r2, u3) // sends c2's again
+ attrs, _ := f.assembled(0)
+ check(t, attrs, "call/c3", map[string]any{"history": "kept", "history_messages": 5, "history_reply": "back"})
+ })
+ t.Run("threads that cannot be followed", func(t *testing.T) {
+ f := newHfix(t)
+ f.input("main", "run")
+ f.ccTurn("main", "c1", create, sysA, nil, u1) // no response: no reply
+ f.ccTurn("main", "c2", cont("c1"), "", r2, u2)
+ f.ccTurn("main", "c3", cont("gone"), "", r2, u3)
+ f.input("side", "srun")
+ f.ccTurn("side", "s1", cont("c1"), "", nil, u1) // another stream's call
+ f.ccTurn("main", "c4", map[string]any{"type": "fork"}, "", nil, u1)
+ attrs, _ := f.assembled(0)
+ check(t, attrs, "call/c2", map[string]any{"history": "unavailable", "history_reason": "thread_reply:no_reply"})
+ check(t, attrs, "call/c3", map[string]any{"history": "unavailable", "history_reason": "thread_unresolved"})
+ check(t, attrs, "call/s1", map[string]any{"history": "unavailable", "history_reason": "thread_unresolved"})
+ check(t, attrs, "call/c4", map[string]any{"history": "unavailable", "history_reason": "messages:unrecognised_shape"})
+ })
+ t.Run("an unknown system goes on through the thread", func(t *testing.T) {
+ f := newHfix(t)
+ f.input("main", "run")
+ f.ccTurn("main", "c1", create, sysA, r1, u1)
+ f.ccTurn("main", "c2", cont("c1"), "000000000000000b", r2, u2) // a system block of its own
+ f.ccTurn("main", "c3", cont("c2"), "", r1, u3)
+ f.ccTurn("main", "c4", create, sysA, none, u1, r1, u2, r2, u3, r1)
+ attrs, _ := f.assembled(0)
+ check(t, attrs, "call/c2", map[string]any{"history": "kept", "system_changed": "unavailable"})
+ check(t, attrs, "call/c3", map[string]any{"history": "kept", "system_changed": "unavailable"})
+ check(t, attrs, "call/c4", map[string]any{"history": "kept", "system_changed": "unavailable"})
+ })
+}
+
+// TestALimitOneWalkReachesIsNotKeptAgainstAnotherCall. With the link bound
+// at one, the chain's tail is too large; a call nearer its start is not, and
+// stays measurable though the tail, landed first, was measured first.
+func TestALimitOneWalkReachesIsNotKeptAgainstAnotherCall(t *testing.T) {
+ defer func(links int) { maxHistoryLinks = links }(maxHistoryLinks)
+ maxHistoryLinks = 1
+ f := newHfix(t)
+ f.input("main", "run")
+ f.ccTurn("main", "c3", cont("c2"), "", none, u3)
+ f.ccTurn("main", "c1", create, sysA, r1, u1)
+ f.ccTurn("main", "c2", cont("c1"), "", r2, u2)
+ attrs, _ := f.assembled(0)
+ check(t, attrs, "call/c3", map[string]any{"history": "unavailable", "history_reason": "messages:too_large"})
+ check(t, attrs, "call/c2", map[string]any{"history": "kept", "history_messages": 3})
+}
+
+// TestASystemWalkAtTheLinkBoundReachesItsStart. A thread of exactly as many
+// links as the bound is measured, so its system is read at its start, though
+// its last call lands first and is walked with nothing known; and that walk
+// leaves nothing behind for the calls it passed.
+func TestASystemWalkAtTheLinkBoundReachesItsStart(t *testing.T) {
+ defer func(links int) { maxHistoryLinks = links }(maxHistoryLinks)
+ maxHistoryLinks = 2
+ f := newHfix(t)
+ f.input("main", "run")
+ f.ccTurn("main", "c3", cont("c2"), "", none, u3)
+ f.ccTurn("main", "c1", create, sysA, r1, u1)
+ f.ccTurn("main", "c2", cont("c1"), "", r2, u2)
+ attrs, _ := f.assembled(0)
+ check(t, attrs, "call/c1", map[string]any{"history": "changed", "history_from": "call/c3", "system_changed": "no"})
+ check(t, attrs, "call/c2", map[string]any{"history": "kept", "history_from": "call/c1", "system_changed": "no"})
+}
+
+// TestAListKeepingTheFrontOfALongerOneCopiesOnlyThatFront. A list over a
+// base that holds many entries takes the front it keeps without copying the
+// rest.
+func TestAListKeepingTheFrontOfALongerOneCopiesOnlyThatFront(t *testing.T) {
+ const many = 50000
+ f := newHfix(t)
+ f.input("main", "run")
+ f.turn("main", "c1", none, sys)
+ add := make([]any, many)
+ for i := range add {
+ add[i] = e(uint64(100+i), uint64(100+i))
+ }
+ f.call("main", "run", "c2", model.ClassTurnCall)
+ f.request("c2", providerbody.FormLangChain, map[string]any{"messages": map[string]any{"base": "c1:request", "keep": 1, "add": add}})
+ f.call("main", "run", "c3", model.ClassTurnCall)
+ f.request("c3", providerbody.FormLangChain, map[string]any{"messages": map[string]any{"base": "c2:request", "keep": 2, "add": []any{q1}}})
+ b := &builder{ix: f.ix}
+ h := &historian{b: b, bodyOf: map[uint32]int{}, lists: map[int]ownList{}}
+ for i := range f.ix.Bodies {
+ h.bodyOf[f.ix.Entries[f.ix.Bodies[i].Entry].Record] = i
+ }
+ row := len(f.ix.Bodies) - 1
+ var before, after runtime.MemStats
+ runtime.ReadMemStats(&before)
+ list, why := h.own(row)
+ runtime.ReadMemStats(&after)
+ if why != "" || len(list) != 3 || list[1].ID != 100 || list[2] != f.ix.Messages[len(f.ix.Messages)-1] {
+ t.Fatalf("c3's list is %v (%q)", list, why)
+ }
+ if grew := after.TotalAlloc - before.TotalAlloc; grew > 64<<10 {
+ t.Fatalf("reading three entries allocated %d bytes", grew)
+ }
+}
+
+// TestLateEvidenceRevisesTheComparison: a response that lands late gives the
+// reply, and the history a thread builds from it; a late record of a call's
+// class takes the call out of its stream's history, and the card it drew
+// goes.
+func TestLateEvidenceRevisesTheComparison(t *testing.T) {
+ f := newHfix(t)
+ f.input("main", "run")
+ f.call("main", "run", "c1")
+ f.request("c1", providerbody.FormClaudeCode, map[string]any{"thread": create, "messages": map[string]any{"entries": []any{u1}}})
+ f.ccTurn("main", "c2", cont("c1"), "", nil, u2)
+ before := uint64(f.seq)
+ f.response("c1", providerbody.FormClaudeCode, r1)
+ early, _ := f.assembled(before)
+ check(t, early, "call/c2", map[string]any{"history": "unavailable", "history_reason": "thread_reply:no_reply"})
+ late, _ := f.assembled(0)
+ check(t, late, "call/c2", map[string]any{"history": "kept", "history_reply": "back"})
+
+ g := newHfix(t)
+ g.input("main", "run")
+ g.turn("main", "c1", a1, sys, q1)
+ g.turn("main", "x1", none, q3) // read as a turn, so c2 is compared with it
+ g.turn("main", "c2", none, sys, q1, a1, q2)
+ cut := uint64(g.seq)
+ // A later arrival of x1, under its own record id, says it was the framework's.
+ g.land(transcriptOf("main"), &sessiondata.Record{ID: "x1-late", From: sessiondata.FromAgent, Call: "x1", Run: "run",
+ Flags: []string{"finished", model.ClassInternalCall}})
+ early, _ = g.assembled(cut)
+ check(t, early, "call/c2", map[string]any{"history": "changed", "history_from": "call/x1"})
+ if _, ok := early["history/c2"]; !ok {
+ t.Fatal("no card before the late record")
+ }
+ late, _ = g.assembled(0)
+ check(t, late, "call/c2", map[string]any{"history": "kept", "history_from": "call/c1"})
+ if _, ok := late["history/c2"]; ok {
+ t.Fatal("the card stayed after its pair was no longer changed")
+ }
+}
diff --git a/internal/assemble/history_test.go b/internal/assemble/history_test.go
new file mode 100644
index 0000000..e769440
--- /dev/null
+++ b/internal/assemble/history_test.go
@@ -0,0 +1,110 @@
+// 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 assemble
+
+import (
+ "fmt"
+ "testing"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/index"
+)
+
+// m is an entry with an id when id is not 0, else one by its digest alone.
+func m(id, digest uint64) index.Message {
+ return index.Message{HasID: id != 0, ID: id, Digest: digest}
+}
+
+// TestTheCountsExplainAChangedHistory holds the counts and the order test on
+// the cases the design names, repeats among them.
+func TestTheCountsExplainAChangedHistory(t *testing.T) {
+ a, b, c, d := m(1, 10), m(2, 20), m(3, 30), m(4, 40)
+ cont := m(0, 99) // "continue", which a person types more than once
+ x := m(0, 50)
+ for name, tc := range map[string]struct {
+ P, N []index.Message
+ kept, changed, dropped, added int
+ order string
+ }{
+ "a trimmed front": {[]index.Message{a, b, c}, []index.Message{c, d}, 1, 0, 2, 1, "no"},
+ "the last message replaced": {[]index.Message{a, b, c}, []index.Message{a, b, d}, 2, 0, 1, 1, "no"},
+ "a shortening": {[]index.Message{a, b, c}, []index.Message{a, b}, 2, 0, 1, 0, "no"},
+ "a reordering": {[]index.Message{a, b}, []index.Message{b, a}, 2, 0, 0, 0, "yes"},
+ "an edit under one id": {[]index.Message{a, b}, []index.Message{a, m(2, 21)}, 1, 1, 0, 0, "no"},
+ "a moved and edited message": {[]index.Message{a, b}, []index.Message{b, m(1, 11)}, 1, 1, 0, 0, "yes"},
+ "repeated identical messages, the first dropped": {[]index.Message{cont, x, cont}, []index.Message{x, cont}, 2, 0, 1, 0, "no"},
+ "a repeated id reordered": {[]index.Message{m(1, 10), m(1, 11)}, []index.Message{m(1, 11), m(1, 10)}, 2, 0, 0, 0, "unavailable"},
+ "no id: an edit is a drop and an add": {[]index.Message{m(0, 1), m(0, 2)}, []index.Message{m(0, 1), m(0, 3)}, 1, 0, 1, 1, "no"},
+ } {
+ if isPrefix(tc.P, tc.N) {
+ t.Errorf("%s: read as kept", name)
+ }
+ got := counts(tc.P, tc.N)
+ want := map[string]any{"kept": tc.kept, "changed": tc.changed, "dropped": tc.dropped, "added": tc.added, "order_changed": tc.order}
+ for k, v := range want {
+ if fmt.Sprint(got[k]) != fmt.Sprint(v) {
+ t.Errorf("%s: %s is %v, want %v (%v)", name, k, got[k], v, got)
+ }
+ }
+ if tc.order == "unavailable" && got["order_changed_reason"] != "repeated_id" {
+ t.Errorf("%s: order unavailable for %v, want repeated_id", name, got["order_changed_reason"])
+ }
+ }
+}
+
+// TestTheOrderTestHasABound. Past the pairs it may visit, the order is not
+// decided rather than decided slowly.
+func TestTheOrderTestHasABound(t *testing.T) {
+ same := m(0, 7)
+ P := make([]index.Message, 2049)
+ N := make([]index.Message, 2049)
+ for i := range P {
+ P[i], N[i] = same, same
+ }
+ if _, why := orderChanged(P, N, len(P)); why != "too_large" {
+ t.Fatalf("%d pairs: %q, want too_large", len(P)*len(N), why)
+ }
+ if order, why := orderChanged(P[:2048], N[:2048], 2048); why != "" || order {
+ t.Fatalf("%d pairs: %v %q, want decided and unchanged", 2048*2048, order, why)
+ }
+}
+
+// TestTheEarlierReplyIsFoundWhereItCanBe.
+func TestTheEarlierReplyIsFoundWhereItCanBe(t *testing.T) {
+ r := m(9, 90)
+ altered := m(9, 91)
+ other := m(0, 5)
+ for want, in := range map[string][]index.Message{
+ replyBack: {other, r},
+ replyBackChanged: {altered, other},
+ replyNotFound: {other},
+ } {
+ if got := find(r, in); got != want {
+ t.Errorf("%v: %s, want %s", in, got, want)
+ }
+ }
+ // On a changed history a reply equal to an earlier message, or sharing
+ // its id, could be either, and none is claimed.
+ h := &historian{}
+ for name, P := range map[string][]index.Message{"equal": {r}, "same id": {m(9, 1)}} {
+ out := &callHistory{attrs: map[string]any{}}
+ h.replyIn(out, r, "", P, []index.Message{r})
+ if out.attrs["history_reply"] != "unavailable" || out.attrs["history_reply_reason"] != "ambiguous" {
+ t.Errorf("%s: %v", name, out.attrs)
+ }
+ }
+}
diff --git a/internal/assemble/provider.go b/internal/assemble/provider.go
index 0c4b1d1..2ccca7f 100644
--- a/internal/assemble/provider.go
+++ b/internal/assemble/provider.go
@@ -161,13 +161,16 @@ func (b *builder) joinProviderBodies() {
}
b.bodiesOfCall = map[string][]sessionflow.ProviderBody{}
+ b.requestRow, b.responseRow = map[string]int{}, map[string]int{}
for _, k := range calls {
var joined []sessionflow.ProviderBody
if i, ok := requestFor[k.node]; ok {
joined = append(joined, bodies[i].published())
+ b.requestRow[k.node] = bodies[i].row
}
if i, ok := responseOf[k.node]; ok {
joined = append(joined, bodies[i].published())
+ b.responseRow[k.node] = bodies[i].row
}
if len(joined) > 0 {
b.bodiesOfCall[k.node] = joined
@@ -177,6 +180,8 @@ func (b *builder) joinProviderBodies() {
// body is one provider body this round may join.
type body struct {
+ // row is the body's place in the index's Bodies.
+ row int
role index.BodyRole
ref sessionflow.Ref
prompt, previous uint32
@@ -206,7 +211,7 @@ func (b *builder) bodiesInWindow() []body {
}
seen[e.Record] = true
out = append(out, body{
- role: y.Role, ref: sessionflow.Ref{Seq: uint64(e.Seq), Row: uint64(e.Row)},
+ row: i, role: y.Role, ref: sessionflow.Ref{Seq: uint64(e.Seq), Row: uint64(e.Row)},
prompt: e.Run, previous: y.Previous, request: y.Request, msg: e.Call,
})
}
diff --git a/internal/scenario/expect/expect.go b/internal/scenario/expect/expect.go
index ae96add..f9cdfbb 100644
--- a/internal/scenario/expect/expect.go
+++ b/internal/scenario/expect/expect.go
@@ -142,7 +142,10 @@ type Checkpoint struct {
Written *bool `yaml:"written"`
// Kinds counts nodes by kind, TalksOn talks by stream, Relations by type.
// A stream may be named by its scenario name.
- Kinds map[string]int `yaml:"kinds"`
+ Kinds map[string]int `yaml:"kinds"`
+ // History counts the model calls by what their history attribute says,
+ // none for a call that carries none.
+ History map[string]int `yaml:"history"`
TalksOn map[string]int `yaml:"talks_on"`
RunsIn map[string]int `yaml:"runs_in"`
Relations map[string]int `yaml:"relations"`
@@ -227,7 +230,7 @@ func (l Lose) Resolve(files []storage.LandedFile, stream string) (*storage.Lande
// Empty reports a checkpoint with nothing to check.
func (cp *Checkpoint) Empty() bool {
- return cp.Rounds == nil && cp.Written == nil && len(cp.Kinds) == 0 && len(cp.TalksOn) == 0 && len(cp.RunsIn) == 0 &&
+ return cp.Rounds == nil && cp.Written == nil && len(cp.Kinds) == 0 && len(cp.History) == 0 && len(cp.TalksOn) == 0 && len(cp.RunsIn) == 0 &&
len(cp.Relations) == 0 && cp.Unresolved == nil && len(cp.UnresolvedKinds) == 0 && len(cp.Nodes) == 0 && cp.Verify == nil &&
cp.Session == nil && cp.View == nil && cp.Delta == nil
}
@@ -392,12 +395,27 @@ func Evaluate(root string, cp *Checkpoint, ctx *Context) ([]string, error) {
bad("delta: round %d wrote %d nodes against a fold of %d, starting at seq %d; it is not a delta", ctx.Round.Number, ctx.Round.Nodes, len(v.Nodes), ctx.Round.FromSeq)
}
}
- kinds, talksOn := map[string]int{}, map[string]int{}
+ kinds, talksOn, history := map[string]int{}, map[string]int{}, map[string]int{}
for _, n := range v.Nodes {
kinds[n.Kind]++
if n.Kind == model.KindTalk {
talksOn[n.Stream]++
}
+ if n.Kind == model.KindLLMCall {
+ var a struct {
+ History string `json:"history"`
+ }
+ _ = json.Unmarshal(n.Attrs, &a)
+ if a.History == "" {
+ a.History = "none"
+ }
+ history[a.History]++
+ }
+ }
+ for value, want := range sorted(cp.History) {
+ if history[value] != want {
+ bad("history: %s is %d calls, want %d", value, history[value], want)
+ }
}
for kind, want := range sorted(cp.Kinds) {
if kinds[kind] != want {
@@ -809,6 +827,12 @@ type Summary struct {
Streams []string
// Records is how many records each stream holds, by stream.
Records map[string]int
+ // History is what each model call's history was and each card says, one
+ // line a node, named by the call or card id, which the evidence gives
+ // and no landed position moves. A card's records are named by their
+ // record ids, so a comparison of two roots cut differently holds every
+ // value.
+ History map[string]int
}
// Summarize reads a fold into a Summary.
@@ -817,7 +841,15 @@ func Summarize(root, session string) (*Summary, error) {
if err != nil {
return nil, err
}
- s := &Summary{Kinds: map[string]int{}, Relations: map[string]int{}, TalksOn: map[string]int{}, Unresolved: map[string]int{}, Records: map[string]int{}}
+ s := &Summary{Kinds: map[string]int{}, Relations: map[string]int{}, TalksOn: map[string]int{}, Unresolved: map[string]int{},
+ Records: map[string]int{}, History: map[string]int{}}
+ histories, err := historyLines(root, session, v)
+ if err != nil {
+ return nil, err
+ }
+ for _, l := range histories {
+ s.History[l]++
+ }
for _, n := range v.Nodes {
s.Kinds[n.Kind]++
if n.Kind == model.KindTalk {
@@ -842,6 +874,109 @@ func Summarize(root, session string) (*Summary, error) {
return s, nil
}
+// historyLines is one line for each model call that says what its history
+// was, with every history attribute, and one for each card, with its
+// attributes, the records it names and stands on, and whether it stands
+// under its call's parent.
+func historyLines(root, session string, v *sessionflow.View) ([]string, error) {
+ var ids map[[2]uint64]string
+ recordOf := func(r sessionflow.Ref) (string, error) {
+ if ids == nil {
+ ids = map[[2]uint64]string{}
+ files, err := storage.LandedFiles(storage.NewZone(root), session)
+ if err != nil {
+ return "", err
+ }
+ for _, lf := range files {
+ f, err := sessiondata.Open(lf.Path)
+ if err != nil {
+ return "", err
+ }
+ _, recs, err := sessiondata.All(f)
+ f.Close()
+ if err != nil {
+ return "", err
+ }
+ for row, rec := range recs {
+ ids[[2]uint64{lf.Seq, uint64(row + 1)}] = rec.ID
+ }
+ }
+ }
+ id, ok := ids[[2]uint64{r.Seq, r.Row}]
+ if !ok || id == "" {
+ return "", fmt.Errorf("expect: a card names %d/%d, where no record with an id landed", r.Seq, r.Row)
+ }
+ return id, nil
+ }
+ var out []string
+ for id, n := range v.Nodes {
+ var a map[string]any
+ _ = json.Unmarshal(n.Attrs, &a)
+ kept := map[string]any{}
+ switch n.Kind {
+ case model.KindLLMCall:
+ for k, x := range a {
+ if strings.HasPrefix(k, "history") || k == "system_changed" {
+ kept[k] = x
+ }
+ }
+ if len(kept) == 0 {
+ continue
+ }
+ case model.KindHistoryChange:
+ kept = a
+ var refs []string
+ for _, r := range n.Refs {
+ rec, err := recordOf(r)
+ if err != nil {
+ return nil, err
+ }
+ refs = append(refs, rec)
+ }
+ kept["refs"] = refs
+ if n.Ref != nil {
+ rec, err := recordOf(*n.Ref)
+ if err != nil {
+ return nil, err
+ }
+ kept["ref"] = rec
+ }
+ if call, ok := v.Nodes[fmt.Sprint(a["to_call"])]; ok {
+ kept["under_its_call's_parent"] = call.Parent == n.Parent
+ }
+ default:
+ continue
+ }
+ line, err := json.Marshal(kept) // keys sorted
+ if err != nil {
+ return nil, err
+ }
+ out = append(out, id+" "+string(line))
+ }
+ return out, nil
+}
+
+// Without is the summary with no nodes of the kinds named.
+func (s *Summary) Without(kinds ...string) *Summary {
+ out := *s
+ out.Kinds = map[string]int{}
+ for k, n := range s.Kinds {
+ if !slices.Contains(kinds, k) {
+ out.Kinds[k] = n
+ }
+ }
+ return &out
+}
+
+// WithoutHistory is the summary with no history lines, for comparing a
+// build with provider bodies against one without, where only the first
+// can say what a call's history was.
+func (s *Summary) WithoutHistory() *Summary {
+ out := *s
+ out.History = map[string]int{}
+ return &out
+}
+
// Compare says how two summaries differ.
func Compare(a, b *Summary) []string {
var out []string
@@ -865,6 +1000,7 @@ func Compare(a, b *Summary) []string {
}
}
diff("kinds", a.Kinds, b.Kinds)
+ diff("history", a.History, b.History)
diff("relations", a.Relations, b.Relations)
diff("talks on", a.TalksOn, b.TalksOn)
diff("unresolved", a.Unresolved, b.Unresolved)
@@ -1734,12 +1870,15 @@ func ProviderBodiesRebuild(root, session string) ([]string, int, error) {
// landing between them shift, so a node with a reference is named by the
// record it stands on: its kind, its stream, its file's source and the
// record's line in that source, and the part.
-func SameIdentity(rootA, sessionA, rootB, sessionB string) ([]string, error) {
- a, err := foldIdentity(rootA, sessionA)
+//
+// Nodes of the kinds named in without are left out of both: what the evidence
+// the second build lacks may add and nothing else.
+func SameIdentity(rootA, sessionA, rootB, sessionB string, without ...string) ([]string, error) {
+ a, err := foldIdentity(rootA, sessionA, without)
if err != nil {
return nil, err
}
- b, err := foldIdentity(rootB, sessionB)
+ b, err := foldIdentity(rootB, sessionB, without)
if err != nil {
return nil, err
}
@@ -1770,7 +1909,7 @@ func SameIdentity(rootA, sessionA, rootB, sessionB string) ([]string, error) {
// sequence and row, and a part.
var positionID = regexp.MustCompile(`^[^/]+/[0-9]+/[0-9]+(:[0-9]+)?$`)
-func foldIdentity(root, session string) (map[string]bool, error) {
+func foldIdentity(root, session string, without []string) (map[string]bool, error) {
v, err := parse.View(root, session)
if err != nil {
return nil, err
@@ -1816,6 +1955,9 @@ func foldIdentity(root, session string) (map[string]bool, error) {
}
out := map[string]bool{}
for id, n := range v.Nodes {
+ if slices.Contains(without, n.Kind) {
+ continue
+ }
out["node "+nameOf(id)+" in "+nameOf(n.Parent)] = true
}
for _, r := range v.Relations {
diff --git a/internal/scenario/plan.go b/internal/scenario/plan.go
index 5171b79..033f449 100644
--- a/internal/scenario/plan.go
+++ b/internal/scenario/plan.go
@@ -243,6 +243,11 @@ type Plan struct {
// bodies, and bodies says it writes them. See ProviderBodies.
Provider bool
bodies bool
+ // threads writes the bodies with provider threads, and marks are what a
+ // forget or resume step does to the bodies the stream sends next. See
+ // ProviderBodies.
+ threads bool
+ marks []bodyMark
// at is the base time the plan was resolved against, which Span
// measures from.
at time.Time
@@ -303,7 +308,7 @@ func (sc *Scenario) Plan(opts Options) (*Plan, error) {
if scale == 0 {
scale = 1
}
- p := &Plan{Session: sc.Session, Project: sc.Project, Title: sc.Title, Entrypoint: sc.Entrypoint, Provider: sc.ProviderBodies, bodies: sc.ProviderBodies && !sc.omitBodies, at: at, interval: interval, scale: scale,
+ p := &Plan{Session: sc.Session, Project: sc.Project, Title: sc.Title, Entrypoint: sc.Entrypoint, Provider: sc.ProviderBodies, bodies: sc.ProviderBodies && !sc.omitBodies, threads: sc.ProviderThreads, at: at, interval: interval, scale: scale,
SystemPrompt: sc.SystemPrompt, Tools: sc.Tools}
if p.Project == "" {
p.Project = "-Users-dev-scenario"
@@ -430,6 +435,10 @@ func (b *planner) step(l *lane, s *Step, id string) error {
l.last = summary.ID
case s.Replay > 0:
b.replay(s.Replay)
+ case s.Forget > 0 || s.Resume:
+ // Nothing a transcript records: it changes only what the stream's
+ // next request sends.
+ b.p.marks = append(b.p.marks, bodyMark{at: len(b.p.Events), stream: l.stream, forget: s.Forget, resume: s.Resume})
case s.System != nil:
at := b.tick(l, s.After)
e := Event{Kind: EvSystem, Stream: l.stream, Batch: l.batch, At: at, ID: id + "-sys-" + s.System.Subtype, Parent: l.last,
diff --git a/internal/scenario/provider.go b/internal/scenario/provider.go
index 5c2b5bc..97e5211 100644
--- a/internal/scenario/provider.go
+++ b/internal/scenario/provider.go
@@ -44,7 +44,8 @@ type ProviderBody struct {
// plan holds, in the order Claude Code writes them: a request when the call
// is built, and its response when the last fragment has arrived.
//
-// They follow what Claude Code 2.1.260 was measured to write. Every call
+// They follow what Claude Code 2.1.260 was measured to write, and with
+// provider_threads what 2.1.302 writes (see request). Every call of 2.1.260
// sends its stream's whole message list again. The newest message carries
// the cache marker, and a one-block text message becomes a plain string once
// the marker has left it. The first system block is a billing header naming
@@ -78,7 +79,12 @@ func (p *Plan) ProviderBodies() []ProviderBody {
}
var out []ProviderBody
done := map[string]bool{}
+ marks := p.marks
for i := range p.Events {
+ for len(marks) > 0 && marks[0].at == i {
+ chainOf(marks[0].stream).mark(marks[0])
+ marks = marks[1:]
+ }
e := &p.Events[i]
if e.Lost || lost[e.Stream] || e.Replayed {
continue
@@ -90,7 +96,7 @@ func (p *Plan) ProviderBodies() []ProviderBody {
c.user(e.Text)
case EvSummary:
c.prompt = e.Run
- c.msgs = nil
+ c.msgs, c.open = nil, false
c.user(e.Text)
case EvResult:
c.push(map[string]any{"role": "user", "content": []any{
@@ -105,7 +111,7 @@ func (p *Plan) ProviderBodies() []ProviderBody {
out = append(out, ProviderBody{
Name: requestName(p.Session, e.Call),
At: e.At.Add(-time.Millisecond),
- Bytes: c.request(p.Session),
+ Bytes: c.request(p.Session, p.threads),
})
}
c.blocks = append(c.blocks, fragmentBlock(e))
@@ -123,6 +129,7 @@ func (p *Plan) ProviderBodies() []ProviderBody {
})
c.push(map[string]any{"role": "assistant", "content": c.blocks})
c.prev, c.call, c.blocks = e.Req, "", nil
+ c.answered, c.held = e.Call, len(c.msgs)
}
}
}
@@ -157,6 +164,31 @@ type bodyChain struct {
// leaves the stand-in.
system string
tools []ToolDef
+ // open says the stream's provider thread may be continued: answered is
+ // the call whose response it goes on from, and held how many messages
+ // the thread holds, that response included.
+ open bool
+ answered string
+ held int
+}
+
+// bodyMark is what a forget or resume step does to the bodies a stream sends
+// next: at is the event it comes before.
+type bodyMark struct {
+ at int
+ stream string
+ forget int
+ resume bool
+}
+
+// mark applies a forget or a resume. Either way the client cannot go on with
+// the thread the provider holds, so the next request starts a new one.
+func (c *bodyChain) mark(m bodyMark) {
+ if m.forget > 0 && len(c.msgs) > 1 {
+ drop := min(m.forget, len(c.msgs)-1)
+ c.msgs = append(c.msgs[:1:1], c.msgs[1+drop:]...)
+ }
+ c.open = false
}
// user adds a prompt. The first message of a chain carries the injected
@@ -207,8 +239,11 @@ func (c *bodyChain) push(m map[string]any) {
c.msgs = append(c.msgs, m)
}
-// request renders the body of the call about to be made.
-func (c *bodyChain) request(session string) []byte {
+// request renders the body of the call about to be made. With threads, a
+// request that can continue the stream's thread carries only what is new,
+// and of the system prompt only the billing header, and no tools: the
+// provider holds the rest.
+func (c *bodyChain) request(session string, threads bool) []byte {
header := "x-anthropic-billing-header: cc_version=2.1.260; cc_entrypoint=cli; cch=00000;"
if c.prev != "" {
header += " cc_prev_req=" + c.prev + ";"
@@ -251,6 +286,33 @@ func (c *bodyChain) request(session string) []byte {
}
}
user, _ := json.Marshal(map[string]string{"device_id": "scenario", "account_uuid": "scenario", "session_id": session})
+ if threads {
+ type thread struct {
+ Type string `json:"type"`
+ Previous string `json:"previous_message_id,omitempty"`
+ }
+ type threaded struct {
+ Model string `json:"model"`
+ Messages []map[string]any `json:"messages"`
+ System []any `json:"system"`
+ // A request that starts a thread sends its tools, an empty list
+ // included; one that continues sends none.
+ Tools *[]any `json:"tools,omitempty"`
+ Metadata map[string]any `json:"metadata"`
+ MaxTokens int `json:"max_tokens"`
+ Thread thread `json:"thread"`
+ }
+ b := threaded{Model: ccModel, Metadata: map[string]any{"user_id": string(user)}, MaxTokens: 64000,
+ System: []any{map[string]any{"type": "text", "text": header}}}
+ if c.open && c.held <= len(c.msgs) {
+ b.Messages, b.Thread = c.msgs[c.held:], thread{Type: "continue", Previous: c.answered}
+ } else {
+ b.Messages, b.Tools, b.Thread = c.msgs, &tools, thread{Type: "create"}
+ b.System = append(b.System, map[string]any{"type": "text", "text": system, "cache_control": map[string]any{"type": "ephemeral"}})
+ c.open = true
+ }
+ return encodeBody(b)
+ }
type body struct {
Model string `json:"model"`
Messages []map[string]any `json:"messages"`
diff --git a/internal/scenario/provider_test.go b/internal/scenario/provider_test.go
index a15a5a1..a7e500a 100644
--- a/internal/scenario/provider_test.go
+++ b/internal/scenario/provider_test.go
@@ -20,6 +20,7 @@ package scenario
import (
"bytes"
"encoding/json"
+ "fmt"
"os"
"path/filepath"
"strings"
@@ -366,3 +367,59 @@ func TestHalfADeclarationIsRefused(t *testing.T) {
}
}
}
+
+// TestThreadedBodiesAreWrittenAsClaudeCode2_1_302WritesThem reads back the
+// bodies of the threads scenario: a process's first request starts a thread
+// with the whole list, the system prompt and the tools; every later request
+// of the process continues it from the response before, with only what is
+// new, the billing header alone and no tools; a resume starts a new thread;
+// and the searcher's thread is its own.
+func TestThreadedBodiesAreWrittenAsClaudeCode2_1_302WritesThem(t *testing.T) {
+ sc, err := Load(filepath.Join("..", "..", "tests", "scenarios", "history-threads.yaml"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ p, err := sc.Plan(Options{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ requests, _ := bodiesOf(t, p)
+ type sent struct {
+ thread, previous string
+ messages, system int
+ tools bool
+ }
+ // A request is the searcher's by its own system prompt when it starts a
+ // thread, and by the call it continues when it does not.
+ got := map[string][]sent{}
+ for _, req := range requests {
+ th, _ := req["thread"].(map[string]any)
+ prev, _ := th["previous_message_id"].(string)
+ system := req["system"].([]any)
+ stream := "main"
+ if text, _ := system[len(system)-1].(map[string]any)["text"].(string); strings.Contains(text, "searches a repository") || strings.HasPrefix(prev, "searcher") {
+ stream = "searcher"
+ }
+ _, tools := req["tools"]
+ // A message id carries the session's tag after the call's own id.
+ if i := strings.LastIndex(prev, "-"); i > 0 {
+ prev = prev[:i]
+ }
+ got[stream] = append(got[stream], sent{th["type"].(string), prev, len(req["messages"].([]any)), len(system), tools})
+ }
+ want := map[string][]sent{
+ "main": {
+ {"create", "", 1, 2, true},
+ {"continue", "s2-call", 1, 1, false},
+ {"create", "", 5, 2, true}, // after the resume: the whole list again
+ {"continue", "s6-call", 1, 1, false},
+ },
+ "searcher": {
+ {"create", "", 1, 2, true},
+ {"continue", "searcher-s1-call", 1, 1, false},
+ },
+ }
+ if fmt.Sprint(got) != fmt.Sprint(want) {
+ t.Fatalf("requests\n got %+v\nwant %+v", got, want)
+ }
+}
diff --git a/internal/scenario/run/run.go b/internal/scenario/run/run.go
index 424aab4..3396619 100644
--- a/internal/scenario/run/run.go
+++ b/internal/scenario/run/run.go
@@ -40,6 +40,7 @@ import (
"github.com/apache/skywalking-ai-sessionizer/internal/scenario/expect"
"github.com/apache/skywalking-ai-sessionizer/internal/storage"
"github.com/apache/skywalking-ai-sessionizer/pkg/execution"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/model"
)
// Options settle a check run.
@@ -503,14 +504,21 @@ func changesLeaveTheFold(sc *scenario.Scenario, f scenario.Format, out, session
// providerBodiesLeaveTheFold does the same for provider bodies: a body is
// what a call sent and got back, joined to the call by a view, and no step.
+// The one node bodies add is the card where a call's history changed, which
+// only a call's request can show; the call's own history attributes are
+// attributes, which this does not compare.
func providerBodiesLeaveTheFold(sc *scenario.Scenario, f scenario.Format, out, session string, opts Options, maxDelta, maxRound int64) ([]string, error) {
- return leavesTheFold("provider_bodies_leave_the_fold", "the provider bodies", sc.WithoutProviderBodies(), f, out, session, opts, maxDelta, maxRound)
+ return leavesTheFold("provider_bodies_leave_the_fold", "the provider bodies", sc.WithoutProviderBodies(), f, out, session, opts, maxDelta, maxRound,
+ model.KindHistoryChange)
}
// The plain build is cut at the scenario's budget, as the build it is
// compared with: a record over a smaller budget lands alone, and a node named
// by where its first record landed would differ for that alone.
-func leavesTheFold(property, what string, plainScenario *scenario.Scenario, f scenario.Format, out, session string, opts Options, maxDelta, maxRound int64) ([]string, error) {
+//
+// added names the kinds of node the evidence may add, which the comparison
+// leaves out.
+func leavesTheFold(property, what string, plainScenario *scenario.Scenario, f scenario.Format, out, session string, opts Options, maxDelta, maxRound int64, added ...string) ([]string, error) {
plain := out + "-plain"
defer os.RemoveAll(plain)
built, err := scenario.Build(plainScenario, f, plain, scenario.Options{At: opts.At, Scale: opts.Scale, Interval: opts.Interval, MaxDelta: maxDelta})
@@ -531,11 +539,16 @@ func leavesTheFold(property, what string, plainScenario *scenario.Scenario, f sc
if err != nil {
return nil, err
}
+ with = with.Without(added...)
+ if len(added) > 0 {
+ // Only the build with the evidence can say what a call's history was.
+ with, without = with.WithoutHistory(), without.WithoutHistory()
+ }
var lines []string
for _, d := range append(expect.Compare(with, without), expect.CompareRecords(with, without)...) {
lines = append(lines, property+": the fold differs with and without "+what+": "+d)
}
- same, err := expect.SameIdentity(out, session, plain, built.Session)
+ same, err := expect.SameIdentity(out, session, plain, built.Session, added...)
if err != nil {
return nil, err
}
diff --git a/internal/scenario/scenario.go b/internal/scenario/scenario.go
index 76ddad0..79233b2 100644
--- a/internal/scenario/scenario.go
+++ b/internal/scenario/scenario.go
@@ -66,6 +66,11 @@ type Scenario struct {
// directory. A claude-code build writes them as files; an sd build lands
// them.
ProviderBodies bool `yaml:"provider_bodies"`
+ // ProviderThreads writes the bodies as Claude Code 2.1.302 does: a
+ // process's first request starts a provider thread and holds the whole
+ // message list, and every later one continues it from the response
+ // before and holds only what is new. A resume step starts a new process.
+ ProviderThreads bool `yaml:"provider_threads"`
// SystemPrompt is the system prompt the main stream sends, written as the
// runtime sends it. Empty leaves the stand-in, which is filler sized to
// exercise cutting rather than something to read.
@@ -132,6 +137,14 @@ type Step struct {
Replay int `yaml:"replay"`
// System is a system record of any subtype.
System *System `yaml:"system"`
+ // Forget drops this many of the oldest messages after the first from the
+ // history the stream's next request sends: what an application trimming
+ // its history does. No transcript record shows it; only a provider body
+ // does, and with threads the next request starts a new thread.
+ Forget int `yaml:"forget"`
+ // Resume is a new process taking the session up again, which starts a
+ // new provider thread. It needs provider_threads.
+ Resume bool `yaml:"resume"`
}
// Queued is input that exists only as an attachment. Mode is what the
@@ -438,10 +451,47 @@ func (sc *Scenario) Validate() error {
if sc.ProviderBodies && namesNoCall(sc.Steps) {
return errors.New("provider_bodies: a body is written for each call, and an unidentified call or error names none")
}
+ forget, resume := bodySteps(sc.Steps)
+ switch {
+ case sc.ProviderThreads && !sc.ProviderBodies:
+ return errors.New("provider_threads: threads are written in provider bodies, and the scenario writes none")
+ case forget && !sc.ProviderBodies:
+ return errors.New("forget: only a provider body shows a history the client trimmed, and the scenario writes none")
+ case resume && !sc.ProviderThreads:
+ return errors.New("resume: a new process starts a new provider thread, and the scenario writes no threads")
+ }
seen := map[string]bool{}
return validateSteps(sc.Steps, "steps", seen, true, advertised(sc.Tools))
}
+// bodySteps reports whether a forget or a resume step is anywhere in the
+// scenario, in any stream.
+func bodySteps(steps []Step) (forget, resume bool) {
+ for i := range steps {
+ s := &steps[i]
+ forget, resume = forget || s.Forget > 0, resume || s.Resume
+ var nested [][]Step
+ if c := s.Call; c != nil {
+ if c.Agent != nil {
+ nested = append(nested, c.Agent.Steps)
+ }
+ if c.Skill != nil {
+ nested = append(nested, c.Skill.Steps)
+ }
+ if c.Workflow != nil {
+ for _, child := range c.Workflow.Children {
+ nested = append(nested, child.Steps)
+ }
+ }
+ }
+ for _, n := range nested {
+ f, r := bodySteps(n)
+ forget, resume = forget || f, resume || r
+ }
+ }
+ return forget, resume
+}
+
// namesNoCall reports whether a step, in any stream, writes model output with no
// call id. Provider bodies are written per call and joined by the call's ids, so
// such output has no body, and the two formats would not agree on what is left.
@@ -570,13 +620,16 @@ func validateSteps(steps []Step, where string, seen map[string]bool, main bool,
at := fmt.Sprintf("%s[%d]", where, i)
n := 0
for _, set := range []bool{s.Input != "", s.Queued != nil, s.Inject != nil, s.Call != nil,
- s.Result != nil, s.Error.Text != "", s.Reset != nil, s.Replay > 0, s.System != nil} {
+ s.Result != nil, s.Error.Text != "", s.Reset != nil, s.Replay > 0, s.System != nil, s.Forget != 0, s.Resume} {
if set {
n++
}
}
if n != 1 {
- return fmt.Errorf("%s: a step is exactly one of input, queued, inject, call, result, error, reset, replay, system", at)
+ return fmt.Errorf("%s: a step is exactly one of input, queued, inject, call, result, error, reset, replay, system, forget, resume", at)
+ }
+ if s.Forget < 0 {
+ return fmt.Errorf("%s: forget is how many messages are dropped, never negative", at)
}
if in := s.Inject; in != nil {
// The runtime carries these in two attachment types, and the
diff --git a/internal/view/step.go b/internal/view/step.go
index 56efbe7..38a44c2 100644
--- a/internal/view/step.go
+++ b/internal/view/step.go
@@ -447,7 +447,10 @@ func candidates(rec *sessiondata.Record) [][]byte {
func carriesContent(kind string) bool {
switch kind {
case model.KindLLMCall, model.KindSession, model.KindSegment,
- model.KindStream, model.KindEpoch, model.KindTalk, model.KindRun:
+ model.KindStream, model.KindEpoch, model.KindTalk, model.KindRun,
+ // A card says what two requests' histories hold, from its
+ // attributes; the record it stands at is its call's.
+ model.KindHistoryChange:
return false
}
return true
diff --git a/pkg/model/model.go b/pkg/model/model.go
index 022f6a6..843e4e6 100644
--- a/pkg/model/model.go
+++ b/pkg/model/model.go
@@ -71,6 +71,11 @@ const (
KindEpochBoundary = "epoch.boundary" // an explicit model-context reset
KindEpochSummary = "epoch.summary" // the carried-forward summary produced by that reset
+ // KindHistoryChange marks where a call's history is not the one before it
+ // with more after it. It names no cause: a context reset is an explicit
+ // record, and a changed history is not one. It carries no content.
+ KindHistoryChange = "history.change"
+
// Control family.
KindErrorAPI = "error.api"
KindControlInterrupt = "control.interrupt"
diff --git a/pkg/model/term.go b/pkg/model/term.go
index 1c5dd99..6619b00 100644
--- a/pkg/model/term.go
+++ b/pkg/model/term.go
@@ -119,7 +119,7 @@ func Vocabulary() []string {
KindMessageExternal, KindMessageAssistant, KindMessageSynthetic,
KindContextInjection, KindLLMCall, KindThinking, KindTool,
KindAgentCall, KindAgentLaunchAck, KindAgentOutput, KindRuntimeNotification,
- KindEpochBoundary, KindEpochSummary,
+ KindEpochBoundary, KindEpochSummary, KindHistoryChange,
KindErrorAPI, KindControlInterrupt, KindControlPermission, KindControlCommand,
KindControlNotice, KindTurnDuration,
// relations
diff --git a/pkg/sessionflow/before_test.go b/pkg/sessionflow/before_test.go
index 3fa4ddc..21dc0c6 100644
--- a/pkg/sessionflow/before_test.go
+++ b/pkg/sessionflow/before_test.go
@@ -49,3 +49,28 @@ func TestNodesOnOneRecordHaveOneOrder(t *testing.T) {
}
}
}
+
+// TestACardComesJustBeforeItsCall. A history.change card stands at its call's
+// first fragment. At one position it comes just before the call it names and
+// is ordered among the other nodes there by that call, so two calls on one
+// record keep their order with each card right before its own; ordered by its
+// kind alone, every card would come first.
+func TestACardComesJustBeforeItsCall(t *testing.T) {
+ node := func(id, kind string) *sessionflow.Node {
+ return &sessionflow.Node{Entity: sessionflow.Entity{ID: id}, Kind: kind, Ref: &sessionflow.Ref{Seq: 1, Row: 1}}
+ }
+ callA, callB := node("call/a", "llm.call"), node("call/b", "llm.call")
+ cardB, other := node("history/b", "history.change"), node("input/1/1", "message.external")
+ all := []*sessionflow.Node{callA, callB, cardB, other}
+ want := "call/a history/b call/b input/1/1"
+ for i := range all {
+ in := append(append([]*sessionflow.Node(nil), all[i:]...), all[:i]...)
+ var ids []string
+ for _, n := range sessionflow.InOrder(in) {
+ ids = append(ids, n.ID)
+ }
+ if got := strings.Join(ids, " "); got != want {
+ t.Errorf("%s, want %s", got, want)
+ }
+ }
+}
diff --git a/pkg/sessionflow/fold.go b/pkg/sessionflow/fold.go
index 72b863e..2195d03 100644
--- a/pkg/sessionflow/fold.go
+++ b/pkg/sessionflow/fold.go
@@ -20,7 +20,10 @@ package sessionflow
import (
"fmt"
"sort"
+ "strings"
"sync"
+
+ "github.com/apache/skywalking-ai-sessionizer/pkg/model"
)
// View is the materialised result of folding a chain.
@@ -336,6 +339,16 @@ func (v *View) Roots() []*Node {
return InOrder(out)
}
+// tieKey is what orders a node among nodes at one position: its own id, or
+// for a history.change card, the id of the call it stands before, which
+// shares the card's key (session-flow.md), and true to come first.
+func tieKey(n *Node) (string, bool) {
+ if n.Kind == model.KindHistoryChange {
+ return "call/" + strings.TrimPrefix(n.ID, "history/"), true
+ }
+ return n.ID, false
+}
+
// Before reports whether a occurred before b.
//
// This is the display order the model calls a deterministic projection. It does
@@ -360,6 +373,17 @@ func Before(a, b *Node) bool {
if ap.Block != nil && *ap.Block != *bp.Block {
return *ap.Block < *bp.Block
}
+ // A history.change card stands at its call's first fragment and comes
+ // just before that call, so at one position a card is ordered by the
+ // call it names and before it; the ids alone would put call/ first.
+ ak, ac := tieKey(a)
+ bk, bc := tieKey(b)
+ if ak != bk {
+ return ak < bk
+ }
+ if ac != bc {
+ return ac
+ }
case ap != nil:
return true // a is positioned, b is not
case bp != nil:
diff --git a/tests/chain/history_round_test.go b/tests/chain/history_round_test.go
new file mode 100644
index 0000000..6b1409f
--- /dev/null
+++ b/tests/chain/history_round_test.go
@@ -0,0 +1,395 @@
+// 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 (
+ "encoding/json"
+ "fmt"
+ "io"
+ "os"
+ "path/filepath"
+ "reflect"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/index"
+ "github.com/apache/skywalking-ai-sessionizer/internal/parse"
+ "github.com/apache/skywalking-ai-sessionizer/internal/scenario"
+ "github.com/apache/skywalking-ai-sessionizer/internal/storage"
+ "github.com/apache/skywalking-ai-sessionizer/internal/verify"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/model"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/providerbody"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessionflow"
+)
+
+const historySession = "00000002-0000-4000-8000-000000000002"
+
+// A history root: landed files written by hand, one record each, so a test
+// lands late evidence between two parses.
+type historyRoot struct {
+ t *testing.T
+ z *storage.Zone
+ seq uint64
+}
+
+func (r *historyRoot) land(dir, kind, stream string, rec sessiondata.Record) {
+ r.t.Helper()
+ r.seq++
+ path := filepath.Join(dir, storage.LandedName(kind, storage.Stamp(time.Unix(int64(r.seq), 0)), r.seq))
+ rec.Ord, rec.Sha, rec.Bytes = 1, "a", 1
+ err := storage.WriteAtomic(path, storage.PermLanded, func(w io.Writer) error {
+ sw, err := sessiondata.NewWriter(w, &sessiondata.Header{H: 1, Seq: r.seq, At: "2026-10-18T00:00:00Z",
+ Kind: sessiondata.Kind(kind), Adapter: "test/0", Dialect: "test/1",
+ Src: fmt.Sprintf("src-%d", r.seq), Session: historySession, Stream: stream})
+ if err != nil {
+ return err
+ }
+ if err := sw.Write(&rec); err != nil {
+ return err
+ }
+ return sw.Close()
+ })
+ if err != nil {
+ r.t.Fatal(err)
+ }
+}
+
+func (r *historyRoot) record(rec sessiondata.Record) {
+ r.land(r.z.StreamDir(historySession, storage.StreamMain), string(sessiondata.KindTranscript), storage.StreamMain, rec)
+}
+
+func (r *historyRoot) body(id, call, role string, fields map[string]any) {
+ r.t.Helper()
+ m := map[string]any{"schema": providerbody.Schema, "role": role, "sha256": "d", "bytes": 2, "depth": 0,
+ "segments": []any{map[string]any{"lit": "{}"}}, "history_form": providerbody.FormLangChain}
+ for k, v := range fields {
+ m[k] = v
+ }
+ data, err := json.Marshal(m)
+ if err != nil {
+ r.t.Fatal(err)
+ }
+ r.land(r.z.ProviderDir(historySession), string(sessiondata.KindProviderBody), "",
+ sessiondata.Record{ID: id, Call: call, Parts: []sessiondata.Part{{Kind: sessiondata.PartData, Data: data, State: "available"}}})
+}
+
+func entry(id, digest uint64) map[string]any {
+ return map[string]any{"id": fmt.Sprintf("%016x", id), "digest": fmt.Sprintf("%016x", digest)}
+}
+
+// turn lands a turn call's record and its request; a nil reply lands no
+// response.
+func (r *historyRoot) turn(call string, reply map[string]any, entries ...map[string]any) {
+ r.record(sessiondata.Record{ID: call + "-r", From: sessiondata.FromAgent, Call: call, Run: "run",
+ Flags: []string{"finished", model.ClassTurnCall}, Parts: []sessiondata.Part{{Kind: sessiondata.PartText, Text: "a", State: "available"}}})
+ list := make([]any, len(entries))
+ for i, e := range entries {
+ list[i] = e
+ }
+ r.body(call+":request", call, providerbody.RoleRequest, map[string]any{"messages": map[string]any{"entries": list}})
+ if reply != nil {
+ r.body(call+":response", call, providerbody.RoleResponse, map[string]any{"reply": reply})
+ }
+}
+
+// parse parses until the chain reaches the landed files, the index built
+// again from them first, as it is after files land that it does not hold.
+func (r *historyRoot) parse(budget int64) []*parse.Round {
+ r.t.Helper()
+ for _, p := range []string{r.z.IndexDir(historySession), r.z.IndexStatePath(historySession)} {
+ if err := os.RemoveAll(p); err != nil {
+ r.t.Fatal(err)
+ }
+ }
+ return parseSession(r.t, r.z, historySession, budget)
+}
+
+func parseSession(t *testing.T, z *storage.Zone, session string, budget int64) []*parse.Round {
+ t.Helper()
+ var rounds []*parse.Round
+ for {
+ rd, err := parse.Session(z, parse.Options{Conversation: session, Session: session, Reindex: index.Rebuild, MaxRoundBytes: budget})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if rd.Changed() {
+ rounds = append(rounds, rd)
+ }
+ if !rd.Changed() || !rd.More {
+ return rounds
+ }
+ if len(rounds) > 1000 {
+ t.Fatal("the chain does not reach the index")
+ }
+ }
+}
+
+// framesOf is every node frame the rounds wrote, by id, in order.
+func framesOf(t *testing.T, z *storage.Zone, session string, rounds []*parse.Round) map[string][]sessionflow.Node {
+ t.Helper()
+ ch := sessionflow.OpenChain(z.Root(), session)
+ out := map[string][]sessionflow.Node{}
+ for _, rd := range rounds {
+ r, err := ch.Open(rd.Path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, n := range r.Nodes {
+ out[n.ID] = append(out[n.ID], n)
+ }
+ }
+ return out
+}
+
+func attrsOf(t *testing.T, n *sessionflow.Node) map[string]any {
+ t.Helper()
+ var a map[string]any
+ if err := json.Unmarshal(n.Attrs, &a); err != nil {
+ t.Fatal(err)
+ }
+ return a
+}
+
+func foldOf(t *testing.T, z *storage.Zone, session string) *sessionflow.View {
+ t.Helper()
+ v, err := parse.View(z.Root(), session)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return v
+}
+
+func verified(t *testing.T, z *storage.Zone, session string) {
+ t.Helper()
+ rep, err := verify.Chain(z, session, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, r := range rep.Rounds {
+ for _, p := range r.Problems {
+ t.Error(p)
+ }
+ }
+ if rep.Problems != 0 {
+ t.Fatalf("the chain has %d problems", rep.Problems)
+ }
+}
+
+// sameNodes says the two folds hold the same nodes and relations, each
+// revision number aside.
+func sameNodes(t *testing.T, got, want *sessionflow.View) {
+ t.Helper()
+ g, w := withoutRevision(got.Nodes), withoutRevision(want.Nodes)
+ for id := range w {
+ if g[id] != w[id] {
+ t.Errorf("node %s: %s, fresh %s", id, g[id], w[id])
+ }
+ }
+ for id := range g {
+ if _, ok := w[id]; !ok {
+ t.Errorf("node %s is not in a fresh parse: %s", id, g[id])
+ }
+ }
+ if !reflect.DeepEqual(withoutRevision(got.Relations), withoutRevision(want.Relations)) {
+ t.Error("the relations differ from a fresh parse")
+ }
+}
+
+// A card is published, revised and removed by rounds of the chain as its
+// evidence lands: a response that lands late gives the earlier reply, and a
+// late record of the earlier call's class takes that call out of the stream's
+// history, so the round after it removes the card with a tombstone. The fold
+// ends as a fresh parse of the same files builds it.
+func TestACardIsRevisedAndRemovedAsItsEvidenceLands(t *testing.T) {
+ r := &historyRoot{t: t, z: storage.NewZone(t.TempDir())}
+ sys, q1, a1, q2, q3 := entry(1, 1), entry(2, 2), entry(3, 3), entry(4, 4), entry(5, 5)
+ r.record(sessiondata.Record{ID: "in", From: sessiondata.FromExternal, Run: "run", Flags: []string{"external_input"},
+ Parts: []sessiondata.Part{{Kind: sessiondata.PartText, Text: "q", State: "available"}}})
+ r.turn("c1", a1, sys, q1)
+ r.turn("x1", nil, q3) // read as a turn, so it and c2 each draw a card
+ r.turn("c2", nil, sys, q1, a1, q2)
+
+ first := r.parse(0)
+ card := framesOf(t, r.z, historySession, first)["history/c2"]
+ if len(card) != 1 || card[0].Tombstone {
+ t.Fatalf("the first rounds wrote the card as %+v, want it once", card)
+ }
+ if a := attrsOf(t, &card[0]); a["from_call"] != "call/x1" || a["reply"] != model.Unavailable || a["reply_reason"] != "no_reply" {
+ t.Fatalf("the first card: %v", a)
+ }
+
+ // x1's response lands: its reply, a1, is in c2's history.
+ r.body("x1:response", "x1", providerbody.RoleResponse, map[string]any{"reply": a1})
+ second := r.parse(0)
+ card = framesOf(t, r.z, historySession, second)["history/c2"]
+ if len(card) != 1 || card[0].Tombstone {
+ t.Fatalf("the late response's rounds wrote the card as %+v, want one revision", card)
+ }
+ if a := attrsOf(t, &card[0]); a["reply"] != "back" || a["reply_reason"] != nil {
+ t.Fatalf("the revised card: %v", a)
+ }
+ if card[0].Revision < 2 {
+ t.Fatalf("the revised card is revision %d", card[0].Revision)
+ }
+
+ // A later arrival of x1, under its own record id, says it was the
+ // framework's: c2 is now compared with c1, whose history it keeps.
+ r.record(sessiondata.Record{ID: "x1-late", From: sessiondata.FromAgent, Call: "x1", Run: "run",
+ Flags: []string{"finished", model.ClassInternalCall}})
+ third := r.parse(0)
+ frames := framesOf(t, r.z, historySession, third)
+ // x1 drew a card of its own, against c1, and is no longer compared.
+ for _, id := range []string{"history/c2", "history/x1"} {
+ if card := frames[id]; len(card) != 1 || !card[0].Tombstone {
+ t.Fatalf("the late class's rounds wrote %s as %+v, want a tombstone", id, card)
+ }
+ }
+ tombstones := 0
+ for _, rd := range third {
+ tombstones += rd.Tombstones
+ }
+ if tombstones != 2 {
+ t.Fatalf("the late class's rounds wrote %d tombstones, want the two cards'", tombstones)
+ }
+ call := frames["call/c2"]
+ if len(call) != 1 {
+ t.Fatalf("the late class's rounds wrote call/c2 %d times", len(call))
+ }
+ if a := attrsOf(t, &call[0]); a["history"] != "kept" || a["history_from"] != "call/c1" {
+ t.Fatalf("call/c2 after the late class: %v", a)
+ }
+
+ fold := foldOf(t, r.z, historySession)
+ if cards := fold.NodesByKind(model.KindHistoryChange); len(cards) != 0 {
+ t.Fatalf("the fold still holds %d card(s)", len(cards))
+ }
+ verified(t, r.z, historySession)
+ again := &historyRoot{t: t, z: storage.NewZone(t.TempDir())}
+ if err := os.CopyFS(again.z.SessionDir(historySession), os.DirFS(r.z.SessionDir(historySession))); err != nil {
+ t.Fatal(err)
+ }
+ for _, p := range []string{filepath.Join(again.z.SessionDir(historySession), "index"), again.z.SessionStatePath(historySession)} {
+ if err := os.RemoveAll(p); err != nil {
+ t.Fatal(err)
+ }
+ }
+ again.parse(0)
+ sameNodes(t, fold, foldOf(t, again.z, historySession))
+}
+
+// A chain whose rounds carry no history, as a parser before the history was
+// compared wrote it, continues under this one: the next rounds, one frame
+// each under a budget of one byte, write every call's history and the card
+// over no new file, and the fold ends as a fresh parse builds it.
+func TestAChainWithNoHistoryGetsItInRoundsOverNoNewFile(t *testing.T) {
+ sc, err := scenario.Load("../scenarios/history-forget.yaml")
+ if err != nil {
+ t.Fatal(err)
+ }
+ root := t.TempDir()
+ b, err := scenario.Build(sc, scenario.FormatSD, root, scenario.Options{At: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)})
+ if err != nil {
+ t.Fatal(err)
+ }
+ z := storage.NewZone(root)
+ session := b.Session
+ parseSession(t, z, session, 0)
+ fresh := foldOf(t, z, session)
+ cards := fresh.NodesByKind(model.KindHistoryChange)
+ if len(cards) == 0 {
+ t.Fatal("the scenario draws no card")
+ }
+
+ // The round an older parser would have written: no card, and no call
+ // that says what its history was.
+ ch := sessionflow.OpenChain(z.Root(), session)
+ files, err := ch.List()
+ if err != nil {
+ t.Fatal(err)
+ }
+ head, err := ch.Open(files[len(files)-1].Path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ h := head.Header
+ h.Round, h.Previous, h.FromSeq = h.Round+1, head.Commit.Digest, h.ThroughSeq+1
+ h.InputDigest = sessionflow.ChainInputDigest(h.InputDigest, nil)
+ w, err := sessionflow.NewWriter(h)
+ if err != nil {
+ t.Fatal(err)
+ }
+ stripped := 0
+ for _, id := range sortedKeys(fresh.Nodes) {
+ n := *fresh.Nodes[id]
+ switch n.Kind {
+ case model.KindHistoryChange:
+ n.Tombstone = true
+ case model.KindLLMCall:
+ a := attrsOf(t, &n)
+ if _, ok := a["history"]; !ok {
+ continue
+ }
+ for k := range a {
+ if strings.HasPrefix(k, "history") || k == "system_changed" {
+ delete(a, k)
+ }
+ }
+ if n.Attrs, err = json.Marshal(a); err != nil {
+ t.Fatal(err)
+ }
+ stripped++
+ default:
+ continue
+ }
+ if err := w.Node(n); err != nil {
+ t.Fatal(err)
+ }
+ }
+ data, digest, err := w.Close()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := ch.Publish(h.Round, digest, data); err != nil {
+ t.Fatal(err)
+ }
+ old := foldOf(t, z, session)
+ if len(old.NodesByKind(model.KindHistoryChange)) != 0 {
+ t.Fatal("the older chain still holds a card")
+ }
+
+ rounds := parseSession(t, z, session, 1)
+ if want := stripped + len(cards); len(rounds) != want {
+ t.Fatalf("the history took %d round(s), want %d, one a frame", len(rounds), want)
+ }
+ for _, rd := range rounds {
+ if rd.FromSeq != rd.ThroughSeq+1 {
+ t.Errorf("round %d covers sequences %d..%d, want no new file", rd.Number, rd.FromSeq, rd.ThroughSeq)
+ }
+ if rd.Nodes != 1 || rd.Relations+rd.Unresolved+rd.Tombstones != 0 {
+ t.Errorf("round %d holds %d nodes, %d relations, %d unresolved entries and %d tombstones, want one node",
+ rd.Number, rd.Nodes, rd.Relations, rd.Unresolved, rd.Tombstones)
+ }
+ }
+ if again := parseSession(t, z, session, 1); len(again) != 0 {
+ t.Fatalf("a parse after the history wrote %d round(s)", len(again))
+ }
+ sameNodes(t, foldOf(t, z, session), fresh)
+ verified(t, z, session)
+}
diff --git a/tests/claudecode_threads_test.go b/tests/claudecode_threads_test.go
index 24eab15..c1529a6 100644
--- a/tests/claudecode_threads_test.go
+++ b/tests/claudecode_threads_test.go
@@ -19,6 +19,7 @@ package tests_test
import (
"encoding/json"
+ "fmt"
"os"
"path/filepath"
"sort"
@@ -240,6 +241,25 @@ func TestAThreadsHistoryIsBuiltFromWhatItContinues(t *testing.T) {
if !spans {
t.Fatalf("the pair %s, %s that is not kept spans no reset", p.ID, n.ID)
}
+
+ // And the assembler says the same: ten kept with the reply back, the
+ // first call of each stream and the first after the compaction with no
+ // history before them, and no card.
+ got := map[string]int{}
+ for _, c := range all {
+ var a struct {
+ History string `json:"history"`
+ Reply string `json:"history_reply"`
+ }
+ _ = json.Unmarshal(c.node.Attrs, &a)
+ got[a.History+"/"+a.Reply]++
+ }
+ if fmt.Sprint(got) != fmt.Sprint(map[string]int{"first/": 3, "kept/back": 10}) {
+ t.Fatalf("history %v; want 3 first and 10 kept with the reply back", got)
+ }
+ if cards := view.NodesByKind(model.KindHistoryChange); len(cards) != 0 {
+ t.Fatalf("%d cards; no history here changes", len(cards))
+ }
}
func equalEntries(a, b []providerbody.Entry) bool {
diff --git a/tests/history_view_test.go b/tests/history_view_test.go
new file mode 100644
index 0000000..5bc6292
--- /dev/null
+++ b/tests/history_view_test.go
@@ -0,0 +1,151 @@
+// 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 tests_test
+
+import (
+ "encoding/json"
+ "fmt"
+ "testing"
+ "time"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/parse"
+ "github.com/apache/skywalking-ai-sessionizer/internal/retention"
+ "github.com/apache/skywalking-ai-sessionizer/internal/scenario"
+ "github.com/apache/skywalking-ai-sessionizer/internal/storage"
+ view_ "github.com/apache/skywalking-ai-sessionizer/internal/view"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/model"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessionview"
+)
+
+// TestACardStandsBeforeItsCallAndCarriesNothing reads the page a reader is
+// served for a history trimmed with no mark, as landed and again after
+// retention removed what the files held: the card is just before its call,
+// under the same parent, names the two requests' records, and carries no
+// text, state, bytes, flags or dropped parts, while its counts stay.
+func TestACardStandsBeforeItsCallAndCarriesNothing(t *testing.T) {
+ sc, err := scenario.Load("scenarios/history-forget.yaml")
+ if err != nil {
+ t.Fatal(err)
+ }
+ out := t.TempDir()
+ b, err := scenario.Build(sc, scenario.FormatSD, out, scenario.Options{At: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)})
+ if err != nil {
+ t.Fatal(err)
+ }
+ z := storage.NewZone(out)
+ for {
+ r, err := parse.Session(z, parse.Options{Conversation: b.Session, Session: b.Session})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !r.Changed() || !r.More {
+ break
+ }
+ }
+ page := func() *sessionview.Conversation {
+ c, err := view_.New(z, nil).Load(b.Session)
+ if err != nil {
+ t.Fatal(err)
+ }
+ doc, err := c.Build()
+ if err != nil {
+ t.Fatal(err)
+ }
+ return doc
+ }
+ var cardAttrs string
+ look := func(label string, doc *sessionview.Conversation) {
+ t.Helper()
+ var card *sessionview.Node
+ byID := map[string]*sessionview.Node{}
+ siblings := map[string][]string{}
+ var walk func(n *sessionview.Node)
+ walk = func(n *sessionview.Node) {
+ byID[n.ID] = n
+ for i := range n.Children {
+ siblings[n.ID] = append(siblings[n.ID], n.Children[i].ID)
+ walk(&n.Children[i])
+ }
+ if n.Kind == model.KindHistoryChange {
+ card = n
+ }
+ }
+ for i := range doc.Talks {
+ walk(&doc.Talks[i])
+ }
+ if card == nil {
+ t.Fatalf("%s: no card on the page", label)
+ }
+ var a struct {
+ From string `json:"from_call"`
+ To string `json:"to_call"`
+ }
+ _ = json.Unmarshal(card.Attrs, &a)
+ call, from := byID[a.To], byID[a.From]
+ if call == nil || from == nil {
+ t.Fatalf("%s: the card names %s and %s, which the page does not hold", label, a.From, a.To)
+ }
+ if card.Parent != call.Parent || *card.Ref != *call.Ref {
+ t.Errorf("%s: the card is under %s at %v, its call under %s at %v", label, card.Parent, card.Ref, call.Parent, call.Ref)
+ }
+ list := siblings[call.Parent]
+ for i, id := range list {
+ if id == card.ID && (i+1 >= len(list) || list[i+1] != call.ID) {
+ t.Errorf("%s: siblings %v; the card is not just before its call", label, list)
+ }
+ }
+ requestOf := func(n *sessionview.Node) string {
+ for _, p := range n.ProviderBodies {
+ if p.Role == "request" {
+ return fmt.Sprint(p.Ref)
+ }
+ }
+ return ""
+ }
+ if len(card.Refs) != 2 || fmt.Sprint(card.Refs[0]) != requestOf(from) || fmt.Sprint(card.Refs[1]) != requestOf(call) {
+ t.Errorf("%s: the card names %v, the requests are %s and %s", label, card.Refs, requestOf(from), requestOf(call))
+ }
+ if card.Text != "" || card.State != "" || card.Bytes != 0 || len(card.Flags) != 0 || len(card.Dropped) != 0 {
+ t.Errorf("%s: the card carries content: %q %q %d %v %v", label, card.Text, card.State, card.Bytes, card.Flags, card.Dropped)
+ }
+ if cardAttrs == "" {
+ cardAttrs = string(card.Attrs)
+ } else if string(card.Attrs) != cardAttrs {
+ t.Errorf("%s: the card says %s, and said %s", label, card.Attrs, cardAttrs)
+ }
+ }
+ look("as landed", page())
+ res := (&retention.Remover{Zone: z, MetadataOnlyAfter: time.Hour, Now: func() time.Time { return time.Date(2027, 1, 1, 0, 0, 0, 0, time.UTC) }}).Pass(retention.Input{})
+ if len(res.Errors) > 0 {
+ t.Fatal(res.Errors)
+ }
+ files, err := storage.LandedFiles(z, b.Session)
+ if err != nil {
+ t.Fatal(err)
+ }
+ skeletons := 0
+ for _, lf := range files {
+ if lf.Skeleton {
+ skeletons++
+ }
+ }
+ if skeletons == 0 {
+ t.Fatal("retention removed nothing; the test needs it to")
+ }
+ look("after retention", page())
+}
diff --git a/tests/langchain_history_test.go b/tests/langchain_history_test.go
index 530e991..513d05d 100644
--- a/tests/langchain_history_test.go
+++ b/tests/langchain_history_test.go
@@ -26,16 +26,19 @@ import (
"os"
"path/filepath"
"slices"
+ "sort"
"strings"
"testing"
"time"
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/langsmith"
"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/pkg/model"
"github.com/apache/skywalking-ai-sessionizer/pkg/providerbody"
"github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessionflow"
"github.com/apache/skywalking-ai-sessionizer/pkg/skeleton"
)
@@ -581,3 +584,148 @@ func TestAReplyIsLookedForOnlyAfterItsCall(t *testing.T) {
}
}
}
+
+// TestEveryLangChainCallSaysWhatItsHistoryWasComparedWith holds what the
+// assembler makes of every capture: a conversation that only grows keeps its
+// history call after call with the reply sent back; a reply the framework
+// sends back with the agent's name added is back_changed; the summariser's
+// own calls and a tool's plain calls are not compared; a marked summary is a
+// reset, so the call after it is the first of its epoch; and no capture
+// changes a history without a mark, so no card is drawn.
+func TestEveryLangChainCallSaysWhatItsHistoryWasComparedWith(t *testing.T) {
+ want := map[string]map[string]int{
+ "abandoned-run": {"first": 1},
+ "js-graph": {"first": 1, "kept/back": 1},
+ "js-traceable": {"first": 1},
+ "large-content": {"first": 1, "kept/back": 1},
+ "long-conversation": {"first": 1, "kept/back": 19},
+ "loop": {"first": 1, "kept/back": 5},
+ "no-thread-key": {"first": 1, "kept/back": 1},
+ "parallel-tools": {"first": 1, "kept/back": 1},
+ "plain": {"first": 1},
+ "shared-thread-key": {"first": 1, "kept/back": 2},
+ "slow-tool": {"first": 1, "kept/back": 1},
+ "streaming": {"first": 2, "kept/back": 2, "kept/back_changed": 1},
+ "subagent": {"first": 2, "kept/back": 2, "kept/back_changed": 1, "not_compared/auxiliary_stream": 1},
+ "subagent-own-thread": {"first": 2, "kept/back": 1, "kept/back_changed": 1},
+ "summarized": {"first": 3, "kept/back": 4, "not_compared/internal_call": 2},
+ "three-turns": {"first": 1, "kept/back": 3},
+ "tool-error": {"first": 1},
+ "traceable-only": {},
+ "two-threads": {"first": 2, "kept/back": 2},
+ "unsafe-thread-key": {"first": 4},
+ }
+ cases, err := os.ReadDir(corpus)
+ if err != nil {
+ t.Skipf("no corpus: %v", err)
+ }
+ seen := 0
+ for _, c := range cases {
+ if !c.IsDir() {
+ continue
+ }
+ seen++
+ zone, sessions := land(t, c.Name())
+ got := map[string]int{}
+ for _, s := range sessions {
+ if _, err := parse.Session(zone, parse.Options{Conversation: s, Session: s, Reindex: index.Rebuild}); err != nil {
+ t.Fatal(err)
+ }
+ view := fold(t, zone, s)
+ checkPredecessors(t, c.Name(), zone, s, view)
+ for _, n := range view.Nodes {
+ var a struct {
+ History string `json:"history"`
+ Reason string `json:"history_reason"`
+ Reply string `json:"history_reply"`
+ }
+ _ = json.Unmarshal(n.Attrs, &a)
+ switch n.Kind {
+ case model.KindHistoryChange:
+ t.Errorf("%s: a card %s; no capture changes a history without a mark", c.Name(), n.ID)
+ case model.KindLLMCall:
+ key := a.History
+ if a.Reason != "" {
+ key += "/" + a.Reason
+ }
+ if a.Reply != "" {
+ key += "/" + a.Reply
+ }
+ got[key]++
+ }
+ }
+ }
+ w, ok := want[c.Name()]
+ if !ok {
+ t.Errorf("%s: a new capture; measure it and add it here", c.Name())
+ continue
+ }
+ if fmt.Sprint(got) != fmt.Sprint(w) {
+ t.Errorf("%s: %v, want %v", c.Name(), got, w)
+ }
+ }
+ if seen != len(want) {
+ t.Errorf("%d captures, %d expected", seen, len(want))
+ }
+}
+
+// checkPredecessors holds each call's comparison to the calls around it: a
+// compared call names the nearest earlier call of its stream that is not the
+// framework's own, a first call has none before it but across a reset, and
+// a call's length is its own request's list.
+func checkPredecessors(t *testing.T, kase string, zone *storage.Zone, session string, view *sessionflow.View) {
+ t.Helper()
+ held, _ := heldInOrder(t, zone, session)
+ before := func(a, b *sessionflow.Ref) bool { return a.Seq < b.Seq || (a.Seq == b.Seq && a.Row < b.Row) }
+ byStream := map[string][]*sessionflow.Node{}
+ for _, n := range view.Nodes {
+ if n.Kind == model.KindLLMCall {
+ byStream[n.Stream] = append(byStream[n.Stream], n)
+ }
+ }
+ resets := view.NodesByKind(model.KindEpochBoundary)
+ for stream, calls := range byStream {
+ sort.Slice(calls, func(i, j int) bool { return before(calls[i].Ref, calls[j].Ref) })
+ var last *sessionflow.Node
+ for _, n := range calls {
+ var a struct {
+ History string `json:"history"`
+ Reason string `json:"history_reason"`
+ From string `json:"history_from"`
+ Messages *int `json:"history_messages"`
+ }
+ _ = json.Unmarshal(n.Attrs, &a)
+ // A request whose list reads says its length, and only such a one.
+ list, _, err := held.ListOf(strings.TrimPrefix(n.ID, "call/") + ":request")
+ switch {
+ case err == nil && a.Messages == nil:
+ t.Errorf("%s: %s says no length; its request holds %d messages", kase, n.ID, len(list))
+ case err == nil && *a.Messages != len(list):
+ t.Errorf("%s: %s says %d messages; its request holds %d", kase, n.ID, *a.Messages, len(list))
+ case err != nil && a.Messages != nil:
+ t.Errorf("%s: %s says %d messages; its request's list does not read: %v", kase, n.ID, *a.Messages, err)
+ }
+ switch a.History {
+ case "kept", "changed":
+ if last == nil || a.From != last.ID {
+ t.Errorf("%s: %s is compared with %s; the call before it in %s is %v", kase, n.ID, a.From, stream, last)
+ }
+ case "first":
+ if last != nil {
+ spans := false
+ for _, r := range resets {
+ if r.Stream == stream && before(last.Ref, r.Ref) && before(r.Ref, n.Ref) {
+ spans = true
+ }
+ }
+ if !spans {
+ t.Errorf("%s: %s is first, and %s comes before it with no reset between", kase, n.ID, last.ID)
+ }
+ }
+ }
+ if a.Reason != model.ClassInternalCall {
+ last = n
+ }
+ }
+ }
+}
diff --git a/tests/scenarios/history-forget.expect.yaml b/tests/scenarios/history-forget.expect.yaml
new file mode 100644
index 0000000..1b0df87
--- /dev/null
+++ b/tests/scenarios/history-forget.expect.yaml
@@ -0,0 +1,38 @@
+# 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.
+
+# Four calls. The fourth's request leaves out the two oldest messages after
+# the first, so its history is not the one before it with more after it: it
+# changed, a card stands just before it, and its cause is unavailable. Of the
+# third call's five messages three are kept and two dropped; the third call's
+# reply and the new question are added. No context reset: one epoch.
+checkpoints:
+ final:
+ kinds: {llm.call: 4, history.change: 1, epoch: 1, epoch.boundary: 0}
+ history: {first: 1, kept: 2, changed: 1}
+ nodes:
+ history/s9-call-2d0e5ed6b2ef:
+ kind: history.change
+ refs: 2
+ attrs: {from_call: call/s6-call-2d0e5ed6b2ef, to_call: call/s9-call-2d0e5ed6b2ef, kept: 3, changed: 0, dropped: 2, added: 2, order_changed: "no", reply: back, system_changed: "no", cause: unavailable}
+ call/s9-call-2d0e5ed6b2ef: {kind: llm.call, attrs: {history: changed, history_from: call/s6-call-2d0e5ed6b2ef, history_messages: 5, history_reply: back}}
+ view:
+ state: verified
+ provider_bodies: 8
+ captured_prompts: 4
+# Small, so the history attributes and the card travel through many rounds.
+parse: {max_round_bytes: 3072}
diff --git a/tests/scenarios/history-forget.yaml b/tests/scenarios/history-forget.yaml
new file mode 100644
index 0000000..8fd86b1
--- /dev/null
+++ b/tests/scenarios/history-forget.yaml
@@ -0,0 +1,34 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# An application trims the history it sends: after three turns, the next
+# request leaves out the two oldest messages after the first. No transcript
+# record shows it, only the request; the call that sends it says its history
+# changed, a card stands before it with what was kept and dropped, and the
+# cause is unavailable. It is no context reset: the calls stay in their epoch.
+title: a history trimmed with no mark
+provider_bodies: true
+steps:
+ - input: what is the timeout?
+ - call: {text: "It is 30 seconds."}
+ - input: and the retries?
+ - call: {text: "There are 3 retries."}
+ - input: and the backoff?
+ - call: {text: "It doubles each time."}
+ - forget: 2
+ - input: summarise the three
+ - call: {text: "30 seconds, 3 retries, doubling backoff."}
diff --git a/tests/scenarios/history-threads.expect.yaml b/tests/scenarios/history-threads.expect.yaml
new file mode 100644
index 0000000..b3d6b45
--- /dev/null
+++ b/tests/scenarios/history-threads.expect.yaml
@@ -0,0 +1,34 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Six calls: four on the main stream across the resume, two on the
+# searcher's thread. The first of each stream has no history before it, and
+# every other one keeps the history of the call before it, through the resume,
+# which starts a new thread with the whole list again; no card.
+checkpoints:
+ final:
+ kinds: {llm.call: 6, history.change: 0}
+ history: {first: 2, kept: 4}
+ nodes:
+ # The call after the resume: its request holds the whole list again.
+ call/s6-call-0b51c11a1403: {kind: llm.call, attrs: {history: kept, history_from: call/s3-call-0b51c11a1403, history_messages: 5, history_reply: back, system_changed: "no"}}
+ # The searcher's second call continues the searcher's own thread.
+ call/searcher-s2-call-0b51c11a1403: {kind: llm.call, attrs: {history: kept, history_from: call/searcher-s1-call-0b51c11a1403, history_messages: 3, history_reply: back}}
+ view:
+ state: verified
+ provider_bodies: 12
+ captured_prompts: 6
diff --git a/tests/scenarios/history-threads.yaml b/tests/scenarios/history-threads.yaml
new file mode 100644
index 0000000..be86525
--- /dev/null
+++ b/tests/scenarios/history-threads.yaml
@@ -0,0 +1,47 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Provider bodies as Claude Code 2.1.302 writes them: a process's first
+# request starts a provider thread and holds the whole message list, and
+# every later request continues it from the response before and holds only
+# what is new. A resumed process starts a new thread with the whole list
+# again, and a child has a thread of its own. Built from each thread, every
+# call's history is the one before it with the reply next: kept, and no card.
+title: provider threads across a resume
+provider_bodies: true
+provider_threads: true
+steps:
+ - input: read the configuration and summarise it
+ - call:
+ text: Reading it.
+ tool:
+ name: Read
+ input: {file_path: /Users/dev/scenario/config.yaml}
+ result: {text: "timeout: 30\nretries: 3\n", after: 200ms}
+ - call: {text: "The timeout is 30 seconds, with 3 retries."}
+ - resume: true
+ - input: find where the timeout is used
+ - call:
+ agent:
+ name: searcher
+ prompt: find every use of the timeout
+ steps:
+ - call:
+ text: Searching.
+ tool: {name: Bash, input: {command: grep -rn timeout .}, result: {text: "server.go:3: var timeout = 30", after: 300ms}}
+ - call: {text: It is used in server.go.}
+ - call: {text: The searcher found it in server.go.}