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/16 23:37:56 UTC
(skywalking-ai-sessionizer) branch main updated: Collect no LangChain run whose conversation cannot be told (#70)
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 01539f4 Collect no LangChain run whose conversation cannot be told (#70)
01539f4 is described below
commit 01539f4e3cd7ef195ed2b5c01442b35a8b9a486d
Author: 吴晟 Wu Sheng <wu...@foxmail.com>
AuthorDate: Sat Oct 17 07:37:50 2026 +0800
Collect no LangChain run whose conversation cannot be told (#70)
A run's conversation can depend on something the receiver no longer
has. An update that names no conversation of its own is completed from
its run's start, and a run that names no thread takes the conversation
of a run above it. The receiver remembers both for a day, and at most
50,000 runs. When what would say the conversation was forgotten, or
never came, the run landed under a name saying its identity was not
supplied. That put one conversation's evidence in another.
Such a run now waits, as a run waits for its ancestry, and is not
collected once its request has waited as long as it may:
- an update with no complete identity whose start was not seen;
- a run whose own thread is not whole yet, with the rest of its key
still to come;
- a run that names no thread under a run above it that is not known,
one that never came, was forgotten, or is known only from updates
that named no conversation of their own.
The receiver's memory records whether a run's start was seen, so an
update no longer proves one, and a dropped run leaves no conversation
behind for its start to take later. Inheritance stops at the first run
above that does not decide. An update that carries its conversation's
whole key still lands with no start seen, as every captured update
from the Python client does. A trace whose runs are all known and name
no thread still lands unassigned. A run whose conversation is known
still lands with whatever ancestry arrived.
The memory moves to version 2, so what it held is forgotten once. Each
pass that drops an arrival says how many, as runs_not_collected.
---
cmd/asz/langsmith_pass_test.go | 63 +++
cmd/asz/refresh.go | 12 +-
docs/en/adapters/langsmith.md | 45 +-
docs/en/changes/changes.md | 20 +-
internal/adapters/langsmith/collect.go | 186 ++++++---
internal/adapters/langsmith/hardening_test.go | 20 +
internal/adapters/langsmith/inherit_test.go | 174 +++++---
internal/adapters/langsmith/lost_test.go | 568 ++++++++++++++++++++++++++
internal/adapters/langsmith/pending.go | 73 +++-
9 files changed, 1015 insertions(+), 146 deletions(-)
diff --git a/cmd/asz/langsmith_pass_test.go b/cmd/asz/langsmith_pass_test.go
new file mode 100644
index 0000000..5583de7
--- /dev/null
+++ b/cmd/asz/langsmith_pass_test.go
@@ -0,0 +1,63 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package main
+
+import (
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/adapters/langsmith"
+ "github.com/apache/skywalking-ai-sessionizer/internal/config"
+ "github.com/apache/skywalking-ai-sessionizer/internal/storage"
+ "github.com/apache/skywalking-ai-sessionizer/internal/view"
+)
+
+// TestAPassSaysHowManyRunsItDidNotCollect. An update whose run's start never
+// came waits, and is dropped once its request has waited as long as it may.
+// That pass lands nothing and parses nothing, and still prints its line, with
+// the count. The quiet pass after it prints nothing.
+func TestAPassSaysHowManyRunsItDidNotCollect(t *testing.T) {
+ root := t.TempDir()
+ zone := storage.NewZone(root)
+ ads := []config.Adapter{{Name: config.AdapterLangSmithIngest, Enabled: true, Listen: "127.0.0.1:0"}}
+ ref, err := newRefresher(view.New(zone, nil), zone, ads, noMetrics, 2<<20, false)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(ref.close)
+ const run = "16161616-1616-7616-8616-161616161611"
+ update := `{"patch":[{"id":"` + run + `","trace_id":"` + run + `","dotted_order":"20260920T100000000000Z` + run + `",` +
+ `"end_time":"2026-09-20T10:00:03Z","outputs":{"answer":"no start came"}}]}`
+ if _, err := langsmith.NewInbox(zone).Put("application/json", "POST", "/runs/batch", []byte(update), time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ ref.pass() // The first pass always says what it did.
+ for i := 0; i < 2; i++ {
+ if out := stdoutOf(t, func() { ref.pass() }); out != "" {
+ t.Fatalf("a pass that only held the request printed %q", out)
+ }
+ }
+ out := stdoutOf(t, func() { ref.pass() })
+ if !strings.Contains(out, " runs_not_collected=1") {
+ t.Errorf("the pass that dropped the update printed %q, want runs_not_collected=1", out)
+ }
+ if out := stdoutOf(t, func() { ref.pass() }); out != "" {
+ t.Errorf("the quiet pass after it printed %q", out)
+ }
+}
diff --git a/cmd/asz/refresh.go b/cmd/asz/refresh.go
index 7be2de4..78fd5d3 100644
--- a/cmd/asz/refresh.go
+++ b/cmd/asz/refresh.go
@@ -520,6 +520,9 @@ func (r *refresher) pass() error {
var errs []error
var landed, records, sessionsSeen, pending, conflicts, busy int
var waiting, unreadable int
+ // dropped counts the LangSmith runs not collected this pass because their
+ // conversation could not be told.
+ var dropped int
changed := map[string]bool{}
// A removal an earlier pass left in progress is finished first, before
@@ -665,6 +668,7 @@ func (r *refresher) pass() error {
landed += landedRuns.Files
records += landedRuns.Records
sessionsSeen += len(landedRuns.Sessions)
+ dropped = landedRuns.Dropped
for _, id := range landedRuns.Sessions {
changed[id] = true
}
@@ -865,7 +869,7 @@ func (r *refresher) pass() error {
}
bodies := waiting != r.waiting || unreadable != r.unreadable
r.waiting, r.unreadable = waiting, unreadable
- if first || landed > 0 || rounds > 0 || busy > 0 || pushed != "" || len(errs) > 0 || bodies {
+ if first || landed > 0 || rounds > 0 || busy > 0 || pushed != "" || len(errs) > 0 || bodies || dropped > 0 {
contended := ""
if busy > 0 {
// Two writers on one root is a supported setup, so this is a
@@ -878,6 +882,12 @@ func (r *refresher) pass() error {
// body. Neither fails the pass, and both are said.
contended += fmt.Sprintf(" bodies_waiting=%d bodies_unreadable=%d", waiting, unreadable)
}
+ if dropped > 0 {
+ // Runs whose conversation was forgotten or never came, said every
+ // pass that drops one: none is an error, and each is evidence
+ // that is not kept.
+ contended += fmt.Sprintf(" runs_not_collected=%d", dropped)
+ }
fmt.Printf("[%s] refreshed: sessions=%d landed=%d records=%d rounds=%d%s%s (%s)\n",
now.Format("15:04:05"), sessionsSeen, landed, records, rounds, contended, pushed,
now.Sub(start).Round(time.Millisecond))
diff --git a/docs/en/adapters/langsmith.md b/docs/en/adapters/langsmith.md
index 198749e..998c37f 100644
--- a/docs/en/adapters/langsmith.md
+++ b/docs/en/adapters/langsmith.md
@@ -62,6 +62,15 @@ a run belonging to no conversation and of no kind. So the runs that have started
are remembered, and an update is completed from what its own start said. Nothing is invented:
every field comes from an earlier arrival about the same run.
+A run is remembered by when it was last seen, and forgotten 24 hours after that, or sooner when
+more than 50,000 runs are remembered, the oldest first. A run that keeps arriving stays remembered.
+An update that names no conversation of its own, whose start is not remembered because it was
+forgotten or never came, cannot be given back its conversation. It waits, as a run waits for its
+ancestry, in case its start is still on its way, and once it has waited as long as it may it is
+not collected. Each pass that drops an arrival says how many, as `runs_not_collected`. An update that
+carries the whole key names its conversation itself, and lands as a start does: all 11 updates in
+the captures, from the Python client, carry the project and the thread.
+
## The mapping
| Model | LangChain |
@@ -134,9 +143,10 @@ Placing such a run would be a guess, and the guess is always the parent lineage,
independent agent's model context and its answers to its caller. That is not a thin conversation
but a wrong one. So a request whose ancestry has not arrived waits: it is tried again at the end
of the same pass, because the batch that overtook it is very often already further down the same
-inbox, and then for up to three more passes. After that it is landed with whatever ancestry did
-arrive, because a trace that never completes must not hold its evidence out of the conversation
-for ever.
+inbox, and then for up to three more passes. After that a run whose conversation is known is
+landed with whatever ancestry did arrive, because a trace that never completes must not hold its
+evidence out of the conversation for ever. A run whose conversation is still not known is not
+collected, as [Ownership is a namespace](#ownership-is-a-namespace) says.
### A graph's own runs land trimmed
@@ -314,10 +324,14 @@ landed apart from the question it answered.
The order runs arrive in does not decide it. A run waits for its ancestry, as
[A nested agent is its own stream](#a-nested-agent-is-its-own-stream) says, and its conversation is
-decided when the ancestry has arrived. A run above it that supplied nothing, or landed unassigned,
-is passed over. A run above it that is not known, because it has not arrived when the run has
-waited as long as it may or was forgotten after 24 hours, stops the search: it may have supplied a
-thread of its own, so the one further up is not taken, and the run lands as one that supplied none.
+decided when the ancestry has arrived. A run above it whose start was seen and that supplied
+nothing is passed over. A run above it that is not known stops the search: one that has not
+arrived, one that was forgotten, and one known only from updates that did not name its whole
+conversation, whose start would have said.
+It may have supplied a thread of its own, so the one further up is not taken, and the run's
+conversation is not known. The run waits for it, and once it has waited as long as it may it is not
+collected. Landing it as a run that supplied none put it in a conversation of its own, where the
+run above may have named another.
A run's own key decides from the first of its arrivals that brings one, as long as the run has not
landed yet; once it has, every later arrival of the same run lands where it did, while the receiver
remembers the run, so one run is not split across two conversations.
@@ -326,12 +340,17 @@ An update carries only what changed, so it can name a thread with no project, an
run's start, which carries the project. A thread named before the run's project is known is not a
whole conversation. It does not decide while the run waits, so a later arrival that carries the
project still does. The run does not take a conversation from a run above it in place of its own
-thread, and a run below it does not pass over it to take one from further up. If it lands before
-any arrival carries the project, it lands under its thread with an empty project. All 11 updates in
-the captures carry their project with their metadata, so none of them has met this.
-
-A trace with no supplied key anywhere is not given one. It lands under a name that says the
-identity was not supplied, and nothing merges it with anything else.
+thread, and a run below it does not pass over it to take one from further up. If no arrival has
+carried the project once it has waited as long as it may, it is not collected. All 11 updates in
+the captures carry their project with their metadata, so none of them has met this. A dimension
+the metadata names, configured in `scope`, is missing the same way only from an update, which
+carries what changed. A start carries its whole metadata, so a dimension it leaves out is supplied
+empty.
+
+A trace with no supplied key anywhere is not given one. When every run of it is known, with its
+start seen, it lands under a name that says the identity was not supplied, and nothing merges it
+with anything else. A later update of one of its runs that carries a key stays there: one run is
+one conversation.
## What the wire cannot supply
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index 5948473..87c431d 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -29,6 +29,24 @@
## LangChain and LangGraph
+- A run whose conversation cannot be told because what would say it is gone is not collected. It
+ landed under a name saying its identity was not supplied, which put one conversation's evidence
+ in another. Two cases do this, and each waits first, as a run waits for its ancestry:
+ - an update that names no conversation of its own, whose start was forgotten, after a day or
+ past 50,000 remembered runs, or never came;
+ - a run that names no thread under a run above it that is not known: one that never came, was
+ forgotten, or is known only from updates that named no conversation of its own. It used to
+ take a thread from further up past such a run in the last case, although that run's start may
+ have named another. An update that named its run's whole conversation still gives it to the
+ runs below.
+
+ A run remembered only from its updates no longer proves that it named no thread. A thread named
+ before its project is known no longer lands with an empty project. An update that carries its
+ conversation's whole key still lands with no start seen, as every captured update from the Python
+ client does, and a trace whose runs are all known and name no thread still lands unassigned. Each
+ pass that drops an arrival says how many, as `runs_not_collected`. The receiver's memory of runs
+ moves to version 2, which records whether a run's start was seen, so every run it remembered is
+ forgotten once.
- Two records that could carry a system prompt unnamed are named. An unfinished model call whose
client put the prompt among its parameters, in the `extra` that lands whole with the envelope, is
named `system_prompt` by that shape, as its tools already were. A graph's first input whose
@@ -56,7 +74,7 @@
no answer. A run that supplies its own key keeps it, and a trace that supplied none anywhere still
lands unassigned. The conversation is decided once the run's ancestry has arrived, so the order
runs arrive in does not change it. A run above that is not known stops the search, and the run
- lands unassigned. A run's own key decides from the first of its arrivals that brings one before
+ waits for it, as the first entry here says. A run's own key decides from the first of its arrivals that brings one before
the run lands, and once it has landed its later arrivals follow it, while the receiver remembers
it. A thread an update names before the run's project is known does not decide while the run
waits, and the run neither takes a thread from above in its place nor passes one below.
diff --git a/internal/adapters/langsmith/collect.go b/internal/adapters/langsmith/collect.go
index fab5762..0b553f7 100644
--- a/internal/adapters/langsmith/collect.go
+++ b/internal/adapters/langsmith/collect.go
@@ -91,6 +91,12 @@ type Landed struct {
// Placed counts the requests landed after waiting as long as they may,
// with whatever ancestry did arrive.
Placed int
+ // Dropped counts the arrivals not collected because their run's
+ // conversation could not be told once their request had waited as long
+ // as it may: an update whose start was never seen or was forgotten, and a
+ // run whose conversation would come from a run above it that is not known.
+ // Two updates of one run count twice.
+ Dropped int
}
// Collect converts every waiting request, oldest first.
@@ -167,7 +173,8 @@ func (c *Collector) Collect() (Landed, error) {
}
// It has been tried again and the ancestry is still not
// here. It waits for another pass, and is landed anyway
- // once it has waited as long as it may.
+ // once it has waited as long as it may, without the runs
+ // whose conversation is still not known.
if open.held(name) {
out.Waiting++
if err := open.save(pendingPath, c.Now()); err != nil {
@@ -202,8 +209,8 @@ func (c *Collector) Collect() (Landed, error) {
// What this request taught is written down BEFORE the request
// is removed. Either order loses something if the two are not
// separated: saving afterwards loses a run's start to a crash in
- // between, and the bare update that follows it lands under a
- // name saying its identity was not supplied; saving beforehand
+ // between, and the bare update that follows it is not collected,
+ // its conversation no longer known; saving beforehand
// would lose a completion to a replay, if a completion were ever
// forgotten here. Nothing is: a run is forgotten only by age, so
// what is written is only ever more than it was.
@@ -221,6 +228,7 @@ func (c *Collector) Collect() (Landed, error) {
out.Files += landed.Files
out.Records += landed.Records
out.Unassigned += landed.Unassigned
+ out.Dropped += landed.Dropped
out.Bodies += landed.Bodies
out.BodyConflicts += landed.BodyConflicts
for _, s := range landed.Sessions {
@@ -245,12 +253,13 @@ type unreadable struct{ err error }
func (u *unreadable) Error() string { return u.err.Error() }
func (u *unreadable) Unwrap() error { return u.err }
-// unplaceable marks a request whose runs name ancestors that have not
-// arrived. It is not a failure: it is a request that is too early.
+// unplaceable marks a request that is too early: a run names an ancestor that
+// has not arrived, or its conversation waits for a start or an ancestor that
+// may still come. It is not a failure.
type unplaceable struct{ run string }
func (u *unplaceable) Error() string {
- return "run " + u.run + " names a run that has not arrived"
+ return "run " + u.run + " waits for a run that has not arrived"
}
// placed is one arrival with the records it became, kept together because the
@@ -285,8 +294,9 @@ func (c *Collector) convert(request Waiting, open *pending) (Landed, error) {
}
// convertAnyway is convert for a request that has waited as long as it may.
-// What is still unknown is placed in the parent lineage, which is what the
-// adapter page says happens and why.
+// A run whose place in its conversation is still unknown is placed in the
+// parent lineage, which is what the adapter page says happens and why. A run
+// whose conversation is still unknown is not collected.
func (c *Collector) convertAnyway(request Waiting, open *pending) (Landed, error) {
return c.convertWith(request, open, false)
}
@@ -313,8 +323,8 @@ func (c *Collector) convertWith(request Waiting, open *pending, wait bool) (Land
// decides for a first that named none, as it does when it waits in a
// request of its own. unsupplied marks a run that named none.
unsupplied := make([]bool, len(operations))
- // partial marks a run that named a thread before its project was known,
- // in this arrival or an earlier one; see pendingRun.Partial.
+ // partial marks a run that named a thread before the rest of its key was
+ // known, in this arrival or an earlier one; see pendingRun.Partial.
partial := make([]bool, len(operations))
follows := make([]int, len(operations))
first := map[string]int{}
@@ -340,7 +350,7 @@ func (c *Collector) convertWith(request Waiting, open *pending, wait bool) (Land
follows[i], decides = j, j
if unsupplied[j] || partial[j] {
if session, missing := c.sessionOf(op, completed); !missing {
- sessions[j], unsupplied[j], partial[j] = session, false, c.projectUnknown(completed)
+ sessions[j], unsupplied[j], partial[j] = session, false, c.partialOf(op, completed)
}
}
default:
@@ -349,7 +359,7 @@ func (c *Collector) convertWith(request Waiting, open *pending, wait bool) (Land
if unsupplied[i] {
partial[i] = known.Partial
} else {
- partial[i] = c.projectUnknown(completed)
+ partial[i] = c.partialOf(op, completed)
}
}
// A run that supplied no conversation is not remembered in one yet.
@@ -372,47 +382,73 @@ func (c *Collector) convertWith(request Waiting, open *pending, wait bool) (Land
if partial[decides] {
open.partial(op.RunID)
}
- open.sawRun(completed, c.Now())
- }
-
- // Every run of this request is now remembered, so an ancestor still
- // missing is one that has not arrived at all.
- if wait {
- for i := range operations {
- if !open.placeable(runs[i]) {
- return Landed{}, &unplaceable{run: runs[i].ID}
- }
+ // A start can follow its update in the same request. Every start is
+ // recorded here, before any conversation is decided.
+ if op.Op == "post" {
+ open.created(op.RunID)
}
+ open.sawRun(completed, c.Now())
}
+ // Which conversation each run belongs to, decided once for its first
+ // arrival in this request, which a later arrival of it follows. Every run
+ // of the request is now remembered, so a run above one that is still
+ // missing has not arrived at all.
+ //
// A run that supplied no conversation takes the one the nearest run it
// ran inside supplied, as pending.inherited reads its dotted order.
// LangSmith JS passes a decorated function's metadata to none of the
// runs inside it, so a model call the function makes names no thread,
// although the function names one and the call's dotted order says it
- // ran there. Only when no run above it supplied one does it land
- // unassigned, under its trace. The count is of runs, not arrivals.
+ // ran there. It lands unassigned, under its trace, only when every run
+ // above it is known and none supplied one. The count is of runs, not
+ // arrivals.
+ //
+ // A conversation that depends on something not yet seen is open: the
+ // start of an update that names none, the rest of the key for a thread
+ // named before it, or a run above that would decide. Each may still come, as
+ // a missing ancestor may, so the request waits for it. Once the request
+ // has waited as long as it may, a run whose conversation is still open
+ // is dropped: what would say it is gone, and guessing put one
+ // conversation's evidence in another.
+ undecided := make([]bool, len(operations))
unassigned := 0
- for i := range operations {
- if !unsupplied[i] {
- continue
- }
- // An earlier arrival named a thread of its own, with no project,
- // and this one names none. Its own is not replaced by another's: it
- // lands unassigned, as it did before a thread was taken from above.
- if partial[i] {
- unassigned++
+ for i, op := range operations {
+ if follows[i] >= 0 {
continue
}
- if session := open.inherited(runs[i]); session != "" {
- sessions[i] = session
- continue
+ switch {
+ case !unsupplied[i] && !partial[i]:
+ case partial[i]:
+ // A thread of its own, with the rest of its key not known yet.
+ // Its own is not replaced by another's, so it neither inherits nor
+ // lands unassigned. It waits for what completes it.
+ undecided[i] = true
+ case op.Op != "post" && !open.Runs[op.RunID].Create:
+ // An update that names no conversation, whose start has not been
+ // seen. The start says where it belongs.
+ undecided[i] = true
+ default:
+ switch session, by := open.inherited(runs[i]); by {
+ case inheritedResolved:
+ sessions[i] = session
+ case inheritedNone:
+ unassigned++
+ default:
+ undecided[i] = true
+ }
}
- unassigned++
}
for i, j := range follows {
if j >= 0 {
- sessions[i] = sessions[j]
+ sessions[i], undecided[i] = sessions[j], undecided[j]
+ }
+ }
+ if wait {
+ for i := range operations {
+ if undecided[i] || !open.placeable(runs[i]) {
+ return Landed{}, &unplaceable{run: runs[i].ID}
+ }
}
}
@@ -440,6 +476,12 @@ func (c *Collector) convertWith(request Waiting, open *pending, wait bool) (Land
for i, op := range operations {
envelope := runs[i]
session := sessions[i]
+ if undecided[i] {
+ // Nor is it remembered in the conversation it fell back to: its
+ // start, if it comes later, would take that one over its own.
+ out.Dropped++
+ continue
+ }
// A run may be updated after it ends as well as before, and that
// update carries none of this. So a run is remembered whether or not
// it has finished, and is forgotten only by age. Forgetting it here
@@ -508,23 +550,7 @@ func (c *Collector) convertWith(request Waiting, open *pending, wait bool) (Land
// in a session whose name says the identity was not supplied, and the second
// return says so. The caller then looks for a run above it that supplied one.
func (c *Collector) sessionOf(op Operation, envelope Run) (string, bool) {
- // The client splits extra out of the envelope only when it is large
- // enough to be worth a part of its own, so both places have to be read.
- // Reading the envelope's extra as though it were already the metadata
- // was the bug this comment exists to keep fixed: it put the whole extra
- // object where the metadata belonged, no supplied key was ever found,
- // and every conversation landed as one whose identity was not supplied.
- var extra struct {
- Metadata map[string]any `json:"metadata"`
- }
- if raw, ok := op.Fields["extra"]; ok {
- _ = json.Unmarshal(raw, &extra)
- } else {
- var inline map[string]any
- unmarshalInto(op.Envelope, "extra", &inline)
- extra.Metadata, _ = inline["metadata"].(map[string]any)
- }
- owner, ok := c.Ownership.Owner(envelope.Session, extra.Metadata)
+ owner, ok := c.Ownership.Owner(envelope.Session, c.metadataOf(op))
if !ok {
// Its own trace, or its own run when it has no trace either. A
// client can create a run with neither, and naming them all after
@@ -534,11 +560,51 @@ func (c *Collector) sessionOf(op Operation, envelope Run) (string, bool) {
return StorageID(owner), false
}
-// projectUnknown reports whether a run's conversation would be named before
-// its project is known, when the project is one of the dimensions that own
-// it. An update names no project, and can arrive before the run's start.
-func (c *Collector) projectUnknown(run Run) bool {
- return run.Session == "" && slices.Contains(c.Ownership.Scope, "project")
+// metadataOf is an arrival's run metadata.
+//
+// The client splits extra out of the envelope only when it is large enough to
+// be worth a part of its own, so both places have to be read. Reading the
+// envelope's extra as though it were already the metadata was the bug this
+// comment exists to keep fixed: it put the whole extra object where the
+// metadata belonged, no supplied key was ever found, and every conversation
+// landed as one whose identity was not supplied.
+func (c *Collector) metadataOf(op Operation) map[string]any {
+ if raw, ok := op.Fields["extra"]; ok {
+ var extra struct {
+ Metadata map[string]any `json:"metadata"`
+ }
+ _ = json.Unmarshal(raw, &extra)
+ return extra.Metadata
+ }
+ var inline map[string]any
+ unmarshalInto(op.Envelope, "extra", &inline)
+ metadata, _ := inline["metadata"].(map[string]any)
+ return metadata
+}
+
+// partialOf reports whether an arrival names a thread but not the rest of its
+// conversation's key. The project is missing when no arrival of the run has
+// carried it yet: an update can name none, and can arrive before the run's
+// start. A metadata dimension is missing only from an update, which carries
+// what changed. A start carries its whole metadata, so a dimension it leaves
+// out is supplied empty.
+func (c *Collector) partialOf(op Operation, run Run) bool {
+ if run.Session == "" && slices.Contains(c.Ownership.Scope, "project") {
+ return true
+ }
+ if op.Op == "post" {
+ return false
+ }
+ metadata := c.metadataOf(op)
+ for _, dimension := range c.Ownership.Scope {
+ if dimension == "project" || dimension == "thread" {
+ continue
+ }
+ if s, _ := metadata[dimension].(string); s == "" {
+ return true
+ }
+ }
+ return false
}
// traceOrRun is what identifies a run that supplied no conversation: its
diff --git a/internal/adapters/langsmith/hardening_test.go b/internal/adapters/langsmith/hardening_test.go
index 0b56867..28d37bb 100644
--- a/internal/adapters/langsmith/hardening_test.go
+++ b/internal/adapters/langsmith/hardening_test.go
@@ -793,6 +793,26 @@ func mentions(t *testing.T, zone *storage.Zone, session, text string) bool {
return false
}
+// landedAnywhere reports whether any landed file of any session holds the
+// text, or whether a session of that name exists.
+func landedAnywhere(t *testing.T, zone *storage.Zone, text string) bool {
+ t.Helper()
+ items, err := os.ReadDir(zone.Root())
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, it := range items {
+ name := it.Name()
+ if !it.IsDir() || strings.HasPrefix(name, "_") || strings.HasPrefix(name, ".") {
+ continue
+ }
+ if name == text || mentions(t, zone, name, text) {
+ return true
+ }
+ }
+ return false
+}
+
// TestAChildThatOvertookItsParentIsNotGivenToIt.
//
// The client drains its queue with several threads at once, so a batch can
diff --git a/internal/adapters/langsmith/inherit_test.go b/internal/adapters/langsmith/inherit_test.go
index c427b89..929603a 100644
--- a/internal/adapters/langsmith/inherit_test.go
+++ b/internal/adapters/langsmith/inherit_test.go
@@ -145,22 +145,30 @@ func TestATraceThatSuppliedNoThreadIsNotGivenOne(t *testing.T) {
}
}
-// TestARunLandedUnassignedStaysWhereItLanded. A call whose function never
-// arrived while it waited is landed under its trace. When the function and
-// an update to the call arrive later, the update stays with the call's first
-// arrival: one run is never split across two conversations.
-func TestARunLandedUnassignedStaysWhereItLanded(t *testing.T) {
+// TestARunWhoseRunAboveNeverCameIsNotCollected. A call that names no thread
+// waits for the function it ran inside, which would say its conversation.
+// When the function has not come once the call has waited as long as it may,
+// the call is not collected: landing it under its trace put it in a
+// conversation of its own, although the function may have named another, a
+// retired one too. When the function and an update to the call arrive later,
+// the update lands in the function's conversation, and nothing of the call is
+// left unassigned.
+func TestARunWhoseRunAboveNeverCameIsNotCollected(t *testing.T) {
_, zone, base := started(t, "")
collector := collectorOver(zone)
start := strings.Replace(inheritCall(``), `"end_time":"2026-09-20T10:00:02Z",`, ``, 1)
postBatch(t, base, `{"post":[`+start+`]}`)
+ dropped := 0
for pass := 0; pass <= deferLimit; pass++ {
- if _, err := collector.Collect(); err != nil {
+ landed, err := collector.Collect()
+ if err != nil {
t.Fatal(err)
}
+ dropped += landed.Dropped
}
- if !mentions(t, zone, UnassignedID(inheritRoot), inheritChild) {
- t.Fatal("the call was not landed under its trace once it had waited as long as it may")
+ if dropped != 1 || landedAnywhere(t, zone, inheritChild) {
+ t.Fatalf("%d dropped, landed %v: the call must not be collected once it has waited as long as it may",
+ dropped, landedAnywhere(t, zone, inheritChild))
}
update := `{"id":"` + inheritChild + `","trace_id":"` + inheritRoot + `",` +
`"dotted_order":"` + inheritOrder + `.20260920T100001000000Z` + inheritChild + `",` +
@@ -170,17 +178,21 @@ func TestARunLandedUnassignedStaysWhereItLanded(t *testing.T) {
if _, err := collector.Collect(); err != nil {
t.Fatal(err)
}
- if !mentions(t, zone, UnassignedID(inheritRoot), "the late answer") {
- t.Error("the call's update left the conversation its start landed in")
+ if !mentions(t, zone, StorageID(Owner{Values: []string{"asz", "t-late"}, Thread: "t-late"}), "the late answer") {
+ t.Error("the call's update is not in the function's conversation")
+ }
+ if landedAnywhere(t, zone, UnassignedID(inheritRoot)) || mentions(t, zone, UnassignedID(inheritRoot), "the late answer") {
+ t.Error("something of the call landed unassigned")
}
}
-// TestAnUnassignedRunAboveIsPassedOver. The step a call ran inside supplied
-// no thread and landed unassigned, because the function above it did not
-// arrive while it waited. The function supplied a thread when it came, and the
-// call arriving with it takes that thread: the step's unassigned name was
-// never supplied by anyone, so it is not inherited.
-func TestAnUnassignedRunAboveIsPassedOver(t *testing.T) {
+// TestADroppedRunAboveIsPassedOver. The step a call ran inside supplied no
+// thread and was not collected, because the function above it did not arrive
+// while it waited. Its start was seen and named no thread, so when the
+// function supplies one, the call arriving with it takes that thread past the
+// step. The conversation the step fell back to was never remembered as its
+// own, so it is not inherited.
+func TestADroppedRunAboveIsPassedOver(t *testing.T) {
_, zone, base := started(t, "")
collector := collectorOver(zone)
const step = "14141414-1414-7414-8414-141414141412"
@@ -188,13 +200,8 @@ func TestAnUnassignedRunAboveIsPassedOver(t *testing.T) {
postBatch(t, base, `{"post":[{"id":"`+step+`","trace_id":"`+inheritRoot+`","parent_run_id":"`+inheritRoot+`",`+
`"dotted_order":"`+stepOrder+`","run_type":"chain","name":"step","start_time":"2026-09-20T10:00:00.5Z",`+
`"end_time":"2026-09-20T10:00:02.5Z","session_name":"asz","extra":{"metadata":{}},"outputs":{"done":true}}]}`)
- for pass := 0; pass <= deferLimit; pass++ {
- if _, err := collector.Collect(); err != nil {
- t.Fatal(err)
- }
- }
- if !mentions(t, zone, UnassignedID(inheritRoot), step) {
- t.Fatal("the step was not landed under its trace once it had waited as long as it may")
+ if dropped := passes(t, collector, deferLimit+1); dropped != 1 || !inboxEmpty(t, zone) || landedAnywhere(t, zone, step) {
+ t.Fatalf("%d dropped: the step must not be collected once it has waited as long as it may", dropped)
}
call := strings.Replace(inheritCall(``), `"parent_run_id":"`+inheritRoot+`","dotted_order":"`+inheritOrder,
`"parent_run_id":"`+step+`","dotted_order":"`+stepOrder, 1)
@@ -209,6 +216,9 @@ func TestAnUnassignedRunAboveIsPassedOver(t *testing.T) {
if !mentions(t, zone, landed.Sessions[0], "the call's own answer") {
t.Error("the call's answer is not in the function's conversation")
}
+ if landedAnywhere(t, zone, `"name":"step"`) {
+ t.Error("the step was collected once the function came, although it was dropped")
+ }
}
// TestOneRunInOneRequestIsOneConversation. A batch carries the call's start,
@@ -254,12 +264,14 @@ func TestOneRunInOneRequestIsOneConversation(t *testing.T) {
}
}
-// TestARunAboveThatArrivedWithNoKindIsStillRead. An update arrives for a run
-// whose start never came, with no kind, and between the call and the function
-// that supplied the thread. It supplied none, so it is passed over, as it is
-// when it landed in an earlier request; it is not taken for a run that has not
-// arrived, which would stop the walk.
-func TestARunAboveThatArrivedWithNoKindIsStillRead(t *testing.T) {
+// TestARunAboveKnownOnlyFromAnUpdateStopsInheriting. An update arrives for a
+// run whose start never came, with no kind and no thread, between the call
+// and the function that supplied one. Its start would have said whether it
+// named a thread of its own, so the call cannot take the function's past it.
+// The request waits for that start. Once it has waited as long as it may, the
+// function lands in its conversation, and the call and the update, whose
+// conversations cannot be told, are not collected.
+func TestARunAboveKnownOnlyFromAnUpdateStopsInheriting(t *testing.T) {
_, zone, base := started(t, "")
const step = "14141414-1414-7414-8414-141414141413"
stepOrder := inheritOrder + ".20260920T100000500000Z" + step
@@ -267,12 +279,27 @@ func TestARunAboveThatArrivedWithNoKindIsStillRead(t *testing.T) {
`"end_time":"2026-09-20T10:00:02.5Z","outputs":{"done":true}}`
call := strings.Replace(inheritCall(``), `"dotted_order":"`+inheritOrder, `"dotted_order":"`+stepOrder, 1)
postBatch(t, base, `{"post":[`+inheritFunction("t-kind")+`,`+call+`],"patch":[`+update+`]}`)
- landed, err := collectorOver(zone).Collect()
+ collector := collectorOver(zone)
+ landed, err := collector.Collect()
if err != nil {
t.Fatal(err)
}
- if landed.Unassigned != 0 || !mentions(t, zone, StorageID(Owner{Values: []string{"asz", "t-kind"}, Thread: "t-kind"}), "the call's own answer") {
- t.Fatalf("sessions %v, %d unassigned, want the call in the function's conversation", landed.Sessions, landed.Unassigned)
+ if landed.Waiting != 1 || landedAnywhere(t, zone, "the call's own answer") {
+ t.Fatalf("%d waiting: the request must wait for the step's start", landed.Waiting)
+ }
+ dropped := 0
+ for pass := 0; pass < deferLimit; pass++ {
+ if landed, err = collector.Collect(); err != nil {
+ t.Fatal(err)
+ }
+ dropped += landed.Dropped
+ }
+ function := StorageID(Owner{Values: []string{"asz", "t-kind"}, Thread: "t-kind"})
+ if !mentions(t, zone, function, "what changed in the docs") {
+ t.Error("the function is not in its own conversation")
+ }
+ if dropped != 2 || landedAnywhere(t, zone, "the call's own answer") || landedAnywhere(t, zone, step) {
+ t.Errorf("%d dropped: the call and the step's update must not be collected", dropped)
}
}
@@ -307,26 +334,37 @@ func TestARunThatSuppliedAThreadWithNoKindIsRemembered(t *testing.T) {
}
// TestInheritingStopsAtARunNotKnown. The run between the call and the
-// function was forgotten, or has not arrived. It may have supplied a thread of
-// its own, so the function's further up is not taken over it. A run between
-// that supplied nothing, by either unassigned name, is passed over.
+// function was forgotten, or has not arrived, or is known only from updates.
+// It may have supplied a thread of its own, so the function's further up is
+// not taken over it, and the call's conversation is not known. A run between
+// whose start was seen and that supplied nothing, by either unassigned name,
+// is passed over.
func TestInheritingStopsAtARunNotKnown(t *testing.T) {
const step = "14141414-1414-7414-8414-141414141415"
call := Run{ID: inheritChild, TraceID: inheritRoot,
Dotted: inheritOrder + ".20260920T100000500000Z" + step + ".20260920T100001000000Z" + inheritChild}
- memory := &pending{Runs: map[string]pendingRun{inheritRoot: {Trace: inheritRoot, Session: "ls-function"}}}
- if got := memory.inherited(call); got != "" {
- t.Errorf("took %q past a run that is not known", got)
+ memory := &pending{Runs: map[string]pendingRun{inheritRoot: {Trace: inheritRoot, Session: "ls-function", Create: true}}}
+ if got, by := memory.inherited(call); got != "" || by != inheritedUnknown {
+ t.Errorf("took %q (%d) past a run that is not known", got, by)
+ }
+ memory.Runs[step] = pendingRun{Trace: inheritRoot, Session: UnassignedID(inheritRoot)}
+ if got, by := memory.inherited(call); got != "" || by != inheritedUnknown {
+ t.Errorf("took %q (%d) past a run known only from its updates", got, by)
}
for _, name := range []string{UnassignedID(inheritRoot), UnassignedID(step)} {
- memory.Runs[step] = pendingRun{Trace: inheritRoot, Session: name}
- if got := memory.inherited(call); got != "ls-function" {
- t.Errorf("a run between landed as %q: took %q, want the function's", name, got)
+ memory.Runs[step] = pendingRun{Trace: inheritRoot, Session: name, Create: true}
+ if got, by := memory.inherited(call); got != "ls-function" || by != inheritedResolved {
+ t.Errorf("a run between landed as %q: took %q (%d), want the function's", name, got, by)
}
}
memory.Runs[step] = pendingRun{Trace: inheritRoot, Session: "ls-step"}
- if got := memory.inherited(call); got != "ls-step" {
- t.Errorf("took %q, want the nearest supplied, the step's", got)
+ if got, by := memory.inherited(call); got != "ls-step" || by != inheritedResolved {
+ t.Errorf("took %q (%d), want the nearest supplied, the step's", got, by)
+ }
+ memory.Runs[inheritRoot] = pendingRun{Trace: inheritRoot, Session: UnassignedID(inheritRoot), Create: true}
+ memory.Runs[step] = pendingRun{Trace: inheritRoot, Create: true}
+ if got, by := memory.inherited(call); got != "" || by != inheritedNone {
+ t.Errorf("took %q (%d), want none: every run above is known and supplied nothing", got, by)
}
}
@@ -346,12 +384,12 @@ func landInOrder(t *testing.T, bodies ...string) *storage.Zone {
}
// TestTheOrderOfRequestsDoesNotDecide. The same runs, sent as requests in
-// different orders and batched differently, land the same. A step with no
-// kind and no thread sits between the call and the function: while its
-// request waited it was not remembered, and the call, placed first, stopped
-// at it as at a run that had not arrived. And a call's start that names no
-// thread, with an update that names its own: the update decided when the two
-// waited apart, and the start when they came together.
+// different orders and batched differently, land the same. A step known only
+// from an update, with no kind and no thread, sits between the call and the
+// function: in every order the call waits for the step's start, and is not
+// collected once it has waited as long as it may. And a call's start that
+// names no thread, with an update that names its own: the update decided
+// when the two waited apart, and the start when they came together.
func TestTheOrderOfRequestsDoesNotDecide(t *testing.T) {
const step = "14141414-1414-7414-8414-141414141416"
stepOrder := inheritOrder + ".20260920T100000500000Z" + step
@@ -366,8 +404,17 @@ func TestTheOrderOfRequestsDoesNotDecide(t *testing.T) {
"call and step, function": {`{"post":[` + call + `],"patch":[` + stepUpdate + `]}`, function},
} {
zone := landInOrder(t, bodies...)
- if !mentions(t, zone, thread, "the call's own answer") {
- t.Errorf("%s: the call is not in the function's conversation", name)
+ if landedAnywhere(t, zone, "the call's own answer") {
+ t.Errorf("%s: the call landed before the step's start came", name)
+ }
+ if dropped := passes(t, collectorOver(zone), deferLimit+1); dropped != 2 || !inboxEmpty(t, zone) {
+ t.Errorf("%s: %d dropped, want the call and the step's update", name, dropped)
+ }
+ if !mentions(t, zone, thread, "what changed in the docs") {
+ t.Errorf("%s: the function is not in its conversation", name)
+ }
+ if landedAnywhere(t, zone, "the call's own answer") {
+ t.Errorf("%s: the call was collected, although the step above it is known only from an update", name)
}
}
@@ -457,4 +504,29 @@ func TestAThreadNamedBeforeItsProjectIsNotWhole(t *testing.T) {
t.Errorf("%s landed in the function's conversation, although the call named a thread of its own", text)
}
}
+ // The start brought the project, and the update, still waiting for it, is
+ // read again with it: the call's thread is now whole, so the call and the
+ // run below it land there, where they used to land unassigned.
+ for _, text := range []string{"the patched answer", "the tool's own result"} {
+ if !mentions(t, zone, own, text) {
+ t.Errorf("%s is not in the call's own conversation", text)
+ }
+ }
+
+ // The same with no run below the call, so nothing else in the start's
+ // request waits: the call itself waits for its own thread to be whole,
+ // rather than taking the function's.
+ _, zone, base = started(t, "")
+ collector = collectorOver(zone)
+ postBatch(t, base, `{"patch":[`+update+`]}`)
+ if _, err := collector.Collect(); err != nil {
+ t.Fatal(err)
+ }
+ postBatch(t, base, `{"post":[`+inheritFunction("t-outer")+`,`+keyless+`]}`)
+ if _, err := collector.Collect(); err != nil {
+ t.Fatal(err)
+ }
+ if mentions(t, zone, outer, `"FakeListChatModel"`) || !mentions(t, zone, own, `"FakeListChatModel"`) {
+ t.Error("alone: the call took the function's thread in place of its own")
+ }
}
diff --git a/internal/adapters/langsmith/lost_test.go b/internal/adapters/langsmith/lost_test.go
new file mode 100644
index 0000000..effb1ee
--- /dev/null
+++ b/internal/adapters/langsmith/lost_test.go
@@ -0,0 +1,568 @@
+// 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 langsmith
+
+import (
+ "os"
+ "path/filepath"
+ "strconv"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/storage"
+)
+
+// A run whose conversation cannot be told, because what would say it was
+// forgotten or never came, is not collected once its request has waited as
+// long as it may. These tests use the function and call of inherit_test.go,
+// and three runs of their own.
+const (
+ lostStep = "15151515-1515-7515-8515-151515151511"
+ lostBelow = "15151515-1515-7515-8515-151515151512"
+ lostOther = "15151515-1515-7515-8515-151515151513"
+)
+
+var lostCallOrder = inheritOrder + ".20260920T100001000000Z" + inheritChild
+
+// lostClock is a collector's clock a test moves by hand, so what is
+// remembered can be forgotten by age.
+type lostClock struct{ at time.Time }
+
+func (c *lostClock) now() time.Time {
+ c.at = c.at.Add(time.Second)
+ return c.at
+}
+
+func clockedOver(zone *storage.Zone) (*Collector, *lostClock) {
+ clock := &lostClock{at: time.Date(2026, 9, 20, 11, 0, 0, 0, time.UTC)}
+ return &Collector{Zone: zone, Now: clock.now}, clock
+}
+
+// sparse is an update as update_run sends it: the id, the trace, the dotted
+// order, the end and the outputs, and nothing that names a conversation. The
+// outputs are a model call's, so a call's update keeps its text; a chain's
+// is not kept, and a test finds it by its record id instead.
+func sparse(id, trace, order, answer string) string {
+ return `{"id":"` + id + `","trace_id":"` + trace + `","dotted_order":"` + order + `",` +
+ `"end_time":"2026-09-20T10:00:02Z",` +
+ `"outputs":{"generations":[[{"message":{"kwargs":{"content":"` + answer + `"}}}]]}}`
+}
+
+// updateOf is what a landed update of a run is named by.
+func updateOf(id string) string { return id + ":patch:" }
+
+// otherRun is a run of a trace of its own. A request holding it touches
+// nothing a test reads, and its save forgets what is old enough.
+func otherRun(thread string) string {
+ order := "20260920T110000000000Z" + lostOther
+ return `{"id":"` + lostOther + `","trace_id":"` + lostOther + `","dotted_order":"` + order + `",` +
+ `"run_type":"chain","name":"other","start_time":"2026-09-20T11:00:00Z","end_time":"2026-09-20T11:00:01Z",` +
+ `"session_name":"asz","extra":{"metadata":{"thread_id":"` + thread + `"},"runtime":{"library":"langsmith"}},` +
+ `"outputs":{"answer":"elsewhere"}}`
+}
+
+// rootRun is a run at the root of a trace of its own, with the metadata
+// given and the project when one is given. It is a traced function, whose
+// outputs are kept, where a graph's own run keeps none.
+func rootRun(id, project, metadata, answer string) string {
+ named := ""
+ if project != "" {
+ named = `"session_name":"` + project + `",`
+ }
+ return `{"id":"` + id + `","trace_id":"` + id + `","dotted_order":"20260920T100000000000Z` + id + `",` +
+ `"run_type":"chain","name":"root","start_time":"2026-09-20T10:00:00Z","end_time":"2026-09-20T10:00:03Z",` +
+ named + `"extra":{"metadata":{` + metadata + `},"runtime":{"library":"langsmith"}},"outputs":{"answer":"` + answer + `"}}`
+}
+
+func threadOf(name string) string {
+ return StorageID(Owner{Values: []string{"asz", name}, Thread: name})
+}
+
+// passes collects n times and adds up what was dropped.
+func passes(t *testing.T, c *Collector, n int) int {
+ t.Helper()
+ dropped := 0
+ for i := 0; i < n; i++ {
+ landed, err := c.Collect()
+ if err != nil {
+ t.Fatal(err)
+ }
+ dropped += landed.Dropped
+ }
+ return dropped
+}
+
+// TestASparseUpdateAfterItsStartWasForgottenIsNotCollected. The call's start
+// landed. A day later what was remembered of it is forgotten, and an update
+// that names no conversation can no longer be given back its own. It waits,
+// in case its start comes again, and is then not collected.
+func TestASparseUpdateAfterItsStartWasForgottenIsNotCollected(t *testing.T) {
+ _, zone, base := started(t, "")
+ c, clock := clockedOver(zone)
+ start := strings.Replace(inheritCall(``), `"end_time":"2026-09-20T10:00:02Z",`, ``, 1)
+ postBatch(t, base, `{"post":[`+inheritFunction("t-lost")+`,`+start+`]}`)
+ passes(t, c, 1)
+ if !mentions(t, zone, threadOf("t-lost"), `"FakeListChatModel"`) {
+ t.Fatal("the call's start did not land in the function's conversation")
+ }
+ clock.at = clock.at.Add(25 * time.Hour)
+ postBatch(t, base, `{"post":[`+otherRun("t-other")+`]}`)
+ postBatch(t, base, `{"patch":[`+sparse(inheritChild, inheritRoot, lostCallOrder, "the forgotten end")+`]}`)
+ if dropped := passes(t, c, deferLimit+1); dropped != 1 || landedAnywhere(t, zone, "the forgotten end") {
+ t.Errorf("%d dropped: an update whose start was forgotten must not be collected", dropped)
+ }
+}
+
+// TestACompleteUpdateLandsWithoutItsStart. The Python client's update
+// carries the project and the thread, so it names its conversation itself.
+// It lands in the first pass, with no start ever seen, and holds back no run
+// of its request.
+func TestACompleteUpdateLandsWithoutItsStart(t *testing.T) {
+ _, zone, base := started(t, "")
+ whole := rootRun(lostStep, "asz", `"thread_id":"t-whole"`, "a whole update")
+ postBatch(t, base, `{"post":[`+otherRun("t-other")+`],"patch":[`+whole+`]}`)
+ landed, err := collectorOver(zone).Collect()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if landed.Waiting != 0 || landed.Dropped != 0 {
+ t.Fatalf("%d waiting, %d dropped: a complete update needs no start", landed.Waiting, landed.Dropped)
+ }
+ if !mentions(t, zone, threadOf("t-whole"), "a whole update") || !mentions(t, zone, threadOf("t-other"), "elsewhere") {
+ t.Error("the update or the run beside it is not in its conversation")
+ }
+}
+
+// TestARunKeptByItsUpdatesIsRememberedPastADay. A run is remembered by when
+// it was last seen, not by when it started. Updates that keep arriving keep
+// it, and one 40 hours after its start still lands in its conversation.
+func TestARunKeptByItsUpdatesIsRememberedPastADay(t *testing.T) {
+ _, zone, base := started(t, "")
+ c, clock := clockedOver(zone)
+ start := strings.Replace(inheritCall(``), `"end_time":"2026-09-20T10:00:02Z",`, ``, 1)
+ postBatch(t, base, `{"post":[`+inheritFunction("t-kept")+`,`+start+`]}`)
+ passes(t, c, 1)
+ clock.at = clock.at.Add(20 * time.Hour)
+ postBatch(t, base, `{"patch":[`+sparse(inheritChild, inheritRoot, lostCallOrder, "still going")+`]}`)
+ passes(t, c, 1)
+ clock.at = clock.at.Add(20 * time.Hour)
+ postBatch(t, base, `{"post":[`+otherRun("t-other")+`]}`)
+ postBatch(t, base, `{"patch":[`+sparse(inheritChild, inheritRoot, lostCallOrder, "done after 40 hours")+`]}`)
+ if dropped := passes(t, c, 1); dropped != 0 {
+ t.Fatalf("%d dropped: the call was seen within the day", dropped)
+ }
+ for _, text := range []string{"still going", "done after 40 hours"} {
+ if !mentions(t, zone, threadOf("t-kept"), text) {
+ t.Errorf("%q is not in the call's conversation", text)
+ }
+ }
+}
+
+// TestASparseUpdateWhoseStartNeverCameIsNotCollected. Updates of a run whose
+// start never came, two in one request, one in a request of its own, and one
+// after a restart. None of them proves a start, so none is collected, and the
+// run stays remembered as one whose start was not seen.
+func TestASparseUpdateWhoseStartNeverCameIsNotCollected(t *testing.T) {
+ _, zone, base := started(t, "")
+ order := "20260920T100000000000Z" + lostStep
+ postBatch(t, base, `{"patch":[`+sparse(lostStep, lostStep, order, "first")+`,`+sparse(lostStep, lostStep, order, "second")+`]}`)
+ postBatch(t, base, `{"patch":[`+sparse(lostStep, lostStep, order, "third")+`]}`)
+ dropped := passes(t, collectorOver(zone), deferLimit+1)
+ postBatch(t, base, `{"patch":[`+sparse(lostStep, lostStep, order, "after a restart")+`]}`)
+ dropped += passes(t, collectorOver(zone), deferLimit+1)
+ if dropped != 4 {
+ t.Errorf("%d dropped, want all four updates", dropped)
+ }
+ if landedAnywhere(t, zone, lostStep) {
+ t.Error("an update was collected, although the run's start never came")
+ }
+ open := loadPending(filepath.Join(zone.Root(), InboxDir, pendingFile))
+ if known, ok := open.Runs[lostStep]; !ok || known.Create || known.Session != "" {
+ t.Errorf("remembered %+v: the run must be remembered with no start seen and no conversation", known)
+ }
+}
+
+// TestASparseUpdateBeforeItsStartInOneRequestLands. A request keeps the order
+// the client wrote, and the update can come before the start. The start is in
+// the request, so it is seen, and the update lands with it.
+func TestASparseUpdateBeforeItsStartInOneRequestLands(t *testing.T) {
+ _, zone, base := started(t, "")
+ start := strings.Replace(inheritCall(``), `"end_time":"2026-09-20T10:00:02Z",`, ``, 1)
+ body, contentType := multipartOf([][2]string{
+ {"patch." + inheritChild, sparse(inheritChild, inheritRoot, lostCallOrder, "ended first")},
+ {"post." + inheritChild, start},
+ {"post." + inheritRoot, inheritFunction("t-together")},
+ })
+ resp := send(t, base, "/runs/multipart", contentType, body, "")
+ resp.Body.Close()
+ landed, err := collectorOver(zone).Collect()
+ if err != nil {
+ t.Fatal(err)
+ }
+ // The update, read before the start, names no kind and lands as the run's
+ // data, so it is found by its record id.
+ if landed.Dropped != 0 || !mentions(t, zone, threadOf("t-together"), updateOf(inheritChild)) ||
+ !mentions(t, zone, threadOf("t-together"), `"FakeListChatModel"`) {
+ t.Errorf("%d dropped: the update and its start are not both in the function's conversation", landed.Dropped)
+ }
+}
+
+// TestASparseUpdateOvertakingItsStartWaitsForIt. The client sends from
+// several threads at once, so the request with the update is published
+// before the one with the start. The update waits, and lands in the same
+// pass once the start has.
+func TestASparseUpdateOvertakingItsStartWaitsForIt(t *testing.T) {
+ _, zone, base := started(t, "")
+ postBatch(t, base, `{"post":[`+inheritFunction("t-overtaken")+`]}`)
+ passes(t, collectorOver(zone), 1)
+ start := strings.Replace(inheritCall(``), `"end_time":"2026-09-20T10:00:02Z",`, ``, 1)
+ postBatch(t, base, `{"patch":[`+sparse(inheritChild, inheritRoot, lostCallOrder, "overtook its start")+`]}`)
+ postBatch(t, base, `{"post":[`+start+`]}`)
+ landed, err := collectorOver(zone).Collect()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if landed.Dropped != 0 || landed.Waiting != 0 || !mentions(t, zone, threadOf("t-overtaken"), "overtook its start") {
+ t.Errorf("%d dropped, %d waiting: the update must land once its start has", landed.Dropped, landed.Waiting)
+ }
+}
+
+// TestAStartPastTheWaitDropsOnlyItsUpdate. The start comes after its update
+// has waited as long as it may. The update is not collected, and the start,
+// which names its own thread, lands there: the conversation the update fell
+// back to was never remembered as the run's.
+func TestAStartPastTheWaitDropsOnlyItsUpdate(t *testing.T) {
+ _, zone, base := started(t, "")
+ order := "20260920T100000000000Z" + lostStep
+ postBatch(t, base, `{"patch":[`+sparse(lostStep, lostStep, order, "too early")+`]}`)
+ c := collectorOver(zone)
+ if dropped := passes(t, c, deferLimit+1); dropped != 1 {
+ t.Fatalf("%d dropped, want the update", dropped)
+ }
+ postBatch(t, base, `{"post":[`+rootRun(lostStep, "asz", `"thread_id":"t-late-start"`, "the start's own")+`]}`)
+ passes(t, c, 1)
+ if !mentions(t, zone, threadOf("t-late-start"), "the start's own") {
+ t.Error("the start is not in its own conversation")
+ }
+ if landedAnywhere(t, zone, updateOf(lostStep)) || landedAnywhere(t, zone, UnassignedID(lostStep)) {
+ t.Error("the update was collected, or the run landed unassigned")
+ }
+}
+
+// TestARunUnderARunKnownOnlyFromAnUpdateIsNotCollected. The function names a
+// live thread. The step inside it named a thread of its own, and was
+// forgotten while the function was not. An update of the step then arrives,
+// with no thread or with a thread but no project, and is not collected. A run
+// inside the step that names no thread cannot take the function's past it:
+// the step's start would say, and it is gone. It is not collected, after a
+// restart too.
+func TestARunUnderARunKnownOnlyFromAnUpdateIsNotCollected(t *testing.T) {
+ stepOrder := inheritOrder + ".20260920T100000500000Z" + lostStep
+ belowOrder := stepOrder + ".20260920T100001000000Z" + lostBelow
+ step := `{"id":"` + lostStep + `","trace_id":"` + inheritRoot + `","parent_run_id":"` + inheritRoot + `",` +
+ `"dotted_order":"` + stepOrder + `","run_type":"chain","name":"step","start_time":"2026-09-20T10:00:00.5Z",` +
+ `"session_name":"asz","extra":{"metadata":{"thread_id":"t-step"}},"inputs":{"q":"the step's own"}}`
+ below := `{"id":"` + lostBelow + `","trace_id":"` + inheritRoot + `","parent_run_id":"` + lostStep + `",` +
+ `"dotted_order":"` + belowOrder + `","run_type":"chain","name":"below","start_time":"2026-09-20T10:00:01Z",` +
+ `"end_time":"2026-09-20T10:00:01.5Z","session_name":"asz","extra":{"metadata":{}},"outputs":{"answer":"below the step"}}`
+ for name, update := range map[string]string{
+ "no thread": sparse(lostStep, inheritRoot, stepOrder, "the step's end"),
+ "a thread with no project": `{"id":"` + lostStep + `","trace_id":"` + inheritRoot + `","dotted_order":"` + stepOrder + `",` +
+ `"end_time":"2026-09-20T10:00:02.5Z","extra":{"metadata":{"thread_id":"t-step"}},"outputs":{"answer":"the step's end"}}`,
+ } {
+ _, zone, base := started(t, "")
+ c, clock := clockedOver(zone)
+ postBatch(t, base, `{"post":[`+inheritFunction("t-function")+`,`+step+`]}`)
+ passes(t, c, 1)
+ // The function is seen again, and the step is not.
+ clock.at = clock.at.Add(20 * time.Hour)
+ postBatch(t, base, `{"patch":[`+sparse(inheritRoot, inheritRoot, inheritOrder, "the function goes on")+`]}`)
+ passes(t, c, 1)
+ clock.at = clock.at.Add(10 * time.Hour)
+ postBatch(t, base, `{"post":[`+otherRun("t-other")+`]}`)
+ postBatch(t, base, `{"patch":[`+update+`]}`)
+ if dropped := passes(t, c, deferLimit+1); dropped != 1 {
+ t.Fatalf("%s: %d dropped, want the step's update", name, dropped)
+ }
+ restarted := &Collector{Zone: zone, Now: clock.now}
+ postBatch(t, base, `{"post":[`+below+`]}`)
+ if dropped := passes(t, restarted, deferLimit+1); dropped != 1 {
+ t.Errorf("%s: %d dropped, want the run below the step", name, dropped)
+ }
+ if landedAnywhere(t, zone, lostBelow) || landedAnywhere(t, zone, updateOf(lostStep)) {
+ t.Errorf("%s: a run whose conversation cannot be told was collected", name)
+ }
+ if !mentions(t, zone, threadOf("t-function"), updateOf(inheritRoot)) {
+ t.Errorf("%s: the function's own update is not in its conversation", name)
+ }
+ }
+}
+
+// TestAStartWithNoThreadUnderAForgottenRunIsNotCollected. The function was
+// forgotten by age. A call inside it that names no thread arrives, and what
+// the function named is not known: the call is not collected.
+func TestAStartWithNoThreadUnderAForgottenRunIsNotCollected(t *testing.T) {
+ _, zone, base := started(t, "")
+ c, clock := clockedOver(zone)
+ postBatch(t, base, `{"post":[`+inheritFunction("t-gone")+`]}`)
+ passes(t, c, 1)
+ clock.at = clock.at.Add(25 * time.Hour)
+ postBatch(t, base, `{"post":[`+otherRun("t-other")+`]}`)
+ postBatch(t, base, `{"post":[`+inheritCall(``)+`]}`)
+ if dropped := passes(t, c, deferLimit+1); dropped != 1 || landedAnywhere(t, zone, "the call's own answer") {
+ t.Errorf("%d dropped: a call under a forgotten function must not be collected", dropped)
+ }
+}
+
+// TestAnUnassignedRunKeepsItsConversation. A run with no thread anywhere
+// above it lands in its own conversation, unassigned. A later update that
+// carries a thread does not move it: one run is one conversation.
+func TestAnUnassignedRunKeepsItsConversation(t *testing.T) {
+ _, zone, base := started(t, "")
+ c := collectorOver(zone)
+ postBatch(t, base, `{"post":[`+rootRun(lostStep, "asz", ``, "no thread anywhere")+`]}`)
+ passes(t, c, 1)
+ postBatch(t, base, `{"patch":[`+rootRun(lostStep, "asz", `"thread_id":"t-later"`, "a thread too late")+`]}`)
+ passes(t, c, 1)
+ if !mentions(t, zone, UnassignedID(lostStep), "a thread too late") || landedAnywhere(t, zone, threadOf("t-later")) {
+ t.Error("the update left the conversation its run landed in")
+ }
+}
+
+// TestAnUpdateIsCompleteOnlyWithEveryDimension. Which dimensions name a
+// conversation is configured. With the thread alone, an update that carries
+// its thread names its conversation and lands. With a tenant too, an update
+// that carries the thread and no tenant does not, and is not collected once
+// it has waited; a start that leaves the tenant out supplies it empty.
+func TestAnUpdateIsCompleteOnlyWithEveryDimension(t *testing.T) {
+ keys := DefaultOwnership().Keys
+ _, zone, base := started(t, "")
+ c := collectorOver(zone)
+ c.Ownership = Ownership{Keys: keys, Scope: []string{"thread"}}
+ postBatch(t, base, `{"patch":[`+rootRun(lostStep, "", `"thread_id":"t-alone"`, "thread alone")+`]}`)
+ if landed, err := c.Collect(); err != nil || landed.Waiting != 0 {
+ t.Fatalf("%v, %d waiting: an update with every dimension needs no start", err, landed.Waiting)
+ }
+ if !mentions(t, zone, StorageID(Owner{Values: []string{"t-alone"}, Thread: "t-alone"}), "thread alone") {
+ t.Error("the update is not in the conversation its thread names")
+ }
+
+ _, zone, base = started(t, "")
+ c = collectorOver(zone)
+ c.Ownership = Ownership{Keys: keys, Scope: []string{"tenant", "thread"}}
+ postBatch(t, base, `{"patch":[`+rootRun(lostStep, "asz", `"thread_id":"t-tenant"`, "no tenant")+`]}`)
+ postBatch(t, base, `{"post":[`+rootRun(lostBelow, "asz", `"thread_id":"t-start"`, "a start with no tenant")+`]}`)
+ if dropped := passes(t, c, deferLimit+1); dropped != 1 || landedAnywhere(t, zone, lostStep) {
+ t.Errorf("%d dropped: an update with no tenant must not be collected", dropped)
+ }
+ if !mentions(t, zone, StorageID(Owner{Values: []string{"", "t-start"}, Thread: "t-start"}), "a start with no tenant") {
+ t.Error("the start that left the tenant out is not in its conversation")
+ }
+}
+
+// inboxEmpty reports whether every request was taken out of the inbox, so a
+// request is known to be done with rather than still waiting.
+func inboxEmpty(t *testing.T, zone *storage.Zone) bool {
+ t.Helper()
+ waiting, err := NewInbox(zone).List()
+ if err != nil {
+ t.Fatal(err)
+ }
+ return len(waiting) == 0
+}
+
+// TestARunForgottenByCountIsNotKnown. More than 50,000 runs are remembered,
+// and the oldest go, whatever their age. The call's update and a new call
+// under the forgotten function that names no thread are then not collected.
+func TestARunForgottenByCountIsNotKnown(t *testing.T) {
+ _, zone, base := started(t, "")
+ c, clock := clockedOver(zone)
+ start := strings.Replace(inheritCall(``), `"end_time":"2026-09-20T10:00:02Z",`, ``, 1)
+ postBatch(t, base, `{"post":[`+inheritFunction("t-counted")+`,`+start+`]}`)
+ passes(t, c, 1)
+ path := filepath.Join(zone.Root(), InboxDir, pendingFile)
+ open := loadPending(path)
+ later := clock.now().UTC().Format(time.RFC3339Nano)
+ for i := 0; i < pendingLimit; i++ {
+ open.Runs["newer-"+strconv.Itoa(i)] = pendingRun{Trace: "newer", Create: true, At: later}
+ }
+ if err := open.save(path, clock.now()); err != nil {
+ t.Fatal(err)
+ }
+ if kept := loadPending(path); len(kept.Runs) != pendingLimit {
+ t.Fatalf("%d remembered, want the limit", len(kept.Runs))
+ } else if _, ok := kept.Runs[inheritChild]; ok {
+ t.Fatal("the call is still remembered, although it is the oldest past the limit")
+ }
+ postBatch(t, base, `{"patch":[`+sparse(inheritChild, inheritRoot, lostCallOrder, "counted out")+`]}`)
+ const another = "15151515-1515-7515-8515-151515151514"
+ newCall := strings.ReplaceAll(inheritCall(``), inheritChild, another)
+ postBatch(t, base, `{"post":[`+newCall+`]}`)
+ if dropped := passes(t, c, deferLimit+1); dropped != 2 || !inboxEmpty(t, zone) {
+ t.Errorf("%d dropped: the update and the new call must not be collected", dropped)
+ }
+ if landedAnywhere(t, zone, "counted out") || landedAnywhere(t, zone, another) {
+ t.Error("a run under a function forgotten by count was collected")
+ }
+}
+
+// TestALaterStartDecidesForItsWaitingUpdate. A call's update, naming no
+// conversation, comes before its start, under a function that is
+// remembered. It is not given the function's conversation while it waits:
+// the call's start names a thread of its own, and both land there, with or
+// without a restart between.
+func TestALaterStartDecidesForItsWaitingUpdate(t *testing.T) {
+ for _, restart := range []bool{false, true} {
+ _, zone, base := started(t, "")
+ c := collectorOver(zone)
+ postBatch(t, base, `{"post":[`+inheritFunction("t-function")+`]}`)
+ passes(t, c, 1)
+ postBatch(t, base, `{"patch":[`+sparse(inheritChild, inheritRoot, lostCallOrder, "the call's own end")+`]}`)
+ if landed, err := c.Collect(); err != nil || landed.Waiting != 1 {
+ t.Fatalf("restart %v: %v, %d waiting: the update must wait for its start", restart, err, landed.Waiting)
+ }
+ if restart {
+ c = collectorOver(zone)
+ }
+ start := strings.Replace(inheritCall(`"thread_id":"t-own"`), `"end_time":"2026-09-20T10:00:02Z",`, ``, 1)
+ postBatch(t, base, `{"post":[`+start+`]}`)
+ if dropped := passes(t, c, 1); dropped != 0 || !inboxEmpty(t, zone) {
+ t.Fatalf("restart %v: %d dropped: both must land once the start has", restart, dropped)
+ }
+ if !mentions(t, zone, threadOf("t-own"), "the call's own end") || mentions(t, zone, threadOf("t-function"), inheritChild) {
+ t.Errorf("restart %v: the update is not with its start in the call's own conversation", restart)
+ }
+ }
+}
+
+// TestAnUnassignedRunBelowAForgottenRootStays. A trace with no thread
+// anywhere lands unassigned, its root and the call inside it. The root is
+// then forgotten, and the call is not. A later update of the call stays in
+// the call's conversation, and a new run inside the call that names no
+// thread is not collected: the root that would say it supplied none is gone.
+func TestAnUnassignedRunBelowAForgottenRootStays(t *testing.T) {
+ _, zone, base := started(t, "")
+ c, clock := clockedOver(zone)
+ function := strings.Replace(inheritFunction("x"), `"thread_id":"x"`, ``, 1)
+ start := strings.Replace(inheritCall(``), `"end_time":"2026-09-20T10:00:02Z",`, ``, 1)
+ postBatch(t, base, `{"post":[`+function+`,`+start+`]}`)
+ passes(t, c, 1)
+ if !mentions(t, zone, UnassignedID(inheritRoot), inheritChild) {
+ t.Fatal("the call did not land unassigned with its root")
+ }
+ clock.at = clock.at.Add(20 * time.Hour)
+ postBatch(t, base, `{"patch":[`+sparse(inheritChild, inheritRoot, lostCallOrder, "still unassigned")+`]}`)
+ passes(t, c, 1)
+ clock.at = clock.at.Add(10 * time.Hour)
+ postBatch(t, base, `{"post":[`+otherRun("t-other")+`]}`)
+ postBatch(t, base, `{"patch":[`+sparse(inheritChild, inheritRoot, lostCallOrder, "after the root went")+`]}`)
+ belowOrder := lostCallOrder + ".20260920T100001500000Z" + lostBelow
+ below := `{"id":"` + lostBelow + `","trace_id":"` + inheritRoot + `","parent_run_id":"` + inheritChild + `",` +
+ `"dotted_order":"` + belowOrder + `","run_type":"tool","name":"lookup","start_time":"2026-09-20T10:00:01.5Z",` +
+ `"end_time":"2026-09-20T10:00:01.8Z","session_name":"asz","extra":{"metadata":{}},"outputs":{"output":"below the call"}}`
+ postBatch(t, base, `{"post":[`+below+`]}`)
+ if dropped := passes(t, c, deferLimit+1); dropped != 1 || !inboxEmpty(t, zone) {
+ t.Errorf("%d dropped, want the run below the call alone", dropped)
+ }
+ for _, text := range []string{"still unassigned", "after the root went"} {
+ if !mentions(t, zone, UnassignedID(inheritRoot), text) {
+ t.Errorf("%q is not in the call's unassigned conversation", text)
+ }
+ }
+ if landedAnywhere(t, zone, "below the call") {
+ t.Error("the run below the call was collected, although the root above it is gone")
+ }
+}
+
+// TestAThreadWithNoTenantIsPartialOnEveryArrival. With a tenant in the
+// configured key, a run's first arrival names nothing and its second, in
+// the same request, names the project and a thread and no tenant. The second decides for
+// the first, and is not whole, so neither lands. An update that carries the
+// tenant and the thread lands at once.
+func TestAThreadWithNoTenantIsPartialOnEveryArrival(t *testing.T) {
+ _, zone, base := started(t, "")
+ c := collectorOver(zone)
+ c.Ownership = Ownership{Keys: DefaultOwnership().Keys, Scope: []string{"tenant", "thread"}}
+ order := "20260920T100000000000Z" + lostStep
+ named := `{"id":"` + lostStep + `","trace_id":"` + lostStep + `","dotted_order":"` + order + `",` +
+ `"end_time":"2026-09-20T10:00:03Z","session_name":"asz","extra":{"metadata":{"thread_id":"t-tenant"}},` +
+ `"outputs":{"answer":"no tenant"}}`
+ postBatch(t, base, `{"patch":[`+sparse(lostStep, lostStep, order, "nothing named")+`,`+named+`]}`)
+ if dropped := passes(t, c, deferLimit+1); dropped != 2 || landedAnywhere(t, zone, lostStep) {
+ t.Errorf("%d dropped: both arrivals of a run with no tenant must not be collected", dropped)
+ }
+ whole := rootRun(lostBelow, "asz", `"thread_id":"t-whole","tenant":"acme"`, "tenant and thread")
+ postBatch(t, base, `{"patch":[`+whole+`]}`)
+ if landed, err := c.Collect(); err != nil || landed.Waiting != 0 || landed.Dropped != 0 {
+ t.Fatalf("%v: %+v: an update with the whole key needs no start", err, landed)
+ }
+ if !mentions(t, zone, StorageID(Owner{Values: []string{"acme", "t-whole"}, Thread: "t-whole"}), "tenant and thread") {
+ t.Error("the whole update is not in its conversation")
+ }
+}
+
+// TestAStartTimeIsNotAStart. An update that names no conversation and
+// carries the run's start time is still an update: only a run's start
+// proves one was seen, so it is not collected.
+func TestAStartTimeIsNotAStart(t *testing.T) {
+ _, zone, base := started(t, "")
+ order := "20260920T100000000000Z" + lostStep
+ update := `{"id":"` + lostStep + `","trace_id":"` + lostStep + `","dotted_order":"` + order + `",` +
+ `"start_time":"2026-09-20T10:00:00Z","end_time":"2026-09-20T10:00:03Z","outputs":{"answer":"timed"}}`
+ postBatch(t, base, `{"patch":[`+update+`]}`)
+ if dropped := passes(t, collectorOver(zone), deferLimit+1); dropped != 1 || landedAnywhere(t, zone, lostStep) {
+ t.Errorf("%d dropped: an update with a start time is not a start", dropped)
+ }
+ open := loadPending(filepath.Join(zone.Root(), InboxDir, pendingFile))
+ if open.Runs[lostStep].Create {
+ t.Error("a start time was remembered as a start")
+ }
+}
+
+// TestAPartialStartUnderAThreadIsNotCollected. A call's start names a thread
+// but carries no project, under a function that names its own. The call does
+// not take the function's in place of its own, and its own is never whole,
+// so once it has waited as long as it may it is not collected.
+func TestAPartialStartUnderAThreadIsNotCollected(t *testing.T) {
+ _, zone, base := started(t, "")
+ c := collectorOver(zone)
+ postBatch(t, base, `{"post":[`+inheritFunction("t-function")+`]}`)
+ passes(t, c, 1)
+ call := strings.Replace(inheritCall(`"thread_id":"t-own"`), `"session_name":"asz",`, ``, 1)
+ postBatch(t, base, `{"post":[`+call+`]}`)
+ if dropped := passes(t, c, deferLimit+1); dropped != 1 || !inboxEmpty(t, zone) {
+ t.Errorf("%d dropped: a start whose own thread is never whole must not be collected", dropped)
+ }
+ if landedAnywhere(t, zone, "the call's own answer") {
+ t.Error("the partial call was collected")
+ }
+}
+
+// TestAnOlderMemoryIsForgotten. What version 1 remembered says nothing about
+// whether a run's start was seen, so it is not read as version 2.
+func TestAnOlderMemoryIsForgotten(t *testing.T) {
+ path := filepath.Join(t.TempDir(), pendingFile)
+ old := `{"version":1,"runs":{"` + inheritChild + `":{"trace":"` + inheritRoot + `","session":"ls-remembered"}}}`
+ if err := os.WriteFile(path, []byte(old), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ if open := loadPending(path); len(open.Runs) != 0 || open.Version != pendingVersion {
+ t.Errorf("version %d with %d runs: an older memory must be forgotten", open.Version, len(open.Runs))
+ }
+}
diff --git a/internal/adapters/langsmith/pending.go b/internal/adapters/langsmith/pending.go
index 6337791..b3b8c91 100644
--- a/internal/adapters/langsmith/pending.go
+++ b/internal/adapters/langsmith/pending.go
@@ -41,14 +41,21 @@ import (
// an arrival that leaves those fields out is completed from what its own
// start said. Nothing is invented: every field comes from an earlier arrival
// about the same run.
+//
+// What is remembered is forgotten by age and by count. An update that names
+// no conversation of its own, and comes after its run was forgotten or with no
+// start ever seen, cannot be given back its conversation, and it is not
+// collected: landing it under a name saying its identity was not supplied put
+// evidence of one conversation under another.
// pendingFile holds the runs that were started and have not ended.
const pendingFile = "open.json"
// pendingVersion is bumped when the file's meaning changes. An older file is
-// discarded rather than migrated, which costs the completion of whatever was
-// in flight at the time and nothing else.
-const pendingVersion = 1
+// discarded rather than migrated: every run it remembered is forgotten once,
+// as if a day had passed. Version 2 records whether a run's start was seen,
+// which version 1 did not.
+const pendingVersion = 2
// pendingLimit bounds the file. A run that never ends would otherwise be
// remembered for ever, and a client that crashes mid-trace leaves many. When
@@ -76,11 +83,15 @@ type pendingRun struct {
// resolved. It is kept rather than derived again because the update
// carries none of what deriving it needs.
Session string `json:"session,omitempty"`
- // Partial records that an arrival named a thread before the run's
- // project was known, so the conversation it named is not whole. Such a
- // run neither takes a conversation from a run above it nor gives one to
- // a run below it.
+ // Partial records that an arrival named a thread before the rest of the
+ // conversation's key was known: the project, or a metadata key the scope
+ // names. The conversation it named is not whole. Such a run neither takes
+ // a conversation from a run above it nor gives one to a run below it.
Partial bool `json:"partial,omitempty"`
+ // Create records that the run's start was seen. An update is remembered
+ // too, for the ancestry it carries, so being remembered proves no start,
+ // and neither does a session or a start time an update brought.
+ Create bool `json:"create,omitempty"`
// At is when this was remembered, for the eviction above.
At string `json:"at,omitempty"`
}
@@ -218,6 +229,13 @@ func (p *pending) started(run Run, method, session string, now time.Time) {
p.Runs[run.ID] = known
}
+// created records that a run's start was seen.
+func (p *pending) created(id string) {
+ known := p.Runs[id]
+ known.Create = true
+ p.Runs[id] = known
+}
+
// complete fills in what this arrival left out, from what the run's own
// start said. A field the arrival carries is never overwritten.
func (p *pending) complete(run Run) (Run, pendingRun, bool) {
@@ -249,33 +267,48 @@ func (p *pending) complete(run Run) (Run, pendingRun, bool) {
return run, known, true
}
+// inheritance is what the runs above one say about its conversation.
+type inheritance int
+
+const (
+ // inheritedNone: every run above it is remembered with its start seen,
+ // and none supplied a conversation. It lands in its own, unassigned.
+ inheritedNone inheritance = iota
+ // inheritedResolved: the nearest run above it that decides supplied one.
+ inheritedResolved
+ // inheritedUnknown: a run above it that would decide is not known, so
+ // neither is its conversation.
+ inheritedUnknown
+)
+
// inherited is the conversation of the nearest run this one ran inside whose
// conversation was supplied, by that run or by one above it. A run supplied
// nothing when its own is empty or unassigned, and is passed over. Every
// arrival is remembered as it is read, before its request lands or waits, so
// a run of this request or of one still waiting is here.
//
-// The walk stops at a run that is not remembered: one that has not arrived,
-// or was forgotten by age. What it supplied is not known, and passing over it
-// could give this run a thread from further up that the run in between
-// replaced. Nothing, then, as when no run above supplied one.
-//
-// A run that named a thread before its project was known stops it too: it
-// supplied a conversation, but not a whole one.
-func (p *pending) inherited(run Run) string {
+// The walk stops at the first run that does not say: one that is not
+// remembered, because it has not arrived or was forgotten; one remembered only
+// from updates that named no conversation of their own, whose start would
+// have said; and one that named a thread before the rest of its key was known,
+// which supplied a conversation but not a whole one. An update that named its
+// run's whole conversation decides, as a start does. What such a run supplied is not known, and passing over it could give
+// this run a thread from further up that the run in between replaced, or a
+// conversation that run's start did not name.
+func (p *pending) inherited(run Run) (string, inheritance) {
for _, id := range ancestors(run.Dotted, run.ID) {
known, ok := p.Runs[id]
if !ok {
- return ""
+ return "", inheritedUnknown
}
if assigned(id, known) {
- return known.Session
+ return known.Session, inheritedResolved
}
- if known.Partial {
- return ""
+ if known.Partial || !known.Create {
+ return "", inheritedUnknown
}
}
- return ""
+ return "", inheritedNone
}
// partial records that a run named a thread before its project was known.