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/09 01:16:57 UTC
(skywalking-ai-sessionizer) branch main updated: Give a run that supplies no thread the thread of the run it ran inside (#57)
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 7208bb3 Give a run that supplies no thread the thread of the run it ran inside (#57)
7208bb3 is described below
commit 7208bb358900c4024fec1b43b54c7764a8858269
Author: 吴晟 Wu Sheng <wu...@foxmail.com>
AuthorDate: Fri Oct 9 09:16:52 2026 +0800
Give a run that supplies no thread the thread of the run it ran inside (#57)
LangSmith JS passes a decorated function's metadata to none of the runs
inside it. A model call made inside a traceable function therefore named
no thread, and landed under its trace in a conversation of its own: the
answer was there, and the conversation of the question had none. The
Python client passes the metadata down itself, so its runs were whole.
A run that supplies no key now takes the conversation of the nearest run
it ran inside whose conversation was supplied, read from its dotted
order. A run that supplies its own key keeps it, and a trace that
supplied none anywhere still lands unassigned. A run above that supplied
nothing, or landed unassigned under either name an unassigned run can
have, is passed over. A run above that is not remembered, because it has
not arrived or was forgotten, stops the search, since it may have
supplied a thread of its own.
The order and the batching of requests do not decide it. The
conversation is decided once the run's ancestry has arrived, which the
receiver already waits for. Every arrival is remembered as it is read,
before its request lands or waits, so a run below it in another request
finds what it supplied, or that it supplied nothing. While a request
waits, a run that supplied no key is remembered with no conversation,
since the unassigned name remembered there was kept when the request
was tried again after its root arrived.
One run is one conversation. Its own key decides from the first of its
arrivals that brings one, in one request or across waiting requests,
and once it has landed every later arrival follows it. A run is counted
unassigned once, not once per arrival.
An update carries only what changed, so it can name a thread with no
project and arrive before the run's start. Remembered as the run's
conversation while it waited, it named the conversation with an empty
project, and the start that carried the project could no longer decide,
as it could before every arrival was remembered. Such a thread is now not
remembered as the run's conversation; the run does not take a thread
from above in place of it, and a run below does not pass over it.
The JS traceable capture now lands as one conversation and assembles
with the question, the model call and its answer. Twelve tests cover
the rule, including one that sends the same runs in several orders and
batchings; breaking each of fourteen parts of the rule fails one.
---
docs/en/adapters/langsmith.md | 36 ++-
docs/en/changes/changes.md | 16 +-
docs/en/setup/langchain.md | 12 +
internal/adapters/langsmith/collect.go | 102 +++++-
internal/adapters/langsmith/inherit_test.go | 460 ++++++++++++++++++++++++++++
internal/adapters/langsmith/pending.go | 51 +++
tests/langchain_test.go | 5 +
tests/langchainjs_test.go | 32 +-
8 files changed, 674 insertions(+), 40 deletions(-)
diff --git a/docs/en/adapters/langsmith.md b/docs/en/adapters/langsmith.md
index 7ae8d74..198749e 100644
--- a/docs/en/adapters/langsmith.md
+++ b/docs/en/adapters/langsmith.md
@@ -51,9 +51,10 @@ lands as the Python one does: the person's question opens the turn, a tool's res
tool said, and a decorated function keeps its arguments. Read as the Python client writes them,
every JS run was refused for its end time, and, with that read, no question opened a LangGraph JS
conversation and a tool's result landed as the serialized message. Because the metadata is not
-passed down, a model call made inside a decorated function supplies no thread of its own, and
-lands under its trace as any run that supplied none does; a LangGraph JS graph puts the thread on
-every run, so its conversation is whole.
+passed down, a model call made inside a decorated function supplies no thread of its own. It takes
+the function's, as [Ownership is a namespace](#ownership-is-a-namespace) says a run that supplies
+none does; landed under its trace, the answer was a conversation of its own and the question's had
+none. A LangGraph JS graph puts the thread on every run.
An update carries only what changed. `update_run(id, outputs=...)` sends the id and the outputs,
with the trace, the project, the kind of run and the dotted order all null — read on its own,
@@ -304,8 +305,33 @@ and `会话-1` as real thread keys, so the session directory is derived from the
a digest of the whole namespace, under an `ls-` prefix that keeps it off the leading underscore
enumeration skips and away from every Windows device name.
-A trace with no supplied key is not given one. It lands under a name that says the identity was not
-supplied, and nothing merges it with anything else.
+A run that supplies no key takes the conversation of the nearest run it ran inside whose
+conversation was supplied. The run's dotted order names every run it ran inside, and that run
+supplied the thread, so nothing is guessed. A run with no dotted order is not given one. A run that
+supplies its own key keeps it. The Python client passes a decorated function's metadata down to the
+runs inside it, and LangSmith JS does not, so without this a model call made inside a JS function
+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.
+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.
+
+An update carries only what changed, so it can name a thread with no project, and arrive before the
+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.
## What the wire cannot supply
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index 6e94272..dbcce47 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -19,10 +19,22 @@
of these now reads the message by its class. A decorated function carries no `ls_method` from it,
and is known by the library that made the run, which both clients write. A model call inside
a decorated function supplies no thread, because LangSmith JS passes the function's metadata to
- none of the runs inside it, and lands under its trace. Measured with LangSmith JS 0.10.8 on a
- LangGraph JS agent and a traced function, the two applications under `tests/apps/langchain-js`.
+ none of the runs inside it, and takes the function's, as the next entry says. Measured with
+ LangSmith JS 0.10.8 on a LangGraph JS agent and a traced function, the two applications under
+ `tests/apps/langchain-js`.
Their captures sit beside the Python ones under `tests/apps/langchain/testdata`, and the recorder
reads the chunked body that client sends.
+- A run that supplies no thread takes the conversation of the nearest run it ran inside whose
+ conversation was supplied. Its dotted order names the runs it ran inside, and the thread was
+ supplied there. A model call inside a LangSmith JS `traceable` supplies none, so it landed in
+ a conversation of its own under its trace, with its answer, while the question's conversation had
+ 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
+ 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.
- A summary `SummarizationMiddleware` wrote now resets the context, as a Claude Code compaction
does. The middleware marks its summary message with `additional_kwargs.lc_source` set to
`summarization`. A model call sent that message starts a new epoch in its stream, with an
diff --git a/docs/en/setup/langchain.md b/docs/en/setup/langchain.md
index 5d5ceb4..e1c9b08 100644
--- a/docs/en/setup/langchain.md
+++ b/docs/en/setup/langchain.md
@@ -70,6 +70,18 @@ metadata instead:
def answer(question): ...
```
+In LangChain JS, the same goes on the client's `traceable`:
+
+```js
+const answer = traceable(async (question) => { /* ... */ }, {
+ name: "answer",
+ metadata: { thread_id: "the-conversation" },
+});
+```
+
+A run inside it that names no thread, such as a model call, is part of the same
+conversation.
+
## What owns a conversation
A thread key alone is not an identity. Two applications can both use the
diff --git a/internal/adapters/langsmith/collect.go b/internal/adapters/langsmith/collect.go
index a6c3b3d..769ebcc 100644
--- a/internal/adapters/langsmith/collect.go
+++ b/internal/adapters/langsmith/collect.go
@@ -26,6 +26,7 @@ import (
"net/http"
"os"
"path/filepath"
+ "slices"
"sort"
"strings"
"time"
@@ -306,7 +307,17 @@ func (c *Collector) convertWith(request Waiting, open *pending, wait bool) (Land
runs := make([]Run, len(operations))
sessions := make([]string, len(operations))
methods := make([]string, len(operations))
- unassigned := 0
+ // One run is one conversation. Its first arrival in this request
+ // decides, and a later arrival of the same run follows it, so one run is
+ // never split across two; a later arrival that names a conversation
+ // 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 := make([]bool, len(operations))
+ follows := make([]int, len(operations))
+ first := map[string]int{}
for i, op := range operations {
var envelope Run
if err := json.Unmarshal(op.Envelope, &envelope); err != nil {
@@ -320,18 +331,46 @@ func (c *Collector) convertWith(request Waiting, open *pending, wait bool) (Land
// later, because a start and an update to it can be in one batch and
// the update carries none of what deciding it needs. The start's
// answer has to be available by the time the update is read.
- switch {
+ follows[i] = -1
+ decides := i
+ switch j, seen := first[op.RunID]; {
case found && known.Session != "":
sessions[i] = known.Session
+ case seen:
+ 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)
+ }
+ }
default:
- var missing bool
- sessions[i], missing = c.sessionOf(op, completed)
- if missing {
- unassigned++
+ first[op.RunID] = i
+ sessions[i], unsupplied[i] = c.sessionOf(op, completed)
+ if unsupplied[i] {
+ partial[i] = known.Partial
+ } else {
+ partial[i] = c.projectUnknown(completed)
}
}
- if completed.Type != "" {
- open.started(completed, methodOf(op), sessions[i], c.Now())
+ // A run that supplied no conversation is not remembered in one yet.
+ // Which it takes is decided below, once its ancestry has arrived, and
+ // remembering the unassigned one here kept it when the request was
+ // tried again after its root had arrived. Every arrival is
+ // remembered, with no kind or no conversation too, so a run below it
+ // that arrives in another request while this one waits finds what it
+ // supplied, or that it supplied nothing.
+ //
+ // A thread named before the run's project was known is not
+ // remembered as its conversation either: a later arrival that
+ // carries the project decides, as it did before any arrival was
+ // remembered while it waited.
+ remembered := sessions[decides]
+ if unsupplied[decides] || partial[decides] {
+ remembered = ""
+ }
+ open.started(completed, methodOf(op), remembered, c.Now())
+ if partial[decides] {
+ open.partial(op.RunID)
}
open.sawRun(completed, c.Now())
}
@@ -346,6 +385,37 @@ func (c *Collector) convertWith(request Waiting, open *pending, wait bool) (Land
}
}
+ // 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.
+ 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++
+ continue
+ }
+ if session := open.inherited(runs[i]); session != "" {
+ sessions[i] = session
+ continue
+ }
+ unassigned++
+ }
+ for i, j := range follows {
+ if j >= 0 {
+ sessions[i] = sessions[j]
+ }
+ }
+
byRun := make(map[string]Run, len(operations))
for i, op := range operations {
if _, seen := byRun[op.RunID]; !seen || runs[i].Type != "" {
@@ -431,11 +501,12 @@ func (c *Collector) convertWith(request Waiting, open *pending, wait bool) (Land
return out, nil
}
-// sessionOf decides where a run's evidence lands.
+// sessionOf decides where a run's evidence lands, by what the run itself
+// supplied.
//
-// A run with no supplied key is not given one: it lands under its trace, in a
-// session whose name says the identity was not supplied, and nothing merges it
-// with anything else.
+// A run with no supplied key is not given one here: it is named for its trace,
+// 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.
@@ -463,6 +534,13 @@ 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")
+}
+
// traceOrRun is what identifies a run that supplied no conversation: its
// trace, or itself when the client sent no trace either.
func traceOrRun(run Run) string {
diff --git a/internal/adapters/langsmith/inherit_test.go b/internal/adapters/langsmith/inherit_test.go
new file mode 100644
index 0000000..c427b89
--- /dev/null
+++ b/internal/adapters/langsmith/inherit_test.go
@@ -0,0 +1,460 @@
+// 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 (
+ "net/http"
+ "strings"
+ "testing"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/storage"
+)
+
+// A decorated function and the model call it makes, as LangSmith JS sends
+// them: the thread is on the function alone, since that client passes a
+// decorated function's metadata to none of the runs inside it.
+const (
+ inheritRoot = "14141414-1414-7414-8414-141414141410"
+ inheritChild = "14141414-1414-7414-8414-141414141411"
+ inheritOrder = "20260920T100000000000Z" + inheritRoot
+)
+
+func inheritFunction(thread string) string {
+ return `{"id":"` + inheritRoot + `","trace_id":"` + inheritRoot + `","dotted_order":"` + inheritOrder + `",` +
+ `"run_type":"chain","name":"review","start_time":"2026-09-20T10:00:00Z","end_time":"2026-09-20T10:00:03Z",` +
+ `"session_name":"asz","extra":{"metadata":{"thread_id":"` + thread + `"},"runtime":{"library":"langsmith"}},` +
+ `"inputs":{"question":"what changed in the docs"},"outputs":{"answer":"two pages"}}`
+}
+
+func inheritCall(metadata string) string {
+ return `{"id":"` + inheritChild + `","trace_id":"` + inheritRoot + `","parent_run_id":"` + inheritRoot + `",` +
+ `"dotted_order":"` + inheritOrder + `.20260920T100001000000Z` + inheritChild + `",` +
+ `"run_type":"llm","name":"FakeListChatModel","start_time":"2026-09-20T10:00:01Z",` +
+ `"end_time":"2026-09-20T10:00:02Z","session_name":"asz","extra":{"metadata":{` + metadata + `}},` +
+ `"outputs":{"generations":[[{"message":{"kwargs":{"content":"the call's own answer"}}}]]}}`
+}
+
+func postBatch(t *testing.T, base, body string) {
+ t.Helper()
+ resp := send(t, base, "/runs/batch", "application/json", []byte(body), "")
+ resp.Body.Close()
+ if resp.StatusCode != http.StatusAccepted {
+ t.Fatalf("status %d", resp.StatusCode)
+ }
+}
+
+// TestARunWithNoThreadTakesTheFunctionsItRanInside. The call names no thread,
+// but the function it ran inside does, and the trace and the parent on the
+// wire say the call ran there. Landed under its trace, the answer was a
+// conversation of its own, and the question's conversation had none. The
+// call is first in the batch, so the order inside a request decides nothing.
+func TestARunWithNoThreadTakesTheFunctionsItRanInside(t *testing.T) {
+ _, zone, base := started(t, "")
+ postBatch(t, base, `{"post":[`+inheritCall(``)+`,`+inheritFunction("t-inherit")+`]}`)
+ landed, err := collectorOver(zone).Collect()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(landed.Sessions) != 1 || landed.Unassigned != 0 {
+ t.Fatalf("sessions %v, %d unassigned, want the function's thread alone", landed.Sessions, landed.Unassigned)
+ }
+ 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")
+ }
+}
+
+// TestARunThatArrivesBeforeItsFunctionStillTakesItsThread. The client sends
+// several batches at once and retries a failed one, so the call can arrive a
+// pass before the function. It waits for its ancestry, and its conversation
+// is decided when the function has arrived, not when it first came: the
+// unassigned name it would have had then was remembered, and kept when the
+// request was tried again.
+func TestARunThatArrivesBeforeItsFunctionStillTakesItsThread(t *testing.T) {
+ _, zone, base := started(t, "")
+ collector := collectorOver(zone)
+ postBatch(t, base, `{"post":[`+inheritCall(``)+`]}`)
+ landed, err := collector.Collect()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if landed.Waiting != 1 || landed.Records != 0 {
+ t.Fatalf("%d waiting, %d records, want the call held for its function", landed.Waiting, landed.Records)
+ }
+ postBatch(t, base, `{"post":[`+inheritFunction("t-later")+`]}`)
+ landed, err = collector.Collect()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(landed.Sessions) != 1 || landed.Unassigned != 0 {
+ t.Fatalf("sessions %v, %d unassigned, want the function's thread alone", landed.Sessions, landed.Unassigned)
+ }
+ 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")
+ }
+}
+
+// TestARunsOwnThreadIsKept. A run that supplies a thread keeps it, whatever
+// the run it ran inside supplied.
+func TestARunsOwnThreadIsKept(t *testing.T) {
+ _, zone, base := started(t, "")
+ postBatch(t, base, `{"post":[`+inheritFunction("t-outer")+`,`+inheritCall(`"thread_id":"t-own"`)+`]}`)
+ landed, err := collectorOver(zone).Collect()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(landed.Sessions) != 2 {
+ t.Fatalf("sessions %v, want the function's and the call's own", landed.Sessions)
+ }
+ for _, session := range landed.Sessions {
+ answer := mentions(t, zone, session, "the call's own answer")
+ if strings.Contains(session, "t-own") != answer {
+ t.Errorf("%s: holds the call's answer %v", session, answer)
+ }
+ }
+}
+
+// TestATraceThatSuppliedNoThreadIsNotGivenOne. With no thread anywhere in the
+// trace, nothing is inherited: the function and the call land together under
+// the trace, in a session named as unassigned.
+func TestATraceThatSuppliedNoThreadIsNotGivenOne(t *testing.T) {
+ _, zone, base := started(t, "")
+ function := strings.Replace(inheritFunction("x"), `"thread_id":"x"`, ``, 1)
+ postBatch(t, base, `{"post":[`+function+`,`+inheritCall(``)+`]}`)
+ landed, err := collectorOver(zone).Collect()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(landed.Sessions) != 1 || landed.Sessions[0] != UnassignedID(inheritRoot) || landed.Unassigned != 2 {
+ t.Fatalf("sessions %v, %d unassigned, want both runs under %s",
+ landed.Sessions, landed.Unassigned, UnassignedID(inheritRoot))
+ }
+}
+
+// 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) {
+ _, zone, base := started(t, "")
+ collector := collectorOver(zone)
+ start := strings.Replace(inheritCall(``), `"end_time":"2026-09-20T10:00:02Z",`, ``, 1)
+ postBatch(t, base, `{"post":[`+start+`]}`)
+ for pass := 0; pass <= deferLimit; pass++ {
+ if _, err := collector.Collect(); err != nil {
+ t.Fatal(err)
+ }
+ }
+ 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")
+ }
+ update := `{"id":"` + inheritChild + `","trace_id":"` + inheritRoot + `",` +
+ `"dotted_order":"` + inheritOrder + `.20260920T100001000000Z` + inheritChild + `",` +
+ `"end_time":"2026-09-20T10:00:02Z",` +
+ `"outputs":{"generations":[[{"message":{"kwargs":{"content":"the late answer"}}}]]}}`
+ postBatch(t, base, `{"post":[`+inheritFunction("t-late")+`],"patch":[`+update+`]}`)
+ 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")
+ }
+}
+
+// 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) {
+ _, zone, base := started(t, "")
+ collector := collectorOver(zone)
+ const step = "14141414-1414-7414-8414-141414141412"
+ stepOrder := inheritOrder + ".20260920T100000500000Z" + step
+ 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")
+ }
+ call := strings.Replace(inheritCall(``), `"parent_run_id":"`+inheritRoot+`","dotted_order":"`+inheritOrder,
+ `"parent_run_id":"`+step+`","dotted_order":"`+stepOrder, 1)
+ postBatch(t, base, `{"post":[`+inheritFunction("t-deep")+`,`+call+`]}`)
+ landed, err := collector.Collect()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(landed.Sessions) != 1 || landed.Unassigned != 0 {
+ t.Fatalf("sessions %v, %d unassigned, want the function's thread alone", landed.Sessions, landed.Unassigned)
+ }
+ 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")
+ }
+}
+
+// TestOneRunInOneRequestIsOneConversation. A batch carries the call's start,
+// which names no thread, and an update to it that names one of its own. The
+// run supplied a thread, so it keeps it, as it does when the two wait in
+// requests of their own: both arrivals land in the call's own conversation,
+// and the run is not split. The run is counted unassigned once at most, not
+// once for each arrival.
+func TestOneRunInOneRequestIsOneConversation(t *testing.T) {
+ _, zone, base := started(t, "")
+ start := strings.Replace(inheritCall(``), `"end_time":"2026-09-20T10:00:02Z",`, ``, 1)
+ update := `{"id":"` + inheritChild + `","trace_id":"` + inheritRoot + `",` +
+ `"dotted_order":"` + inheritOrder + `.20260920T100001000000Z` + inheritChild + `",` +
+ `"end_time":"2026-09-20T10:00:02Z","extra":{"metadata":{"thread_id":"t-other"}},` +
+ `"outputs":{"generations":[[{"message":{"kwargs":{"content":"the patched answer"}}}]]}}`
+ postBatch(t, base, `{"post":[`+inheritFunction("t-one")+`,`+start+`],"patch":[`+update+`]}`)
+ landed, err := collectorOver(zone).Collect()
+ if err != nil {
+ t.Fatal(err)
+ }
+ own := StorageID(Owner{Values: []string{"asz", "t-other"}, Thread: "t-other"})
+ function := StorageID(Owner{Values: []string{"asz", "t-one"}, Thread: "t-one"})
+ if !mentions(t, zone, own, `"FakeListChatModel"`) || !mentions(t, zone, own, "the patched answer") {
+ t.Errorf("sessions %v: the call's start and its update are not both in its own conversation", landed.Sessions)
+ }
+ if mentions(t, zone, function, inheritChild) {
+ t.Error("an arrival of the call landed in the function's conversation: the run is split")
+ }
+
+ // And with no thread anywhere, the run's two arrivals are one run unassigned.
+ _, zone, base = started(t, "")
+ keyless := strings.Replace(inheritFunction("x"), `"thread_id":"x"`, ``, 1)
+ opened := strings.Replace(keyless, `"end_time":"2026-09-20T10:00:03Z",`, ``, 1)
+ ended := `{"id":"` + inheritRoot + `","trace_id":"` + inheritRoot + `","dotted_order":"` + inheritOrder + `",` +
+ `"end_time":"2026-09-20T10:00:03Z","outputs":{"answer":"two pages"}}`
+ postBatch(t, base, `{"post":[`+opened+`],"patch":[`+ended+`]}`)
+ landed, err = collectorOver(zone).Collect()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if landed.Unassigned != 1 {
+ t.Errorf("%d unassigned, want 1: one run supplied no conversation", landed.Unassigned)
+ }
+}
+
+// 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) {
+ _, zone, base := started(t, "")
+ const step = "14141414-1414-7414-8414-141414141413"
+ stepOrder := inheritOrder + ".20260920T100000500000Z" + step
+ update := `{"id":"` + step + `","trace_id":"` + inheritRoot + `","dotted_order":"` + stepOrder + `",` +
+ `"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()
+ 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)
+ }
+}
+
+// TestARunThatSuppliedAThreadWithNoKindIsRemembered. The function's update,
+// naming the thread but no kind, waits for the run it ran inside. The call
+// below it, arriving first, is placed in the same pass once that run arrives,
+// before the function's request is tried again. It finds the function's
+// thread only if the waiting update was remembered with it.
+func TestARunThatSuppliedAThreadWithNoKindIsRemembered(t *testing.T) {
+ _, zone, base := started(t, "")
+ const outer = "14141414-1414-7414-8414-141414141414"
+ outerOrder := "20260920T095959000000Z" + outer
+ functionOrder := outerOrder + ".20260920T100000000000Z" + inheritRoot
+ callOrder := functionOrder + ".20260920T100001000000Z" + inheritChild
+ call := strings.Replace(inheritCall(``), `"trace_id":"`+inheritRoot+`"`, `"trace_id":"`+outer+`"`, 1)
+ call = strings.Replace(call, `"dotted_order":"`+inheritOrder+`.20260920T100001000000Z`+inheritChild, `"dotted_order":"`+callOrder, 1)
+ postBatch(t, base, `{"post":[`+call+`]}`)
+ postBatch(t, base, `{"patch":[{"id":"`+inheritRoot+`","trace_id":"`+outer+`","parent_run_id":"`+outer+`",`+
+ `"dotted_order":"`+functionOrder+`","end_time":"2026-09-20T10:00:03Z","session_name":"asz",`+
+ `"extra":{"metadata":{"thread_id":"t-remembered"}},"outputs":{"answer":"two pages"}}]}`)
+ postBatch(t, base, `{"post":[{"id":"`+outer+`","trace_id":"`+outer+`","dotted_order":"`+outerOrder+`",`+
+ `"run_type":"chain","name":"outer","start_time":"2026-09-20T09:59:59Z","end_time":"2026-09-20T10:00:04Z",`+
+ `"session_name":"asz","extra":{"metadata":{}},"inputs":{"question":"what changed"}}]}`)
+ landed, err := collectorOver(zone).Collect()
+ if err != nil {
+ t.Fatal(err)
+ }
+ thread := StorageID(Owner{Values: []string{"asz", "t-remembered"}, Thread: "t-remembered"})
+ if !mentions(t, zone, thread, "the call's own answer") {
+ t.Errorf("sessions %v: the call is not in the conversation of the function it ran inside", landed.Sessions)
+ }
+}
+
+// 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.
+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)
+ }
+ 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: "ls-step"}
+ if got := memory.inherited(call); got != "ls-step" {
+ t.Errorf("took %q, want the nearest supplied, the step's", got)
+ }
+}
+
+// landInOrder sends each body as a request of its own, in order, and then
+// collects them in one pass, as a client sending several batches at once
+// leaves them in the inbox.
+func landInOrder(t *testing.T, bodies ...string) *storage.Zone {
+ t.Helper()
+ _, zone, base := started(t, "")
+ for _, body := range bodies {
+ postBatch(t, base, body)
+ }
+ if _, err := collectorOver(zone).Collect(); err != nil {
+ t.Fatal(err)
+ }
+ return 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.
+func TestTheOrderOfRequestsDoesNotDecide(t *testing.T) {
+ const step = "14141414-1414-7414-8414-141414141416"
+ stepOrder := inheritOrder + ".20260920T100000500000Z" + step
+ function := `{"post":[` + inheritFunction("t-order") + `]}`
+ thread := StorageID(Owner{Values: []string{"asz", "t-order"}, Thread: "t-order"})
+ stepUpdate := `{"id":"` + step + `","trace_id":"` + inheritRoot + `","dotted_order":"` + stepOrder + `",` +
+ `"end_time":"2026-09-20T10:00:02.5Z","outputs":{"done":true}}`
+ call := strings.Replace(inheritCall(``), `"dotted_order":"`+inheritOrder, `"dotted_order":"`+stepOrder, 1)
+ for name, bodies := range map[string][]string{
+ "call, step, function": {`{"post":[` + call + `]}`, `{"patch":[` + stepUpdate + `]}`, function},
+ "step, call, function": {`{"patch":[` + stepUpdate + `]}`, `{"post":[` + call + `]}`, function},
+ "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)
+ }
+ }
+
+ start := strings.Replace(inheritCall(``), `"end_time":"2026-09-20T10:00:02Z",`, ``, 1)
+ update := `{"id":"` + inheritChild + `","trace_id":"` + inheritRoot + `",` +
+ `"dotted_order":"` + inheritOrder + `.20260920T100001000000Z` + inheritChild + `",` +
+ `"end_time":"2026-09-20T10:00:02Z","extra":{"metadata":{"thread_id":"t-own"}},` +
+ `"outputs":{"generations":[[{"message":{"kwargs":{"content":"the patched answer"}}}]]}}`
+ own := StorageID(Owner{Values: []string{"asz", "t-own"}, Thread: "t-own"})
+ for name, bodies := range map[string][]string{
+ "start, update, function": {`{"post":[` + start + `]}`, `{"patch":[` + update + `]}`, function},
+ "start and update, function": {`{"post":[` + start + `],"patch":[` + update + `]}`, function},
+ } {
+ zone := landInOrder(t, bodies...)
+ if !mentions(t, zone, own, `"FakeListChatModel"`) || !mentions(t, zone, own, "the patched answer") {
+ t.Errorf("%s: the call's start and its update are not both in its own conversation", name)
+ }
+ if mentions(t, zone, thread, inheritChild) {
+ t.Errorf("%s: an arrival of the call landed in the function's conversation", name)
+ }
+ }
+}
+
+// TestAThreadNamedBeforeItsProjectIsNotWhole. An update names the call's own
+// thread but no project, since an update carries only what changed, and it
+// can arrive before the call's start, which carries the project. Remembered as
+// the call's conversation while it waited, it named the conversation with an
+// empty project, and the start, carrying both, could no longer decide.
+func TestAThreadNamedBeforeItsProjectIsNotWhole(t *testing.T) {
+ order := inheritOrder + ".20260920T100001000000Z" + inheritChild
+ update := `{"id":"` + inheritChild + `","trace_id":"` + inheritRoot + `","dotted_order":"` + order + `",` +
+ `"end_time":"2026-09-20T10:00:02Z","extra":{"metadata":{"thread_id":"t-own"}},` +
+ `"outputs":{"generations":[[{"message":{"kwargs":{"content":"the patched answer"}}}]]}}`
+ start := strings.Replace(inheritCall(`"thread_id":"t-own"`), `"end_time":"2026-09-20T10:00:02Z",`, ``, 1)
+ own := StorageID(Owner{Values: []string{"asz", "t-own"}, Thread: "t-own"})
+ outer := StorageID(Owner{Values: []string{"asz", "t-outer"}, Thread: "t-outer"})
+
+ // In requests of their own, the update first.
+ zone := landInOrder(t, `{"patch":[`+update+`]}`, `{"post":[`+start+`]}`, `{"post":[`+inheritFunction("t-outer")+`]}`)
+ if !mentions(t, zone, own, `"FakeListChatModel"`) || !mentions(t, zone, own, "the patched answer") {
+ t.Error("apart: the call's update and its start are not both in its own conversation, project and all")
+ }
+
+ // In one request, the update before the start.
+ _, zone, base := started(t, "")
+ // A multipart body carries the outputs as a part of their own.
+ body, contentType := multipartOf([][2]string{
+ {"patch." + inheritChild, strings.Replace(update,
+ `,"outputs":{"generations":[[{"message":{"kwargs":{"content":"the patched answer"}}}]]}`, ``, 1)},
+ {"patch." + inheritChild + ".outputs", `{"generations":[[{"message":{"kwargs":{"content":"the patched answer"}}}]]}`},
+ {"post." + inheritChild, start},
+ {"post." + inheritRoot, inheritFunction("t-outer")},
+ })
+ resp := send(t, base, "/runs/multipart", contentType, body, "")
+ resp.Body.Close()
+ if _, err := collectorOver(zone).Collect(); err != nil {
+ t.Fatal(err)
+ }
+ // The update, read before the start, names no kind and lands as the
+ // run's data, so where it landed is found by its record id.
+ if !mentions(t, zone, own, `"FakeListChatModel"`) || !mentions(t, zone, own, inheritChild+":patch:") {
+ t.Error("together: the call's update and its start are not both in its own conversation, project and all")
+ }
+
+ // The update waits alone, and the start that follows names no thread.
+ // The update's thread is still the call's own: the call does not take
+ // the function's in its place, and a run below it does not pass over it
+ // to take the function's either.
+ const below = "14141414-1414-7414-8414-141414141417"
+ keyless := strings.Replace(inheritCall(``), `"end_time":"2026-09-20T10:00:02Z",`, ``, 1)
+ child := `{"id":"` + below + `","trace_id":"` + inheritRoot + `","parent_run_id":"` + inheritChild + `",` +
+ `"dotted_order":"` + order + `.20260920T100001500000Z` + below + `",` +
+ `"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":"the tool's own result"}}`
+ _, 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+`,`+child+`]}`)
+ if _, err := collector.Collect(); err != nil {
+ t.Fatal(err)
+ }
+ for _, text := range []string{`"FakeListChatModel"`, "the patched answer", "the tool's own result"} {
+ if mentions(t, zone, outer, text) {
+ t.Errorf("%s landed in the function's conversation, although the call named a thread of its own", text)
+ }
+ }
+}
diff --git a/internal/adapters/langsmith/pending.go b/internal/adapters/langsmith/pending.go
index 04b1ba2..6337791 100644
--- a/internal/adapters/langsmith/pending.go
+++ b/internal/adapters/langsmith/pending.go
@@ -76,6 +76,11 @@ 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 bool `json:"partial,omitempty"`
// At is when this was remembered, for the eviction above.
At string `json:"at,omitempty"`
}
@@ -244,6 +249,52 @@ func (p *pending) complete(run Run) (Run, pendingRun, bool) {
return run, known, true
}
+// 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 {
+ for _, id := range ancestors(run.Dotted, run.ID) {
+ known, ok := p.Runs[id]
+ if !ok {
+ return ""
+ }
+ if assigned(id, known) {
+ return known.Session
+ }
+ if known.Partial {
+ return ""
+ }
+ }
+ return ""
+}
+
+// partial records that a run named a thread before its project was known.
+func (p *pending) partial(id string) {
+ known := p.Runs[id]
+ known.Partial = true
+ p.Runs[id] = known
+}
+
+// assigned reports whether a run's conversation was supplied rather than
+// unassigned. An unassigned conversation is named for the run's trace, or for
+// the run when it arrived with no trace, so either name says it.
+func assigned(id string, known pendingRun) bool {
+ if known.Session == "" || known.Session == UnassignedID(id) {
+ return false
+ }
+ return known.Trace == "" || known.Session != UnassignedID(known.Trace)
+}
+
// sawRun remembers one run of a trace: that it was seen at all, and whether
// it was a tool.
func (p *pending) sawRun(run Run, now time.Time) {
diff --git a/tests/langchain_test.go b/tests/langchain_test.go
index 8a36f40..794e602 100644
--- a/tests/langchain_test.go
+++ b/tests/langchain_test.go
@@ -83,6 +83,11 @@ func TestALangChainConversationAssembles(t *testing.T) {
// and end sent as one post, its end time in milliseconds, each message
// serialized with its kind named only by the class that ends its id.
{"js-graph", 1, 1, 1},
+ // LangSmith JS's traceable around a chat model. The thread is on the
+ // function alone, and the model call takes it from the function it
+ // ran inside, so the question, the call and its answer are one
+ // conversation.
+ {"js-traceable", 1, 0, 1},
} {
t.Run(tc.kase, func(t *testing.T) {
zone, sessions := land(t, tc.kase)
diff --git a/tests/langchainjs_test.go b/tests/langchainjs_test.go
index 5bbed30..2f6fa62 100644
--- a/tests/langchainjs_test.go
+++ b/tests/langchainjs_test.go
@@ -140,40 +140,30 @@ func TestALangGraphJSConversationLandsWhatWasSaid(t *testing.T) {
// ls_method alone, the function was taken for a graph's own run and its
// arguments and result were dropped as a repeat; its system message is then
// named where it lands. LangSmith JS also passes the function's metadata to
-// none of the runs inside it, so the model call it made supplies no thread,
-// and lands under its trace, as any run that supplied none does.
+// none of the runs inside it, so the model call it made supplies no thread.
+// It takes the function's, the run it ran inside: landed under its trace, the
+// answer was in a conversation of its own and the question's had none.
func TestALangChainJSTracedFunctionKeepsItsArguments(t *testing.T) {
zone, sessions := land(t, "js-traceable")
- var thread, unassigned string
- for _, s := range sessions {
- if strings.HasPrefix(s, "ls-unassigned-") {
- unassigned = s
- } else {
- thread = s
- }
- }
- if thread == "" || unassigned == "" || len(sessions) != 2 {
- t.Fatalf("sessions %v, want the thread's and the model call's own", sessions)
+ if len(sessions) != 1 || strings.HasPrefix(sessions[0], "ls-unassigned-") {
+ t.Fatalf("sessions %v, want the thread's alone", sessions)
}
- kept := false
- for _, l := range transcriptLines(t, zone, thread) {
+ kept, answered := false, false
+ for _, l := range transcriptLines(t, zone, sessions[0]) {
if strings.Contains(l.raw, "You are a careful reviewer.") {
kept = strings.Contains(l.raw, `"limit"`)
if !slices.Contains(l.Flags, sessiondata.FlagSystemPrompt) {
t.Errorf("the function's arguments hold the system prompt unnamed: %.200s", l.raw)
}
}
- }
- if !kept {
- t.Error("the function's arguments were not kept")
- }
- answered := false
- for _, l := range transcriptLines(t, zone, unassigned) {
for _, p := range l.Parts {
answered = answered || p.Text == "Two pages changed: the setup page and the formats page."
}
}
+ if !kept {
+ t.Error("the function's arguments were not kept")
+ }
if !answered {
- t.Error("the model call's reply did not land")
+ t.Error("the model call's reply is not in the thread's conversation")
}
}