You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2026/09/06 05:50:19 UTC
(skywalking-ai-sessionizer) branch main updated: Let a scenario say what the original lost, and push it through the Collector
This is an automated email from the ASF dual-hosted git repository.
wu-sheng pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-ai-sessionizer.git
The following commit(s) were added to refs/heads/main by this push:
new d23a1d3 Let a scenario say what the original lost, and push it through the Collector
d23a1d3 is described below
commit d23a1d387696a99c24c6613627b6644661e7f850
Author: Wu Sheng <wu...@foxmail.com>
AuthorDate: Sun Sep 6 13:50:12 2026 +0800
Let a scenario say what the original lost, and push it through the Collector
A person trims a transcript, a child's file is deleted, a result never
reaches the disk. The collector lands what exists and numbers the lines it
finds, so nothing asz holds is damaged and the loss has to show in the
structure. Until now no scenario could say it, since a step describes what
the agent did and the loss happened to the file afterwards.
lost: true on a step leaves its records out of the file; on a result inside
a tool, only the result; on an agent, a skill or a workflow child, the
child's whole file and its meta file. The planner marks the records and
keeps the clock, the ids and the parent chain as they would be, so the
record after a lost one still names it as its parent. Both writers leave
the same records out, and a replayed copy is still written, since the
runtime copies from memory. A replay step cannot itself be lost.
tests/scenarios/lost-records.yaml covers each kind of loss. Every property
holds in both formats and the folds agree, and the five losses show as
open references: tool_result, tool_use, child_stream, notified_child and
spawn_of_child. The document stays verified with four files, not six. The
Collector job builds this scenario into the same root as the one holding
every kind of file, so a session with open references is rebuilt and
verified from what a real Collector wrote, over gRPC and over HTTP. The
Collector check now compares kinds against what the root holds.
---
docs/en/changes/changes-0.2.0.md | 9 ++++
docs/en/guides/scenario.md | 27 +++++++++++-
internal/scenario/plan.go | 76 ++++++++++++++++++++++----------
internal/scenario/scenario.go | 20 +++++++++
internal/scenario/write_claudecode.go | 12 ++++-
internal/scenario/write_sd.go | 14 ++++--
tests/scenarios/lost-records.expect.yaml | 49 ++++++++++++++++++++
tests/scenarios/lost-records.yaml | 51 +++++++++++++++++++++
tools/collectorcheck/main.go | 39 +++++++++++++---
tools/e2e-collector.sh | 4 ++
10 files changed, 266 insertions(+), 35 deletions(-)
diff --git a/docs/en/changes/changes-0.2.0.md b/docs/en/changes/changes-0.2.0.md
index e41541e..b7d2a53 100644
--- a/docs/en/changes/changes-0.2.0.md
+++ b/docs/en/changes/changes-0.2.0.md
@@ -64,6 +64,15 @@
an OTLP receiver in the test and checked against the export page, one file per request, both
formats and all six file kinds, then rebuilt from the wire and verified.
+- A scenario can say what the original lost. `lost: true` on a step leaves its records out of
+ the file, on a `result` only the result, and on an `agent`, a `skill` or a workflow child the
+ child's whole file, while the clock, the ids and the parent chain move as if the records were
+ there. Both writers leave the same records out, so the landed data verifies and the loss shows
+ as open references. `tests/scenarios/lost-records.yaml` covers each kind of loss, is pushed
+ over both transports like every scenario, and is pushed into the real Collector in CI next to
+ the scenario that holds every kind of file, so a session with open references is rebuilt and
+ verified from what the Collector wrote.
+
## Read
- `asz.view`, version 1.0, is one conversation rebuilt from its rounds and its landed files as one
diff --git a/docs/en/guides/scenario.md b/docs/en/guides/scenario.md
index 3f69e34..3120d93 100644
--- a/docs/en/guides/scenario.md
+++ b/docs/en/guides/scenario.md
@@ -86,7 +86,7 @@ steps:
- reset: {summary: "Summary: the build was run and checked."}
```
-Every step is exactly one of these, plus an optional `after` and `checkpoint`:
+Every step is exactly one of these, plus an optional `after`, `checkpoint` and `lost`:
| Step | What it writes |
| --- | --- |
@@ -106,6 +106,31 @@ steps}]}` starts children as one batch, with a journal, a manifest and a script.
the same in every format: the step's position names its records, a tool is `tool/<id>`, a run is
`<step>-cycle`, a child stream's id is derived from its name.
+### Records the original lost
+
+`lost: true` on a step says the original file does not hold what the step wrote: a person trimmed
+the transcript, or the write never reached the disk. The step still happened. The clock and the
+ids move as if the records were there, and the record after them still names them as its parent,
+so the loss shows where it shows in a real corpus: in references the assembler cannot resolve. On
+a `result` inside a `tool`, only the result is lost and the call stays. On an `agent`, a `skill` or
+a workflow child, the child's file never reached the collector, nor its meta file, while the
+parent's records about the child stay.
+
+```yaml
+steps:
+ - call: {text: Reading it., tool: {name: Read, result: {text: "root: ./data", lost: true}}}
+ - call: {text: Checking the port., tool: {name: Bash}}
+ lost: true # the call is gone; the result below is not
+ - result: {of: s2-tool, text: "8787 LISTEN"}
+ - call: {text: Asking the helper., agent: {name: helper, lost: true, steps: [{call: {text: Done.}}]}}
+```
+
+Both formats leave the same records out. The collector numbers the lines it finds, so the landed
+data verifies and the chain is intact, and the document stays `verified`: nothing asz holds is
+damaged, and the incompleteness is in the evidence. What a check sees is the `unresolved` list, a
+`tool_result` for the first step, a `tool_use` for the second, a `child_stream` for the third.
+`tests/scenarios/lost-records.yaml` covers every kind of loss.
+
## Check
```sh
diff --git a/internal/scenario/plan.go b/internal/scenario/plan.go
index 885798e..a0b5f9b 100644
--- a/internal/scenario/plan.go
+++ b/internal/scenario/plan.go
@@ -110,6 +110,10 @@ type Event struct {
Text string
// Replayed marks a copy the runtime re-emitted, with its run rewritten.
Replayed bool
+ // Lost marks a record the original file does not hold. It is planned
+ // like any other, so the clock, the ids and the parent chain are what
+ // they would be, and every writer leaves it out.
+ Lost bool
// A fragment.
Frag FragKind
@@ -150,6 +154,20 @@ type Stream struct {
Prompt string
Tool string
Batch string
+ // Lost says the stream's file never reached the collector.
+ Lost bool
+}
+
+// lostStreams is the set of child streams whose file never reached the
+// collector. A writer writes nothing for them: no transcript and no meta.
+func (p *Plan) lostStreams() map[string]bool {
+ out := map[string]bool{}
+ for _, s := range p.Streams {
+ if s.Lost {
+ out[s.ID] = true
+ }
+ }
+ return out
}
// Run is one workflow run: its children and the files filed with it.
@@ -244,6 +262,9 @@ type lane struct {
first bool
n int
prefix string
+ // lost says the whole stream's file is lost: every record on the lane
+ // is marked lost as it is planned.
+ lost bool
}
type planner struct {
@@ -275,11 +296,12 @@ func (b *planner) emit(e Event) {
}
func (b *planner) step(l *lane, s *Step, id string) error {
+ lost := l.lost || s.Lost
switch {
case s.Input != "":
at := b.tick(l, s.After)
run := id + "-cycle"
- e := Event{Kind: EvInput, Stream: l.stream, Batch: l.batch, At: at, ID: id + "-input", Run: run, Text: s.Input}
+ e := Event{Kind: EvInput, Stream: l.stream, Batch: l.batch, At: at, ID: id + "-input", Run: run, Text: s.Input, Lost: lost}
if l.stream != "main" {
e.Parent = l.last
}
@@ -287,12 +309,12 @@ func (b *planner) step(l *lane, s *Step, id string) error {
l.last = e.ID
case s.Queued != nil:
at := b.tick(l, s.After)
- e := Event{Kind: EvQueued, Stream: l.stream, Batch: l.batch, At: at, ID: id + "-queued", Parent: l.last, Text: s.Queued.Text, Mode: s.Queued.Mode}
+ e := Event{Kind: EvQueued, Stream: l.stream, Batch: l.batch, At: at, ID: id + "-queued", Parent: l.last, Text: s.Queued.Text, Mode: s.Queued.Mode, Lost: lost}
b.emit(e)
l.last = e.ID
case s.Inject != nil:
at := b.tick(l, s.After)
- e := Event{Kind: EvInject, Stream: l.stream, Batch: l.batch, At: at, ID: id + "-inject", Parent: l.last, Type: s.Inject.Type, Text: s.Inject.Text}
+ e := Event{Kind: EvInject, Stream: l.stream, Batch: l.batch, At: at, ID: id + "-inject", Parent: l.last, Type: s.Inject.Type, Text: s.Inject.Text, Lost: lost}
b.emit(e)
l.last = e.ID
case s.Call != nil:
@@ -301,24 +323,24 @@ func (b *planner) step(l *lane, s *Step, id string) error {
at := b.tick(l, s.After)
r := s.Result
e := Event{Kind: EvResult, Stream: l.stream, Batch: l.batch, At: at, ID: id + "-result", Parent: l.last,
- Run: b.runOf(l), Of: r.Of, Text: r.Text, Failed: r.Failed, StringEnrichment: r.String}
+ Run: b.runOf(l), Of: r.Of, Text: r.Text, Failed: r.Failed, StringEnrichment: r.String, Lost: lost || r.Lost}
b.emit(e)
l.last = e.ID
case s.Error != "":
at := b.tick(l, s.After)
e := Event{Kind: EvSynthetic, Stream: l.stream, Batch: l.batch, At: at, ID: id + "-synthetic", Parent: l.last,
- Call: id + "-synthetic-call", Req: id + "-req", Text: s.Error}
+ Call: id + "-synthetic-call", Req: id + "-req", Text: s.Error, Lost: lost}
b.emit(e)
l.last = e.ID
case s.Reset != nil:
at := b.tick(l, s.After)
- boundary := Event{Kind: EvBoundary, Stream: l.stream, At: at, ID: id + "-boundary", Continues: l.last}
+ boundary := Event{Kind: EvBoundary, Stream: l.stream, At: at, ID: id + "-boundary", Continues: l.last, Lost: lost}
b.emit(boundary)
// The summary is timestamped before the boundary that produced it,
// as the runtime writes it, so anything ordering an epoch by time
// gets it backwards. That is the property a scenario must reproduce.
summary := Event{Kind: EvSummary, Stream: l.stream, At: at.Add(-400 * time.Millisecond), ID: id + "-summary",
- Parent: boundary.ID, Run: id + "-cycle-compact", Text: s.Reset.Summary}
+ Parent: boundary.ID, Run: id + "-cycle-compact", Text: s.Reset.Summary, Lost: lost}
b.emit(summary)
l.last = summary.ID
case s.Replay > 0:
@@ -326,7 +348,7 @@ func (b *planner) step(l *lane, s *Step, id string) error {
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,
- Type: s.System.Subtype, Fields: s.System.Fields}
+ Type: s.System.Subtype, Fields: s.System.Fields, Lost: lost}
b.emit(e)
l.last = e.ID
}
@@ -334,6 +356,8 @@ func (b *planner) step(l *lane, s *Step, id string) error {
}
// runOf is the run a record on a lane belongs to: the last input's cycle.
+// A lost input still opened its run: the runtime stamped the records after
+// it with that run, whatever the file holds now.
func (b *planner) runOf(l *lane) string {
for i := len(b.p.Events) - 1; i >= 0; i-- {
e := &b.p.Events[i]
@@ -346,6 +370,7 @@ func (b *planner) runOf(l *lane) string {
func (b *planner) call(l *lane, s *Step, id string) error {
c := s.Call
+ lost := l.lost || s.Lost
at := b.tick(l, s.After)
usage := Usage{In: 2, Out: 40, CacheRead: 900, CacheWrite: 100}
if c.Usage != nil {
@@ -402,6 +427,7 @@ func (b *planner) call(l *lane, s *Step, id string) error {
f.ID = fmt.Sprintf("%s-call-f%d", id, i+1)
f.Parent = l.last
f.Call, f.Req, f.Usage = call, req, usage
+ f.Lost = lost
f.Last = i == len(frags)-1
f.Stop = "end_turn"
if tool != nil {
@@ -417,7 +443,7 @@ func (b *planner) call(l *lane, s *Step, id string) error {
r := c.Tool.Result
rat := b.tick(l, r.After)
e := Event{Kind: EvResult, Stream: l.stream, Batch: l.batch, At: rat, ID: id + "-result", Parent: l.last, Run: run,
- Of: tool.ID, Text: r.Text, Failed: r.Failed, StringEnrichment: r.String}
+ Of: tool.ID, Text: r.Text, Failed: r.Failed, StringEnrichment: r.String, Lost: lost || r.Lost}
b.emit(e)
l.last = e.ID
case c.Agent != nil:
@@ -425,18 +451,18 @@ func (b *planner) call(l *lane, s *Step, id string) error {
child := agentID(a.Name)
ack := Event{Kind: EvResult, Stream: l.stream, Batch: l.batch, At: b.tick(l, 100*time.Millisecond), ID: id + "-ack",
Parent: l.last, Run: run, Of: tool.ID, Text: "launched",
- Ack: &Ack{Child: child, Prompt: a.Prompt, Label: a.Name}}
+ Ack: &Ack{Child: child, Prompt: a.Prompt, Label: a.Name}, Lost: lost}
b.emit(ack)
l.last = ack.ID
- b.p.Streams = append(b.p.Streams, Stream{ID: child, Label: a.Name, Prompt: a.Prompt, Tool: tool.ID})
- cl := &lane{stream: child, t: ack.At, first: true, prefix: a.Name}
+ b.p.Streams = append(b.p.Streams, Stream{ID: child, Label: a.Name, Prompt: a.Prompt, Tool: tool.ID, Lost: a.Lost})
+ cl := &lane{stream: child, t: ack.At, first: true, prefix: a.Name, lost: a.Lost}
if a.After == 0 {
cl.t = cl.t.Add(b.scaled(b.p.interval))
} else {
cl.t = cl.t.Add(b.scaled(a.After))
}
if a.Prompt != "" {
- prompt := Event{Kind: EvInput, Stream: child, At: cl.t, ID: child + "-prompt", Run: child + "-cycle", Text: a.Prompt}
+ prompt := Event{Kind: EvInput, Stream: child, At: cl.t, ID: child + "-prompt", Run: child + "-cycle", Text: a.Prompt, Lost: a.Lost}
cl.first = false
b.emit(prompt)
cl.last = prompt.ID
@@ -450,7 +476,7 @@ func (b *planner) call(l *lane, s *Step, id string) error {
}
nat := b.tick(l, 0)
notice := Event{Kind: EvNotice, Stream: l.stream, Batch: l.batch, At: nat, ID: id + "-notice", Parent: l.last,
- Run: id + "-cycle-notification", NoticeTool: tool.ID, NoticeChild: child,
+ Run: id + "-cycle-notification", NoticeTool: tool.ID, NoticeChild: child, Lost: lost,
Text: "<task-notification>\n<task-id>" + child + "</task-id>\n<tool-use-id>" + tool.ID + "</tool-use-id>\n<status>completed</status>\n</task-notification>"}
b.emit(notice)
l.last = notice.ID
@@ -459,12 +485,12 @@ func (b *planner) call(l *lane, s *Step, id string) error {
sk := c.Skill
child := agentID(sk.Agent)
res := Event{Kind: EvResult, Stream: l.stream, Batch: l.batch, At: b.tick(l, 0), ID: id + "-result", Parent: l.last, Run: run,
- Of: tool.ID, Text: "forked", Fork: &Fork{Child: child}}
+ Of: tool.ID, Text: "forked", Fork: &Fork{Child: child}, Lost: lost}
b.emit(res)
l.last = res.ID
- b.p.Streams = append(b.p.Streams, Stream{ID: child, Label: sk.Agent, Tool: tool.ID})
- cl := &lane{stream: child, t: res.At.Add(b.scaled(b.p.interval)), prefix: sk.Agent}
- prompt := Event{Kind: EvInput, Stream: child, At: cl.t, ID: child + "-prompt", Run: child + "-cycle", Text: "run the skill " + sk.Name}
+ b.p.Streams = append(b.p.Streams, Stream{ID: child, Label: sk.Agent, Tool: tool.ID, Lost: sk.Lost})
+ cl := &lane{stream: child, t: res.At.Add(b.scaled(b.p.interval)), prefix: sk.Agent, lost: sk.Lost}
+ prompt := Event{Kind: EvInput, Stream: child, At: cl.t, ID: child + "-prompt", Run: child + "-cycle", Text: "run the skill " + sk.Name, Lost: sk.Lost}
b.emit(prompt)
cl.last = prompt.ID
if err := b.steps(cl, sk.Steps); err != nil {
@@ -474,7 +500,7 @@ func (b *planner) call(l *lane, s *Step, id string) error {
w := c.Workflow
runID := "wf_" + strings.ReplaceAll(w.Name, " ", "-")
res := Event{Kind: EvResult, Stream: l.stream, Batch: l.batch, At: b.tick(l, 0), ID: id + "-result", Parent: l.last, Run: run,
- Of: tool.ID, Text: "workflow launched", Launch: &Launch{Run: runID, Name: w.Name}}
+ Of: tool.ID, Text: "workflow launched", Launch: &Launch{Run: runID, Name: w.Name}, Lost: lost}
b.emit(res)
l.last = res.ID
r := Run{ID: runID, Name: w.Name, Script: "export const meta = {\n name: '" + w.Name + "',\n}", ScriptProject: w.ScriptProject}
@@ -482,11 +508,11 @@ func (b *planner) call(l *lane, s *Step, id string) error {
for _, ch := range w.Children {
child := agentID(w.Name + "/" + ch.Name)
r.Children = append(r.Children, child)
- b.p.Streams = append(b.p.Streams, Stream{ID: child, Label: ch.Name, Prompt: ch.Prompt, Batch: runID})
+ b.p.Streams = append(b.p.Streams, Stream{ID: child, Label: ch.Name, Prompt: ch.Prompt, Batch: runID, Lost: ch.Lost})
t = t.Add(b.scaled(b.p.interval))
r.Journal = append(r.Journal, JournalLine{Type: "started", Child: child, At: t})
- cl := &lane{stream: child, batch: runID, t: t, prefix: ch.Name}
- prompt := Event{Kind: EvInput, Stream: child, Batch: runID, At: t, ID: child + "-prompt", Run: child + "-cycle", Text: ch.Prompt}
+ cl := &lane{stream: child, batch: runID, t: t, prefix: ch.Name, lost: ch.Lost}
+ prompt := Event{Kind: EvInput, Stream: child, Batch: runID, At: t, ID: child + "-prompt", Run: child + "-cycle", Text: ch.Prompt, Lost: ch.Lost}
if prompt.Text == "" {
prompt.Text = "do " + ch.Name
}
@@ -520,7 +546,9 @@ func (b *planner) steps(l *lane, steps []Step) error {
// replay copies the last n main-stream records with the run rewritten, as
// the runtime does before a reset. The copies keep their ids and times: the
-// later copy is the worse one, and the assembler must keep the first.
+// later copy is the worse one, and the assembler must keep the first. The
+// runtime copies from memory, so a copy is written even when its original
+// was lost from the file.
func (b *planner) replay(n int) {
var idx []int
for i := len(b.p.Events) - 1; i >= 0 && len(idx) < n; i-- {
@@ -530,7 +558,7 @@ func (b *planner) replay(n int) {
}
for i := len(idx) - 1; i >= 0; i-- {
e := b.p.Events[idx[i]]
- e.Replayed = true
+ e.Replayed, e.Lost = true, false
if e.Run != "" {
e.Run = "replayed-cycle"
}
diff --git a/internal/scenario/scenario.go b/internal/scenario/scenario.go
index e2a3dc8..c945b6f 100644
--- a/internal/scenario/scenario.go
+++ b/internal/scenario/scenario.go
@@ -61,6 +61,13 @@ type Step struct {
// Checkpoint names a point a build can stop at, so a test can land what
// exists so far, parse it, and check the fold before the rest arrives.
Checkpoint string `yaml:"checkpoint"`
+ // Lost says the original file does not hold what this step wrote: a
+ // person trimmed the transcript, or the write never reached the disk.
+ // The step still happened, so the clock and the ids move as if the
+ // records were there, and the record after them still names them as
+ // its parent. A writer leaves them out, in either format, and the
+ // collector lands what exists.
+ Lost bool `yaml:"lost"`
// Input is a person's message: it opens a run and a talk.
Input string `yaml:"input"`
@@ -146,6 +153,8 @@ type Result struct {
// String writes the runtime's enrichment as a bare string rather than
// an object, a shape a real corpus has on a noticeable share of results.
String bool `yaml:"string"`
+ // Lost says the result never reached the file, while the call did.
+ Lost bool `yaml:"lost"`
}
// UnmarshalYAML lets a result be a plain string.
@@ -174,6 +183,9 @@ type Agent struct {
After time.Duration `yaml:"after"`
Steps []Step `yaml:"steps"`
Notify bool `yaml:"notify"`
+ // Lost says the child's file never reached the collector, nor its meta
+ // file. The parent's records about the child stay.
+ Lost bool `yaml:"lost"`
}
// Skill is a fork: a child announced only by the parent's result block,
@@ -182,6 +194,8 @@ type Skill struct {
Name string `yaml:"name"`
Agent string `yaml:"agent"`
Steps []Step `yaml:"steps"`
+ // Lost says the child's file never reached the collector.
+ Lost bool `yaml:"lost"`
}
// Workflow starts several children as one batch. ScriptProject files the
@@ -198,6 +212,9 @@ type Child struct {
Name string `yaml:"name"`
Prompt string `yaml:"prompt"`
Steps []Step `yaml:"steps"`
+ // Lost says the child's file never reached the collector. The run's
+ // journal still names the child.
+ Lost bool `yaml:"lost"`
}
// Reset is a context reset and the summary that replaced the context.
@@ -278,6 +295,9 @@ func validateSteps(steps []Step, where string, seen map[string]bool, main bool)
if s.Replay > 0 && !main {
return fmt.Errorf("%s: replay belongs to the main stream", at)
}
+ if s.Replay > 0 && s.Lost {
+ return fmt.Errorf("%s: a replay copies what the runtime holds in memory; mark the original step lost instead", at)
+ }
if s.Reset != nil && !main {
return fmt.Errorf("%s: a reset belongs to the main stream", at)
}
diff --git a/internal/scenario/write_claudecode.go b/internal/scenario/write_claudecode.go
index e1b42da..7274ae0 100644
--- a/internal/scenario/write_claudecode.go
+++ b/internal/scenario/write_claudecode.go
@@ -36,8 +36,15 @@ import (
func writeClaudeCode(p *Plan, root string) ([]string, error) {
proj := filepath.Join(root, p.Project)
w := &ccWriter{p: p, files: map[string][]string{}}
+ lost := p.lostStreams()
for i := range p.Events {
- if err := w.event(&p.Events[i]); err != nil {
+ e := &p.Events[i]
+ // A lost record leaves no line; a lost stream leaves no file. The
+ // records around them are written as they are, still naming them.
+ if e.Lost || lost[e.Stream] {
+ continue
+ }
+ if err := w.event(e); err != nil {
return nil, err
}
}
@@ -72,6 +79,9 @@ func writeClaudeCode(p *Plan, root string) ([]string, error) {
return nil, err
}
for _, s := range p.Streams {
+ if s.Lost {
+ continue
+ }
lines := w.files[s.ID]
if s.Batch != "" {
if err := put(fmt.Sprintf("%s/subagents/workflows/%s/agent-%s.jsonl", p.Session, s.Batch, s.ID), lines); err != nil {
diff --git a/internal/scenario/write_sd.go b/internal/scenario/write_sd.go
index 8eb0cb5..aaebb90 100644
--- a/internal/scenario/write_sd.go
+++ b/internal/scenario/write_sd.go
@@ -62,15 +62,23 @@ func writeSD(p *Plan, root string, now time.Time) ([]string, error) {
w := &sdWriter{p: p, z: z, state: state, now: now}
// Streams first, main before its children, then the runs' files, so the
- // sequences a round refers to come out in landed order.
+ // sequences a round refers to come out in landed order. A lost record
+ // is left out and the records after it are numbered as the collector
+ // would number the lines it finds; a lost stream lands nothing.
+ lost := p.lostStreams()
byStream := map[string][]*Event{}
for i := range p.Events {
e := &p.Events[i]
+ if e.Lost || lost[e.Stream] {
+ continue
+ }
byStream[e.Stream] = append(byStream[e.Stream], e)
}
order := []string{"main"}
for _, s := range p.Streams {
- order = append(order, s.ID)
+ if !s.Lost {
+ order = append(order, s.ID)
+ }
}
for _, stream := range order {
if err := w.stream(stream, byStream[stream]); err != nil {
@@ -78,7 +86,7 @@ func writeSD(p *Plan, root string, now time.Time) ([]string, error) {
}
}
for _, s := range p.Streams {
- if s.Batch != "" {
+ if s.Batch != "" || s.Lost {
continue
}
if err := w.meta(s); err != nil {
diff --git a/tests/scenarios/lost-records.expect.yaml b/tests/scenarios/lost-records.expect.yaml
new file mode 100644
index 0000000..aaefaac
--- /dev/null
+++ b/tests/scenarios/lost-records.expect.yaml
@@ -0,0 +1,49 @@
+#
+# 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.
+#
+
+# Five kinds of loss, each an open reference and nothing else: the
+# structure around them is kept, the document is verified, and every
+# property holds, since the landed data is intact and only the original
+# was not. The lost child leaves no file, so the root has four files, not
+# six; the child whose launch was lost is a stream nothing starts.
+checkpoints:
+ final:
+ rounds: 1
+ written: true
+ kinds: {talk: 2, tool: 2, llm.call: 4, stream: 2, agent.launch_ack: 1, runtime.notification: 1, agent.output: 1}
+ talks_on: {main: 1, reviewer: 1}
+ relations: {starts: 0, ends_with: 1, reports: 0}
+ unresolved: {open: 5, resolved: 0}
+ unresolved_kinds:
+ tool_result: open # the result of s2-tool never reached the file
+ tool_use: open # a result names s3-tool, whose call is gone
+ child_stream: open # the helper was launched, its file is gone
+ notified_child: open # the helper's notification names a stream that never landed
+ spawn_of_child: open # the reviewer's file landed, the call that started it is gone
+ nodes:
+ tool/s2-tool: {refs: 1, attrs: {result: unavailable}}
+ stream/helper: {absent: true}
+ stream/reviewer: {kind: stream}
+ view:
+ state: verified
+ talks: 2
+ files: 4
+ rounds: 1
+push:
+ kinds: [transcript, agent_meta, round]
diff --git a/tests/scenarios/lost-records.yaml b/tests/scenarios/lost-records.yaml
new file mode 100644
index 0000000..9b72cb6
--- /dev/null
+++ b/tests/scenarios/lost-records.yaml
@@ -0,0 +1,51 @@
+#
+# 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.
+#
+
+# Records the original file lost. A person trimmed the transcript, a child's
+# file was deleted, a result never reached the disk. The collector lands what
+# exists and numbers the lines it finds, so the landed data verifies and the
+# chain is intact. The loss shows in the structure instead: every reference
+# into what is gone stays open as unresolved, and the document stays
+# verified, because nothing asz holds is damaged.
+title: lost records
+steps:
+ - input: read the configuration
+ # The result never reached the file; the call did.
+ - call:
+ text: Reading it.
+ tool: {name: Read, input: {path: asz.yaml}, result: {text: "storage:\n root: ./data", lost: true}}
+ # The call is gone; its result, arriving on its own, is not.
+ - call:
+ text: Checking the port.
+ tool: {name: Bash, input: {command: lsof -i :8787}}
+ lost: true
+ - result: {of: s3-tool, text: "asz 8787 LISTEN"}
+ # The child's file was deleted; the launch and the notification stay.
+ - call:
+ text: Asking the helper.
+ agent: {name: helper, prompt: read the docs, lost: true, notify: true, steps: [{call: {text: The docs say 11800.}}]}
+ # The launch and the notification are gone; the child's file is not.
+ - call:
+ text: Asking the reviewer.
+ agent: {name: reviewer, prompt: review the change, notify: true, steps: [{call: {text: Looks right.}}]}
+ lost: true
+ # The person's message is gone; the reply to it is not.
+ - input: thanks
+ lost: true
+ - call: {text: You are welcome.}
diff --git a/tools/collectorcheck/main.go b/tools/collectorcheck/main.go
index dca8fe7..46d6f63 100644
--- a/tools/collectorcheck/main.go
+++ b/tools/collectorcheck/main.go
@@ -164,18 +164,45 @@ func run(root, logs string) error {
kinds[r.kind]++
sessions[r.session] = true
}
- for _, k := range []string{"transcript", "agent_meta", "journal", "workflow_manifest", "workflow_script", "round"} {
- if kinds[k] == 0 {
- return fmt.Errorf("no record of kind %s reached the Collector; it received %v", k, kinds)
- }
- }
+ // Every file of the root reached the Collector, and every kind the root
+ // holds is among the records. The all-kinds scenario is built into the
+ // root, so that is every kind the export page names.
files := 0
+ rootKinds := map[string]int{}
_ = filepath.WalkDir(root, func(path string, d os.DirEntry, err error) error {
- if err == nil && !d.IsDir() && (strings.HasSuffix(path, ".sd") || strings.HasSuffix(path, ".sf")) {
+ if err != nil || d.IsDir() {
+ return nil
+ }
+ switch {
+ case strings.HasSuffix(path, ".sf"):
files++
+ rootKinds["round"]++
+ case strings.HasSuffix(path, ".sd"):
+ files++
+ data, rerr := os.ReadFile(path)
+ if rerr != nil {
+ return nil
+ }
+ first, _, _ := bytes.Cut(data, []byte("\n"))
+ var h struct {
+ Kind string `json:"kind"`
+ }
+ if json.Unmarshal(first, &h) == nil {
+ rootKinds[h.Kind]++
+ }
}
return nil
})
+ for k, n := range rootKinds {
+ if kinds[k] != n {
+ return fmt.Errorf("the root holds %d files of kind %s, the Collector received %d; it received %v", n, k, kinds[k], kinds)
+ }
+ }
+ for _, k := range []string{"transcript", "agent_meta", "journal", "workflow_manifest", "workflow_script", "round"} {
+ if rootKinds[k] == 0 {
+ return fmt.Errorf("the root holds no file of kind %s, so the check exercised less than the export page names", k)
+ }
+ }
if len(recs) != files {
return fmt.Errorf("the Collector received %d records, the root holds %d files", len(recs), files)
}
diff --git a/tools/e2e-collector.sh b/tools/e2e-collector.sh
index f4a7707..e65dff1 100755
--- a/tools/e2e-collector.sh
+++ b/tools/e2e-collector.sh
@@ -53,7 +53,11 @@ YAML
push_over() {
protocol="$1"; endpoint="$2"
root="$WORK/$protocol"
+ # Two scenarios in one root: every kind of file the export page names,
+ # and a session whose original lost records, so what the Collector gets
+ # back is a session with open references, rebuilt and verified as such.
./bin/asz scenario build tests/scenarios/all-kinds.yaml --format claude-code --out "$root" --at 2026-06-01T09:00:00Z --repeat 2 >/dev/null
+ ./bin/asz scenario build tests/scenarios/lost-records.yaml --format claude-code --out "$root" --at 2026-06-01T10:00:00Z >/dev/null
./bin/asz collect -once -config "$root/asz.yaml" >/dev/null
./bin/asz parse -config "$root/asz.yaml" >/dev/null
./bin/asz verify -config "$root/asz.yaml" >/dev/null