From 78df2f1de21825c760eb53d74728f7abafae3c30 Mon Sep 17 00:00:00 2001 From: Brian Love Date: Mon, 21 Sep 2026 20:25:00 -0700 Subject: [PATCH] feat: expose LangGraph state through native snapshots --- fixtures/react-parity/README.md | 24 +- fixtures/react-parity/runtime/README.md | 40 +- fixtures/react-parity/runtime/angular-app.ts | 1 + fixtures/react-parity/runtime/evidence.json | 246 ++++--- .../react-parity/runtime/installed-types.ts | 3 +- fixtures/react-parity/runtime/react-app.tsx | 1 + .../react-parity/runtime/runtime-entry.ts | 5 +- fixtures/react-parity/runtime/scenarios.ts | 10 +- libs/angular/README.md | 18 +- libs/angular/src/observe-agent.spec.ts | 44 +- libs/angular/src/observe-agent.ts | 13 +- libs/angular/src/observe-agent.type-test.ts | 37 ++ .../transport/fetch-stream.transport.spec.ts | 19 + .../lib/transport/fetch-stream.transport.ts | 8 +- libs/langgraph/src/runtime/create-session.ts | 80 ++- .../src/runtime/langgraph-snapshot.ts | 18 + libs/langgraph/src/runtime/ownership.ts | 54 ++ .../langgraph/src/runtime/publication.spec.ts | 50 +- libs/langgraph/src/runtime/publication.ts | 14 +- .../src/runtime/stream-projection.ts | 2 +- .../src/runtime/testing/binding-fixture.ts | 46 +- .../src/runtime/transport.integration.spec.ts | 79 +++ .../src/runtime/values-projection.spec.ts | 278 ++++++++ .../src/runtime/values-projection.ts | 52 ++ libs/langgraph/src/runtime/values.spec.ts | 619 ++++++++++++++++++ .../langgraph/src/runtime/values.type-test.ts | 42 ++ libs/react/README.md | 18 +- libs/react/src/use-agent.spec.tsx | 54 +- libs/react/src/use-agent.ts | 13 +- libs/react/src/use-agent.type-test.ts | 36 + scripts/react-parity/baseline.json | 33 +- scripts/react-parity/dispositions.json | 34 +- scripts/react-parity/runtime-consumer.mjs | 43 +- .../react-parity/runtime-consumer.spec.mjs | 12 + .../react-parity/verify-angular-package.mjs | 2 +- 35 files changed, 1825 insertions(+), 223 deletions(-) create mode 100644 libs/langgraph/src/runtime/langgraph-snapshot.ts create mode 100644 libs/langgraph/src/runtime/values-projection.spec.ts create mode 100644 libs/langgraph/src/runtime/values-projection.ts create mode 100644 libs/langgraph/src/runtime/values.spec.ts create mode 100644 libs/langgraph/src/runtime/values.type-test.ts diff --git a/fixtures/react-parity/README.md b/fixtures/react-parity/README.md index 0be99c106..aafbce4c5 100644 --- a/fixtures/react-parity/README.md +++ b/fixtures/react-parity/README.md @@ -13,7 +13,13 @@ errors, read-only reconciliation after uncertain failures and fixed function-too execution. The private session also supports explicit fixed-thread history loading when its transport can read history. The latest checkpoint replaces the transcript; equal reads preserve identity, failures preserve the prior snapshot, and loaded -tools never execute. Core contracts and native binding implementations are unchanged. +tools never execute. Private snapshots now include broad readonly application +`values`, published atomically with messages. Full root state, history and conclusive +correlated recovery replace the map; unchanged nested data retains identity, and +token-only updates reuse it without traversal. `undefined` is unobserved state and +`{}` an observed empty map. Child/update/custom/live-interrupt envelopes are ignored. +Core contracts are unchanged; native structural signatures preserve the concrete +snapshot extension and tool inference, with the same receiver and lifetime behavior. Angular and React borrow the app-owned session and observe it through their native lifecycles. The installed consumers run ten scenarios each: inert mount, explicit history load, equal refresh, empty replacement, text, weather tool @@ -22,8 +28,9 @@ unmount followed by explicit disposal and an aborted post-disposal submission. Five live component submissions plus one tool continuation produce six exact wire requests, with one handler invocation and zero page errors or unexpected requests. The three explicit loads make three history reads and no run requests or handler -calls. This is partial T10 coverage: thread switching, pagination, branching, full -backend state and interrupt resume remain outside this proof. +calls. This is partial T09/T10 coverage: thread switching, pagination, branching, +state writes, application-schema inference and interrupt resume remain outside this +proof. The factory and snapshot extension remain private. A local HTTP/SSE fixture serves production-built apps and writes real held response bytes. Browser assertions observe incremental DOM text and a native response-close @@ -33,12 +40,12 @@ exercise that subscription replay. See [runtime/README.md](./runtime/README.md) reproduction and [runtime/evidence.json](./runtime/evidence.json) for fresh commands, counts, source provenance, cleanup assertions and limitations. -The current inventory has **1,453 records**: the historical 1,438 plus ten private +The current inventory has **1,455 records**: the historical 1,438 plus twelve private runtime production sources, three testing helpers, a runtime Vitest config and its type-test config asset. Public export occurrences remain 550 with 514 distinct local definitions. The original runtime extraction changed fourteen existing export -records' declaration/import text; history loading adds two private sources and -changes no legacy public export records. There are no legacy export-name additions +records' declaration/import text; history loading and values observation each add +two private sources and change no legacy public export records. There are no legacy export-name additions or removals. Existing task assignments are preserved; touched extraction/configuration subsets are in progress, not whole T03–T16 completion. Core and native package contracts remain outside this legacy @@ -218,7 +225,7 @@ burst streams and repeated agent/thread disposal. The foundation branch was `codex/react-support-baseline`; the runtime branch is `codex/shared-runtime-quality`, based on `bdcc22ed31aa94f420077e046e88e1481088d453`. -The history-loading increment is `codex/langgraph-history-loading`; its verified +The current values increment is `codex/langgraph-state-values`; its verified base and working-source fingerprint are recorded in `runtime/evidence.json`. The local maintenance branch `codex/angular-maintenance-v0.2` points to released tag `v0.2.0` @@ -236,7 +243,8 @@ research documents. It is a scope map, not evidence that the tasks are complete. T01/T02 describe the foundation increment. The current G1 proof is deliberately limited to shared LangGraph text streaming and fixed function-tool execution with borrowed native Angular and React bindings: the runtime owns execution while each binding observes -it. Explicit fixed-thread history loading now covers a further subset of T10. +it. Explicit history loading and application-values observation cover further +subsets of T09/T10. Renderer reuse and SSR are deferred gates, alongside the broader T01–T39 map. This bounded runtime proof does not establish complete migration parity. diff --git a/fixtures/react-parity/runtime/README.md b/fixtures/react-parity/runtime/README.md index 474a8b0d3..0fb8974b5 100644 --- a/fixtures/react-parity/runtime/README.md +++ b/fixtures/react-parity/runtime/README.md @@ -27,13 +27,16 @@ or backend SDK; Angular installs no React or backend SDK. Framework/compiler versions come from the root lockfile. Contract probes compile the installed public entries with `strict` and `skipLibCheck:false`, standard DOM signals, and no workspace aliases. Negative probes check names, arguments, results and deep -readonly types directly on each binding's inferred snapshot. +readonly types directly on each binding's inferred snapshot, including broad +backend values without application-schema inference. `runtime-entry.ts` is development-only composition around private `createSession`, the production `FetchStreamTransport`, and the real LangGraph SDK. A focused TypeScript check resolves its public core imports against the installed tarball -declarations, then emits its narrow annotated `AgentSession` return -type with an optional fixture `load` capability. Vite bundles the private backend +declarations, then emits its narrow annotated fixture return type. It replaces +the core getter with `Omit, 'getSnapshot'>` and a concrete +snapshot getter, avoiding an intersected overload that would hide `values` from +inference; `load` remains optional. Vite bundles the private backend and SDK into temporary ESM, externalizing `@threadplane/core` and `@threadplane/core/tools`. Only that JavaScript bundle and entry declaration are copied into each installed consumer. No private TypeScript, @@ -48,6 +51,27 @@ transport owns its retry policy. Canonical updates may replace or remove pending tool calls for the same assistant message while retaining other messages' calls and completed results. +The private `LangGraphSnapshot.values` is a broad +`Readonly> | undefined`. It is observed application data, +not a validated application schema. `undefined` means no current values map is +observed; `{}` means a root record was observed with no application fields. Root +`values` and `checkpoints` records, explicit history, and conclusive recovery +correlated to the attempted run replace the whole map, including deleted fields. +`messages` and `__interrupt__` are excluded. Missing history values clear the map +to `undefined`. Child streams, node updates, custom events and live interrupt +envelopes do not replace it. + +Messages and values publish together as one owned immutable snapshot. Equal maps +retain identity, changed maps share unchanged nested branches, and token-only +updates reuse the owned map without traversing it. This adds no I/O. Narrow SDK +normalization prevents raw data fields from overriding protocol type/namespace; +`messageMetadata` selects delta text semantics only for actual message events. +The application field remains observable as data. The native bindings infer +the concrete snapshot through structural getter/subscription signatures while +retaining typed tool results, method receivers and borrowed lifetime semantics. +The factory and snapshot extension remain private; state writes, application +schema inference, SSR and package-root cutover remain outside this slice. + The private `LangGraphSession` offers `load({ signal })` only when its transport supports history reads. Loading is explicit: construction, mount and subscription perform no I/O. The latest checkpoint authoritatively replaces the transcript, @@ -63,12 +87,12 @@ History is observation only: loading never executes pending tools. Execution deduplication survives a load, while locally authored result provenance is cleared. Persisted ToolMessage strings remain transcript text rather than becoming typed handler results, including on later stream replay. This is a fixed-thread history -subset of T10, not thread switching, pagination, branching, full backend state, +subset of T10, not thread switching, pagination, branching, state writes, interrupt resume, SSR, or a public LangGraph package cutover. Core public contracts -and the native binding implementations are unchanged. +are unchanged; the native signatures now retain the concrete snapshot extension. The native fixtures expose Load, Send, Tool, Error, Hold and Stop buttons plus text, -transcript, load completion/error, status, tool result, delivery, submission and +transcript, values, load completion/error, status, tool result, delivery, submission and handler count outputs. A single app-owned session is created outside component lifetime and outside React's StrictMode tree; owner buttons perform framework unmount and explicit session disposal. @@ -89,6 +113,10 @@ and no run requests or handler calls. Every completed load must leave its visibl error output empty, so retained text cannot conceal a failed equal refresh. Both registered handlers increment the same counter if executed. Request bodies check the catalog and actual serialized ToolMessage payload. +Values assertions distinguish unobserved from empty state, show loaded application +fields, and verify replacement/deletion across root, tool, held and reused runs. +Separate native component tests make four history reads to cover a values-only +refresh with unchanged messages; installed browser scenarios still make three. A small in-process HTTP fixture serves only built artifacts and the expected LangGraph run/history routes on dynamic port 0. The held response writes an actual SSE diff --git a/fixtures/react-parity/runtime/angular-app.ts b/fixtures/react-parity/runtime/angular-app.ts index c15697590..8623fda32 100644 --- a/fixtures/react-parity/runtime/angular-app.ts +++ b/fixtures/react-parity/runtime/angular-app.ts @@ -23,6 +23,7 @@ const submit = (input: string) => { submissions += 1; return session.submit(inpu {{ snapshot().status }} {{ view().text }} {{ view().transcript }} + {{ view().values }} {{ loadsFinished() }} {{ loadError() }} {{ view().error }} diff --git a/fixtures/react-parity/runtime/evidence.json b/fixtures/react-parity/runtime/evidence.json index 2bf822a74..ee65e38a6 100644 --- a/fixtures/react-parity/runtime/evidence.json +++ b/fixtures/react-parity/runtime/evidence.json @@ -1,14 +1,14 @@ { "schemaVersion": 1, "status": "verified-local", - "increment": "Explicit fixed-thread history loading (partial T10)", + "increment": "Readonly application-state values observation (partial T09/T10)", "observedOn": "2026-09-21", - "recordedAt": "2026-09-22T02:09:48.896Z", + "recordedAt": "2026-09-22T03:19:34.895Z", "source": { - "branch": "codex/langgraph-history-loading", - "baseCommit": "e1da2bd10d0f009924eeb2ea67203da92d71cd0b", - "verificationHead": "e1da2bd10d0f009924eeb2ea67203da92d71cd0b", - "workingTree": "Verified uncommitted history-loading, fixture, test and metadata changes on the runtime integration HEAD. The fingerprint identifies the actual selected working-tree bytes. No future commit or future CI result is claimed.", + "branch": "codex/langgraph-state-values", + "baseCommit": "192f94e9b57b77e1858e2e4337f0d34bcd6a9c64", + "verificationHead": "192f94e9b57b77e1858e2e4337f0d34bcd6a9c64", + "workingTree": "Verified uncommitted values-projection, native signature, transport normalization, fixture/test and metadata changes on the merged history-loading HEAD. The fingerprint identifies actual selected working-tree bytes; no future commit or future CI result is claimed.", "fingerprint": { "algorithm": "SHA-256 of a UTF-8 manifest: one line per selected file, lowercase SHA-256(file bytes), two ASCII spaces, repo-relative path, LF; unique paths sorted by JavaScript default string ordering.", "pathspecs": [ @@ -38,9 +38,9 @@ "excludedPaths": [ "fixtures/react-parity/runtime/evidence.json" ], - "fileCount": 770, - "sha256": "5ad0835c76b0a039165f1cc1419a553690d25c8c0a34452211aeb69e5e5002b2", - "selection": "git ls-files -z --cached --others --exclude-standard -- ; keep existing files and remove excludedPaths. Includes tracked and non-ignored untracked files. Evidence itself is excluded to avoid self-reference; the seven local research/planning reports are outside all selected paths.", + "fileCount": 775, + "sha256": "0ec1ffeca45fa046535ccd6f95c3754735ab5de0f8a355d23afe41f0dbe1d9ab", + "selection": "git ls-files -z --cached --others --exclude-standard -- ; keep existing files and remove excludedPaths. Includes tracked and non-ignored untracked files. Evidence itself is excluded to avoid self-reference; eight local research/planning reports are outside all selected paths.", "reproduce": "node --input-type=module <<'JS'\nimport {createHash} from 'node:crypto';\nimport {execFileSync} from 'node:child_process';\nimport {readFileSync,existsSync} from 'node:fs';\nconst {fingerprint:f}=JSON.parse(readFileSync('fixtures/react-parity/runtime/evidence.json')).source;\nconst sha=value=>createHash('sha256').update(value).digest('hex');\nconst paths=[...new Set(execFileSync('git',['ls-files','-z','--cached','--others','--exclude-standard','--',...f.pathspecs],{encoding:'utf8'}).split('\\0').filter(Boolean))].filter(path=>!f.excludedPaths.includes(path)&&existsSync(path)).sort();\nconst actual=sha(paths.map(path=>sha(readFileSync(path))+' '+path+'\\n').join(''));\nif(paths.length!==f.fileCount||actual!==f.sha256) throw new Error('Source fingerprint mismatch');\nconsole.log(paths.length+' files: '+actual);\nJS" }, "sourceState": { @@ -48,26 +48,39 @@ "fixtures/react-parity/README.md", "fixtures/react-parity/runtime/README.md", "fixtures/react-parity/runtime/angular-app.ts", + "fixtures/react-parity/runtime/installed-types.ts", "fixtures/react-parity/runtime/react-app.tsx", "fixtures/react-parity/runtime/runtime-entry.ts", "fixtures/react-parity/runtime/scenarios.ts", + "libs/angular/README.md", "libs/angular/src/observe-agent.spec.ts", + "libs/angular/src/observe-agent.ts", + "libs/angular/src/observe-agent.type-test.ts", + "libs/langgraph/src/lib/transport/fetch-stream.transport.spec.ts", + "libs/langgraph/src/lib/transport/fetch-stream.transport.ts", "libs/langgraph/src/runtime/create-session.ts", + "libs/langgraph/src/runtime/ownership.ts", + "libs/langgraph/src/runtime/publication.spec.ts", + "libs/langgraph/src/runtime/publication.ts", "libs/langgraph/src/runtime/stream-projection.ts", "libs/langgraph/src/runtime/testing/binding-fixture.ts", "libs/langgraph/src/runtime/transport.integration.spec.ts", + "libs/react/README.md", "libs/react/src/use-agent.spec.tsx", + "libs/react/src/use-agent.ts", + "libs/react/src/use-agent.type-test.ts", "scripts/react-parity/baseline.json", "scripts/react-parity/dispositions.json", "scripts/react-parity/runtime-consumer.mjs", - "scripts/react-parity/runtime-consumer.spec.mjs" + "scripts/react-parity/runtime-consumer.spec.mjs", + "scripts/react-parity/verify-angular-package.mjs" ], "untracked": [ - "libs/langgraph/src/runtime/history-projection.spec.ts", - "libs/langgraph/src/runtime/history-projection.ts", - "libs/langgraph/src/runtime/history.spec.ts", - "libs/langgraph/src/runtime/history.type-test.ts", - "libs/langgraph/src/runtime/wire-message.ts" + "libs/langgraph/src/runtime/langgraph-snapshot.ts", + "libs/langgraph/src/runtime/values-projection.spec.ts", + "libs/langgraph/src/runtime/values-projection.ts", + "libs/langgraph/src/runtime/values.spec.ts", + "libs/langgraph/src/runtime/values.type-test.ts" ] } }, @@ -102,10 +115,10 @@ { "command": "NX_DAEMON=false node --test scripts/ci-scope.spec.mjs scripts/ci-workflow.spec.mjs scripts/react-parity/*.spec.mjs fixtures/react-parity/traces.spec.mjs", "exitCode": 0, - "testsPassed": 412, + "testsPassed": 413, "testsFailed": 0, - "log": "/tmp/h04-final-focused.log", - "logSha256": "8d929ce23bb0c125cdbb859166c4a9fef70b8304ae4de7fa0bc12ee035c77fe8" + "log": "/tmp/v04-final-focused.log", + "logSha256": "6415b5511886a8f22b7681fd61e0d0a125372f0a12ffaae04282c968b9bf7bed" }, { "command": "NX_DAEMON=false npx nx run-many -t lint test type-tests build --projects=core,content,angular,react --parallel=2 --skip-nx-cache", @@ -118,21 +131,21 @@ "react": 7 }, "content": "Empty scaffold with passWithNoTests; no content behavior count claimed.", - "log": "/tmp/h04-final-foundations.log", - "logSha256": "499d4f3a8ed8fdd6af1c26252aa418a9f70b667ce6fb28c39f5f4793d654d18c" + "log": "/tmp/v04-final-foundations.log", + "logSha256": "01c0cf761887e34b3411d98372c328d0f1fda388ce6f51b0708a8ffb58ac4230" }, { "command": "NX_DAEMON=false npx nx run langgraph:runtime-quality --skip-nx-cache", "exitCode": 0, - "testFiles": 11, - "testsPassed": 207, - "log": "/tmp/h04-final-runtime.log", - "logSha256": "6b18035d54c771dfba77a48230512e9001b268075bc834e7f5b13dc5d4fd2e1c" + "testFiles": 13, + "testsPassed": 270, + "log": "/tmp/v04-final-runtime.log", + "logSha256": "c447e885bbe74aa3a2c8da9bd46e7db99b6e8953cb9cbcc40d16a9e1d92857db" }, { "command": "NX_DAEMON=false npx nx run langgraph:runtime-type-tests --skip-nx-cache", "exitCode": 0, - "log": "/tmp/h04-final-runtime-types.log", + "log": "/tmp/v04-final-runtime-types.log", "logSha256": "6e5c64695f3aad8fd1b474649ebb26c81204d54e45d589424d65e2b0843c921d" }, { @@ -140,27 +153,34 @@ "exitCode": 0, "errors": 0, "existingWarnings": 68, - "log": "/tmp/h04-final-langgraph-lint.log", - "logSha256": "1fc98b0ca0707512f2ebf1084ed37e8e979f82be6b3cfb7a7194e17795710432" + "log": "/tmp/v04-final-langgraph-lint.log", + "logSha256": "3d2546b889adf06c0dedeb1966a4e9614458ce84bf40678192f46a7df21d4af4" + }, + { + "command": "NX_DAEMON=false npx nx test langgraph --testFile=fetch-stream.transport.spec.ts --testFile=fetch-stream.transport.integration.spec.ts --testFile=create-langgraph-client.spec.ts --testFile=client-options.spec.ts --skip-nx-cache", + "exitCode": 0, + "reason": "Fresh selected legacy regression run after the transport source changed. Nx success is recorded; no undisplayed test count is inferred.", + "log": "/tmp/v04-final-legacy.log", + "logSha256": "d13fb1fe2f58b24cacb8d09fde4a4b08fcd8f5735cff62dbe1ee840b72414e4f" }, { "command": "NX_DAEMON=false npx nx run-many -t build --projects=chat,langgraph,ag-ui,render,a2ui,telemetry --configuration=production --parallel=2 --skip-nx-cache", "exitCode": 0, "productionProjects": 6, - "log": "/tmp/h04-final-production-builds.log", - "logSha256": "b757c69ada7ddf18e2ef60f8b702aec960243099cea7b66ab3af8d54c445d453" + "log": "/tmp/v04-final-production-builds.log", + "logSha256": "6ab27d4499cf297148dde8b8699232b230ca86d085ca4d26b9730fe874617d7b" }, { "command": "node scripts/react-parity/verify-boundaries.mjs", "exitCode": 0, - "log": "/tmp/h04-final-source-boundaries.log", + "log": "/tmp/v04-final-source-boundaries.log", "logSha256": "995db896b38cf7de5ca9db6590dba47e082d26abdf69bab04b92eee484696063" }, { "command": "node scripts/react-parity/verify-boundaries.mjs --built", "exitCode": 0, "order": "Executed after successful foundation and six legacy production builds.", - "log": "/tmp/h04-final-built-boundaries.log", + "log": "/tmp/v04-final-built-boundaries.log", "logSha256": "c983a97d17ff7ced7aa9113afa031f464103f566f1bc94eec8160558ec2553f7" }, { @@ -170,62 +190,45 @@ "esmTypeExports": 9, "isolatedCoreExports": 3, "browserScenariosPassed": 10, - "log": "/tmp/h04-final-packages.log", - "logSha256": "8e749830c928e0102cb2b74c34361667d62022c2767fe1d36d11acd338213943" + "log": "/tmp/v04-final-packages.log", + "logSha256": "1e6e80a4a9c4568bfe6639d61d168ee7ffa9b08e7a435f57c4615c0e231c5075" }, { "command": "node scripts/react-parity/verify-angular-package.mjs", "exitCode": 0, "angularAPFExports": 1, "browserScenariosPassed": 10, - "log": "/tmp/h04-final-angular-package.log", - "logSha256": "908c659075d0bba45f969d7bed4a268cafca74eda94bb1128527aa67f6e8a080" - }, - { - "command": "NX_DAEMON=false npx nx run-many -t test --projects=angular,react --parallel=2 --skip-nx-cache", - "exitCode": 0, - "behaviorTests": { - "angular": 7, - "react": 7 - }, - "reason": "Fresh native tests after replacing a test-fixture non-null assertion with an explicit capability guard. Installed browser bundles exclude this helper; runtime production code and browser fixture inputs were unchanged.", - "log": "/tmp/h04-final-native-after-lint-fix.log", - "logSha256": "d6cfb218eddc90aea5872a9bb60f7c4fae8843adb753639ad853b62066d01bc3" + "order": "Successful retry after production builds completed; earlier artifact-availability failure is recorded separately.", + "log": "/tmp/v04-final-angular-package.log", + "logSha256": "ec02d28e5233ca0ba1276047d5dd83ed72048a6e0c1e3d2e3eebf995607ba07f" }, { "command": "node scripts/react-parity/inventory.mjs --write-baseline; node scripts/react-parity/inventory.mjs --check", "exitCode": 0, - "inventoryRows": 1453, - "reviewedChange": "Two private production sources added; three existing private source hashes changed. No public export drift or existing assignment changes.", - "log": "/tmp/h04-final-inventory.log", - "logSha256": "97f59e86a59a01fc04a5802a60c8b439a30f922131445dbbdbb8c4cbe1003293" + "inventoryRows": 1455, + "reviewedChange": "Two private production sources added; six existing source hashes changed. No legacy public export drift or prior disposition reassignment.", + "log": "/tmp/v04-final-inventory.log", + "logSha256": "757a321eb090a7dc7ef343f29202a3ad540d3bd70f386161e11271f06abefb0f" }, { "command": "git diff --check", "exitCode": 0, - "log": "/tmp/h04-final-diff-check.log", + "log": "/tmp/v04-final-diff-check.log", "logSha256": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" } ], - "reusedUnchangedLegacyChecks": [ - { - "commandAsPrintedByNx": "nx run langgraph:test --testFile=fetch-stream.transport.spec.ts --testFile=fetch-stream.transport.integration.spec.ts --testFile=create-langgraph-client.spec.ts --testFile=client-options.spec.ts", - "exitCode": 0, - "log": "/tmp/h02-legacy-green.log", - "logSha256": "f95c0a34f354e73d7b94871ca4e048b0ef48fc7c8c356575e3398d42c6cec60d", - "provenance": "Same-increment H02 selected legacy transport/client run; success output inspected. Legacy source and these tests are unchanged since that run. Original launcher flags and a test count are not inferred from the Nx log." - } - ], "scope": { - "privateCapability": "LangGraphSession.load?(options?: { readonly signal?: AbortSignal }): Promise, exposed only when the captured transport supports history reads.", - "projection": "Latest checkpoint is authoritative: replace/delete/reorder messages, accept shorter/empty corrections, own immutable data and share unchanged explicit IDs. Equal reads retain snapshot identity.", - "lifecycle": "Construction/mount remain inert; pending or failed loads preserve snapshots. Read failure rejects protected diagnostics; abort/supersession/stop/dispose settle promptly and prevent stale publication even with a noncooperative transport.", - "admission": "Refuses active execution, uncertain recovery, staged tool results and pending claim/record/write work before issuing history I/O.", - "toolSafety": "History loading never executes tools. Execution deduplication is retained; successful loads clear locally authored result provenance so persisted wire strings cannot masquerade as typed results on later stream replay.", - "unchanged": "No core public API or native binding implementation changes. Legacy LangGraph package root remains Angular; the factory and history capability remain private and fixture-only." + "privateSnapshot": "LangGraphSnapshot extends the core snapshot with readonly values: Readonly> | undefined. Undefined means no current application-values map; an empty object is an observed empty root.", + "projection": "Root full values/checkpoints, explicit latest history and conclusive correlated recovery replace the map, including deletions. Only messages and __interrupt__ are excluded from application values. Missing authoritative history clears values to undefined.", + "excludedEvents": "Child namespaces, updates, custom events and live interrupt envelopes do not replace root application values.", + "publication": "Messages and values are owned and published as one immutable aggregate. Equal state retains root identity; changed state shares unchanged nested branches. Token-only updates reuse owned values without traversing them.", + "normalization": "Legacy SDK normalization protects protocol type/namespace from raw fields. messageMetadata selects delta text semantics only for actual message events; a same-named application field remains observable.", + "nativeInference": "Structural getSnapshot/subscribe parameters retain concrete backend snapshot extensions, typed tool results, method receivers and borrowed lifetimes. Omit replaces the getter instead of intersecting incompatible overloads.", + "history": "Explicit fixed-thread load remains optional; pending/failing/stale reads, active work admission, historical tool non-execution and execution deduplication retain the previously verified behavior.", + "boundary": "No core public API changes, application-schema inference, state writes, new I/O, SSR or public factory/root cutover. Private runtime and fixture extensions remain unpublished." }, "acceptanceMatrix": { - "status": "Both production-built installed browser verifiers passed.", + "status": "Both production-built installed browser verifiers passed, including values assertions within the existing ten scenarios.", "frameworks": [ "Angular installed APF production app", "React installed Vite production app" @@ -289,7 +292,35 @@ "toolAnswerContains": "20 degrees", "heldTextContains": "Held partial", "stopDelivery": "complete:aborted", - "postDisposalOutcome": "aborted" + "postDisposalOutcome": "aborted", + "values": { + "inertMount": "unobserved", + "historyLoad": { + "stage": "saved", + "profile": { + "name": "Saved user" + } + }, + "equalRefresh": { + "stage": "saved", + "profile": { + "name": "Saved user" + } + }, + "emptyHistoryReplacement": "unobserved", + "textSuccess": { + "stage": "complete" + }, + "toolRoundtrip": {}, + "protectedError": {}, + "heldAndStopped": { + "stage": "held", + "transient": true + }, + "reuseAfterStop": { + "stage": "complete" + } + } }, "notificationAssertions": { "source": "libs/langgraph/src/runtime/history.spec.ts", @@ -299,19 +330,32 @@ "emptyReplacementCumulative": 2, "failedRefreshAdditional": 0, "limit": "Unit-test behavioral counts, not render/paint or performance measurements." + }, + "nativeHistoryObservation": { + "readsPerFrameworkTest": 4, + "coverage": "Initial load, equal refresh, values-only replacement with unchanged messages, then empty history. Installed browser scenarios still use three reads.", + "lifetime": "No mount reads; observer teardown does not own session execution/disposal." + }, + "valuesUnitEvidence": { + "sources": [ + "libs/langgraph/src/runtime/values-projection.spec.ts", + "libs/langgraph/src/runtime/values.spec.ts", + "libs/langgraph/src/runtime/publication.spec.ts" + ], + "coverage": "Root/checkpoint/history/recovery replacement and deletion; nested sharing/no-op identity; atomic publication; stale candidate guards; immutable caller isolation; unobserved versus empty; ignored child/control envelopes; no token-time values traversal. These are behavioral assertions, not performance measurements." } }, "cleanup": { - "assertions": "Unmount removes component controls; app-owned disposal resolves; post-disposal submit resolves aborted without extra I/O. Only three explicit loads read history. Held SSE waits for the native response-close handshake.", + "assertions": "Unmount removes component controls; app-owned disposal resolves; post-disposal submit resolves aborted without extra I/O. Only three explicit browser loads read history. Held SSE waits for native response close.", "resources": "Browser contexts, browsers, held responses, server connections and temporary install/bundle artifacts close in finally, including assertion failures.", "processes": "Both final verifier processes exited 0 after awaited cleanup.", "temporaryDirectories": [ { - "path": "/var/folders/_b/0t5_pyt94n7dlqkv1gmt29300000gn/T/threadplane-consumer-dtU2kZ", + "path": "/var/folders/_b/0t5_pyt94n7dlqkv1gmt29300000gn/T/threadplane-consumer-x4TyEp", "existsAfterExit": false }, { - "path": "/var/folders/_b/0t5_pyt94n7dlqkv1gmt29300000gn/T/threadplane-angular-consumer-3OGMOu", + "path": "/var/folders/_b/0t5_pyt94n7dlqkv1gmt29300000gn/T/threadplane-angular-consumer-zl68Zs", "existsAfterExit": false } ], @@ -319,19 +363,19 @@ }, "inventory": { "historicalFoundationRows": 1438, - "previousRuntimeRows": 1451, - "currentRows": 1453, - "newHistorySourceFiles": [ - "libs/langgraph/src/runtime/history-projection.ts", - "libs/langgraph/src/runtime/wire-message.ts" + "previousHistoryRows": 1453, + "currentRows": 1455, + "newValuesSourceFiles": [ + "libs/langgraph/src/runtime/langgraph-snapshot.ts", + "libs/langgraph/src/runtime/values-projection.ts" ], - "sourceFiles": 475, + "sourceFiles": 477, "legacyExports": 550, "uniqueLocalDefinitions": 514, "legacyExportRecordChangesThisIncrement": 0, - "existingAssignmentsPreserved": 1451, - "dispositions": "Existing IDs, task assignments, treatments, reasons and statuses preserved; three scoped notes updated and two internal in-progress rows added under T10 and T09/T10. No whole-task completion claim.", - "scope": "Existing 16-library inventory only. Core/native contracts remain separately checked. Historical baseline-evidence.json is untouched; the previous runtime evidence remains in Git history." + "existingAssignmentsPreserved": 1453, + "dispositions": "All prior IDs, task assignments, treatments, reasons and statuses preserved. Six scoped notes updated; two internal in-progress rows added under T09/T10. No whole-task completion claim.", + "scope": "Existing 16-library inventory only; core/native contracts are separately checked. Historical baseline-evidence.json untouched; prior runtime/history evidence retained in Git history." }, "runtimePolicy": { "ownedSDKDefaultMaxRetries": 0, @@ -347,11 +391,11 @@ }, "reactConsumer": { "installedPackages": 24, - "fileBytes": 51239870, + "fileBytes": 51240541, "lockLocationsIncludingOptionalPlatforms": 74, "productionAppModulesTransformed": 34, - "productionAppJavaScriptRawReported": "412.73 kB", - "productionAppJavaScriptGzipReported": "124.43 kB", + "productionAppJavaScriptRawReported": "414.48 kB", + "productionAppJavaScriptGzipReported": "124.92 kB", "developmentRootImportProbe": { "inputs": 5, "bytes": 47778, @@ -360,35 +404,47 @@ }, "angularConsumer": { "installedPackages": 411, - "fileBytes": 196915823, + "fileBytes": 196916614, "lockLocationsIncludingOptionalPlatforms": 516, "bundleInputs": 257, "contentParserInputs": 0, - "productionAppRawReported": "314.20 kB", - "productionAppEstimatedTransferReported": "83.30 kB", + "productionAppRawReported": "315.92 kB", + "productionAppEstimatedTransferReported": "83.75 kB", "includes": "Angular CLI/compiler/build tooling" }, - "limits": "Installed files, optional lock locations, development root-import probes and full production apps are separate diagnostics. App bundles include the staged SDK. These are not performance benchmarks or comparable framework overhead measurements." + "limits": "Installed files, optional lock locations, development probes and production apps are separate diagnostics. App bundles include the staged SDK. These are not performance benchmarks or comparable framework overhead measurements." }, "ci": { "definition": "Existing library lane covers runtime, source/built boundaries and packed browser checks; it installs Chromium with --with-deps before browser verification.", - "result": "No new history-loading PR CI result is claimed by this local record. Prior runtime integration CI does not verify these uncommitted history changes." + "result": "No new values PR CI result is claimed by this local record. Prior merged history/runtime CI does not verify these uncommitted changes." }, "limits": [ - "Fixed-thread latest-history transcript projection only; no thread switching, pagination, branching, full backend state or interrupt resume.", - "Private staged runtime is fixture-only; no neutral LangGraph tarball or public factory/root cutover.", - "The narrow fixture declaration, including optional load, is compiler-generated against installed core; it is not hand-written or a new core API.", - "Native inferred type probes retain heterogeneous tool and readonly checks; observation bindings themselves are unchanged.", + "Read-only broad application-data observation only; no application-schema inference, state writes, thread switching, pagination, branching or interrupt resume.", + "Private staged runtime/factory and snapshot extension remain fixture-only; no neutral LangGraph tarball or public root cutover.", + "The narrow fixture declaration is compiler-generated against installed core; Omit replaces its getter while preserving the concrete snapshot.", + "Native signatures retain backend fields and tool inference; lifecycle ownership remains with the app.", "HTTP/SSE proves incremental DOM updates and native abort, not compositor timing.", "Production React StrictMode does not replay development effects; native unit tests separately exercise replay.", - "No full product parity, renderer reuse, SSR, calibrated latency, retained-heap or performance claim.", + "No full parity, renderer reuse, SSR, calibrated latency, retained-heap or performance claim.", "CI Node22/Linux and Angular20/22 consumer lanes were not run locally.", "Test-only imports still add Nx legacy build dependencies although production package graphs remain isolated.", "No production backend, release, deployment or publication was exercised." ], "warnings": [ - "LangGraph lint passes with 68 existing warnings and no errors; the one newly introduced test-helper non-null assertion was removed and both native suites rerun.", + "LangGraph lint passes with 68 existing warnings and no errors.", "Nx reports NO_COLOR/FORCE_COLOR overlap; legacy builds report stale Browserslist data, ng-packagr export-condition overrides and keepLifecycleScripts notices.", "Vite ignores use client directives in the client-only production consumer bundle; no SSR claim is made." - ] + ], + "verificationRetry": { + "command": "node scripts/react-parity/verify-angular-package.mjs", + "initialExitCode": 1, + "log": "/tmp/v04-angular-before-build-complete.log", + "logSha256": "62a271c8ae1aba31afa623380080b550015b46f4ea18c421f2ac343441efe5f3", + "cause": "Verifier was initially scheduled concurrently with legacy production builds, which rebuild core; it saw ENOENT for dist/libs/core/package.json. This was local verification ordering, not a source failure.", + "resolution": "Waited for successful build completion, confirmed the core manifest existed, then reran the unchanged verifier successfully. No production/test changes were made." + }, + "documentation": { + "generatorsRun": [], + "reason": "The private runtime and native package READMEs are outside the registered legacy API/narrative/agent-context generation inputs. Legacy export records are unchanged; only directly affected README guidance was updated." + } } diff --git a/fixtures/react-parity/runtime/installed-types.ts b/fixtures/react-parity/runtime/installed-types.ts index 22d5c79ec..e7dd00e6b 100644 --- a/fixtures/react-parity/runtime/installed-types.ts +++ b/fixtures/react-parity/runtime/installed-types.ts @@ -1,10 +1,11 @@ -import type { AgentSession, AgentSnapshot, ToolCall } from '@threadplane/core'; +import type { AgentSession, AgentSnapshot, PlainValue, ToolCall } from '@threadplane/core'; import type { FunctionTool } from '@threadplane/core/tools'; import type { FixtureTools } from './scenarios'; /* BINDING_IMPORT */ export function assertSnapshot(snapshot: AgentSnapshot) { + /* BACKEND_VALUES */ for (const call of snapshot.toolCalls) { if (call.name === 'weather') { const city: string = call.args.city; diff --git a/fixtures/react-parity/runtime/react-app.tsx b/fixtures/react-parity/runtime/react-app.tsx index 8abbd55bf..5099c128e 100644 --- a/fixtures/react-parity/runtime/react-app.tsx +++ b/fixtures/react-parity/runtime/react-app.tsx @@ -32,6 +32,7 @@ function App() { {snapshot.status} {view.text} {view.transcript} + {view.values} {loadsFinished} {loadError} {view.error} diff --git a/fixtures/react-parity/runtime/runtime-entry.ts b/fixtures/react-parity/runtime/runtime-entry.ts index 644edd326..ac5181a6b 100644 --- a/fixtures/react-parity/runtime/runtime-entry.ts +++ b/fixtures/react-parity/runtime/runtime-entry.ts @@ -1,14 +1,15 @@ import type { AgentSession } from '@threadplane/core'; // eslint-disable-next-line @nx/enforce-module-boundaries -- This development-only entry composes private source into a temporary fixture bundle, never a package export. import { createSession } from '../../../libs/langgraph/src/runtime/create-session'; -import type { FixtureTools } from './scenarios'; +import type { FixtureSnapshot, FixtureTools } from './scenarios'; /** Development-only composition, never a package entry or a shipped factory. */ export function createFixtureSession( endpoint: string, threadId: string, onHandler: () => void = () => undefined -): AgentSession & { +): Omit, 'getSnapshot'> & { + getSnapshot(): FixtureSnapshot; load?: (options?: { signal?: AbortSignal }) => Promise; } { return createSession({ diff --git a/fixtures/react-parity/runtime/scenarios.ts b/fixtures/react-parity/runtime/scenarios.ts index c74d4256f..6ec22c8fe 100644 --- a/fixtures/react-parity/runtime/scenarios.ts +++ b/fixtures/react-parity/runtime/scenarios.ts @@ -1,16 +1,22 @@ -import type { AgentSession, AgentSnapshot } from '@threadplane/core'; +import type { AgentSession, AgentSnapshot, PlainValue } from '@threadplane/core'; export interface FixtureTools { weather: { args: { city: string }; result: { city: string; temperature: number } }; count: { args: { values: readonly string[] }; result: number }; } -export function display(snapshot: AgentSnapshot) { +/** Fixture-local backend extension, expressed entirely through installed core. */ +export type FixtureSnapshot = AgentSnapshot & { + readonly values: Readonly> | undefined; +}; + +export function display(snapshot: FixtureSnapshot) { const assistant = snapshot.messages.filter((message) => message.role === 'assistant'); const delivery = assistant.at(-1)?.delivery; return { text: assistant.map((message) => message.content).join('\n'), transcript: snapshot.messages.map((message) => message.content).join('\n'), + values: JSON.stringify(snapshot.values) ?? 'unobserved', error: snapshot.error?.message ?? '', tool: JSON.stringify(snapshot.toolCalls), delivery: delivery?.phase === 'complete' ? `complete:${delivery.outcome}` : delivery?.phase ?? '', diff --git a/libs/angular/README.md b/libs/angular/README.md index f01734ab9..c32aa1d22 100644 --- a/libs/angular/README.md +++ b/libs/angular/README.md @@ -1,9 +1,10 @@ # @threadplane/angular -Private, unpublished Angular binding for app-owned `AgentSession` values from -`@threadplane/core`. The root exports `observeAgent(session)`, which returns a -read-only `Signal`. Tool names, arguments, and results retain the -session's declared types. +Private, unpublished Angular binding for app-owned sessions. The root exports +`observeAgent(session)`, which accepts `getSnapshot()` and `subscribe(notify)` +methods and returns a read-only `Signal`. The concrete snapshot must +extend the core `AgentSnapshot`; its additional fields and tool names, arguments, +and results retain their inferred types. Call `observeAgent` in an Angular injection context, such as a component field initializer or provider factory, with a session supplied by the app: @@ -19,11 +20,20 @@ export function observeStatus(session: AgentSession) { ``` The function borrows the session. Reading or subscribing does not start a run. +Snapshot reads and subscription calls preserve the session method receiver. The injected `DestroyRef` releases only this observer's subscription when its context is destroyed. Pending work continues, and other observers remain connected. The app calls `session.submit(text)`, `session.stop()`, and `session.dispose()` and owns the session's lifetime. +Keep the concrete session type when observing backend-specific fields. The private +LangGraph fixture exposes a broad readonly `values` map on its snapshot: `undefined` +means no current application-values map is observed, while `{}` is an observed +empty map. The binding preserves that field without inferring an application +schema, validating values, or issuing extra reads. Values and messages arrive in +the same immutable snapshot. This does not make the private backend factory public +or add state-writing, SSR, or hydration support. + No backend constructor is exported here; the current real LangGraph runtime is still a private development composition seam. This API is under development and does not claim full SSR, hydration, or feature parity. The Angular peer range diff --git a/libs/angular/src/observe-agent.spec.ts b/libs/angular/src/observe-agent.spec.ts index 94c2a81ef..d08858d61 100644 --- a/libs/angular/src/observe-agent.spec.ts +++ b/libs/angular/src/observe-agent.spec.ts @@ -31,6 +31,7 @@ const SESSION = new InjectionToken< {{ snapshot().status }}
{{ messages }}
+ {{ values }}
{{ delivery }}
{{ tools }}
{{ snapshot().error?.message }}
@@ -54,6 +55,9 @@ class Chat { get delivery() { return JSON.stringify(this.snapshot().messages.at(-1)?.delivery); } + get values() { + return JSON.stringify(this.snapshot().values) ?? 'unobserved'; + } get tools() { return JSON.stringify(this.snapshot().toolCalls); } @@ -93,29 +97,56 @@ describe('observeAgent borrowed session', () => { view.detectChanges(); expect(f.session.load).toBeTypeOf('function'); expect(f.history.reads).toBe(0); + expect( + view.nativeElement.querySelector('[data-testid="values"]').textContent + ).toBe('unobserved'); let notifications = 0; - const release = f.session.subscribe(() => { notifications++; }); + const release = f.session.subscribe(() => { + notifications++; + }); await f.session.load(); view.detectChanges(); - expect(view.nativeElement.querySelector('[data-testid="messages"]').textContent).toBe('Saved question\nSaved answer'); + expect( + view.nativeElement.querySelector('[data-testid="messages"]').textContent + ).toBe('Saved question\nSaved answer'); + expect( + JSON.parse( + view.nativeElement.querySelector('[data-testid="values"]').textContent + ) + ).toEqual({ counter: 1, stable: { items: ['saved'] } }); const snapshot = view.componentInstance.snapshot(); await f.session.load(); expect(view.componentInstance.snapshot()).toBe(snapshot); expect(notifications).toBe(1); expect(f.history.reads).toBe(2); + const saved = f.history.value[0]; + f.history.value = [{ ...saved, values: { ...saved.values, counter: 2 } }]; + await f.session.load(); + view.detectChanges(); + const refreshed = view.componentInstance.snapshot(); + expect(refreshed.values?.['stable']).toBe(snapshot.values?.['stable']); + expect( + JSON.parse( + view.nativeElement.querySelector('[data-testid="values"]').textContent + ) + ).toEqual({ counter: 2, stable: { items: ['saved'] } }); + expect(notifications).toBe(2); release(); view.destroy(); const reattached = observe(f.session); - expect(reattached.snapshot()).toBe(snapshot); - expect(f.history.reads).toBe(2); + expect(reattached.snapshot()).toBe(refreshed); + expect(f.history.reads).toBe(3); reattached.destroy(); f.history.value = []; await f.session.load(); expect(f.session.getSnapshot().messages).toEqual([]); - expect(f.history.reads).toBe(3); + expect(f.session.getSnapshot().values).toBeUndefined(); + expect(f.history.reads).toBe(4); expect(f.handlerCalls).toBe(0); expect(f.streams).toHaveLength(0); - expect(f.session.submitCalls + f.session.stopCalls + f.session.disposeCalls).toBe(0); + expect( + f.session.submitCalls + f.session.stopCalls + f.session.disposeCalls + ).toBe(0); }); it('renders streamed text, tool results, errors and stop outcomes through native controls', async () => { @@ -150,6 +181,7 @@ describe('observeAgent borrowed session', () => { expect(await view.componentInstance.run).toBe('success'); view.detectChanges(); expect(text('[data-testid="messages"]')).toContain('Visible final'); + expect(text('[data-testid="values"]')).toBe('{"stage":"complete"}'); expect(text('[data-testid="status"]')).toBe('idle'); expect(text('[data-testid="delivery"]')).toContain('success'); diff --git a/libs/angular/src/observe-agent.ts b/libs/angular/src/observe-agent.ts index e32d7cc0b..67cdf766e 100644 --- a/libs/angular/src/observe-agent.ts +++ b/libs/angular/src/observe-agent.ts @@ -1,15 +1,12 @@ import { DestroyRef, inject, signal, type Signal } from '@angular/core'; -import type { - AgentSession, - AgentSnapshot, - ToolContract, -} from '@threadplane/core'; +import type { AgentSnapshot } from '@threadplane/core'; /** Observe an app-owned session in an injection context. Destroying that * context releases only this subscription; the app retains session lifetime. */ -export function observeAgent< - TTools extends { [K in keyof TTools]: ToolContract } ->(session: AgentSession): Signal> { +export function observeAgent(session: { + getSnapshot(): TSnapshot; + subscribe(notify: () => void): () => void; +}): Signal { const destroyRef = inject(DestroyRef); const snapshot = signal(session.getSnapshot()); const release = session.subscribe(() => snapshot.set(session.getSnapshot())); diff --git a/libs/angular/src/observe-agent.type-test.ts b/libs/angular/src/observe-agent.type-test.ts index 37303e6d5..159a4e5c5 100644 --- a/libs/angular/src/observe-agent.type-test.ts +++ b/libs/angular/src/observe-agent.type-test.ts @@ -7,6 +7,43 @@ interface Tools { count: { args: { values: readonly string[] }; result: number }; } +interface ConcreteSnapshot extends AgentSnapshot { + readonly backend: 'concrete'; + readonly values: { + readonly counter: number; + readonly items: readonly string[]; + }; +} + +class ConcreteObserver { + constructor(readonly snapshot: ConcreteSnapshot) {} + getSnapshot() { + return this.snapshot; + } + subscribe(notify: () => void) { + void notify; + return () => undefined; + } +} + +export function observeConcreteSession(session: ConcreteObserver) { + const observation = observeAgent(session); + const exact: Signal = observation; + const snapshot = observation(); + const count: number = snapshot.values.counter; + const backend: 'concrete' = snapshot.backend; + // @ts-expect-error The concrete values field remains readonly. + snapshot.values = { counter: 2, items: [] }; + // @ts-expect-error Concrete fields remain readonly. + snapshot.values.counter = 2; + // @ts-expect-error Nested concrete arrays remain readonly. + snapshot.values.items.push('mutable'); + // @ts-expect-error Exact concrete fields cannot widen to any. + const invalid: string = snapshot.values.counter; + void [count, backend, invalid]; + return exact; +} + export function observeTypedSession(session: AgentSession) { const signal = observeAgent(session); const exact: Signal> = signal; diff --git a/libs/langgraph/src/lib/transport/fetch-stream.transport.spec.ts b/libs/langgraph/src/lib/transport/fetch-stream.transport.spec.ts index 688a3ef59..50984d88a 100644 --- a/libs/langgraph/src/lib/transport/fetch-stream.transport.spec.ts +++ b/libs/langgraph/src/lib/transport/fetch-stream.transport.spec.ts @@ -589,6 +589,25 @@ describe('FetchStreamTransport', () => { ]); }); + it('keeps protocol routing authoritative over colliding application fields', async () => { + const root = { type: 'domain', namespace: ['application'], count: 1 }; + const child = { type: 'values', namespace: [], count: 2 }; + mocks.runsStream.mockReturnValue( + (async function* () { + yield { event: 'values', data: root }; + yield { event: 'values|child', data: child }; + })(), + ); + const transport = new FetchStreamTransport('http://example.test'); + const events = await collect( + transport.stream('a', 't', {}, new AbortController().signal), + ); + expect(events).toEqual([ + { ...root, type: 'values', namespace: undefined, data: root }, + { ...child, type: 'values|child', namespace: ['child'], data: child }, + ]); + }); + it('normalizes message tuple events without dropping metadata', async () => { const message = { id: 'ai-1', type: 'ai', content: 'pong' }; const metadata = { langgraph_node: 'model', run_id: 'run-1' }; diff --git a/libs/langgraph/src/lib/transport/fetch-stream.transport.ts b/libs/langgraph/src/lib/transport/fetch-stream.transport.ts index a69852e00..20640b492 100644 --- a/libs/langgraph/src/lib/transport/fetch-stream.transport.ts +++ b/libs/langgraph/src/lib/transport/fetch-stream.transport.ts @@ -302,7 +302,13 @@ function normalizeSdkEvent(type: StreamEvent['type'], data: unknown): StreamEven } if (isRecord(data)) { - return { type, ...(namespace ? { namespace } : {}), ...data, data }; + // Application fields remain in data; they cannot change protocol routing. + return { + ...data, + type, + ...(namespace || Object.hasOwn(data, 'namespace') ? { namespace } : {}), + data, + }; } return { type, ...(namespace ? { namespace } : {}), data }; diff --git a/libs/langgraph/src/runtime/create-session.ts b/libs/langgraph/src/runtime/create-session.ts index 72c0bd6c3..6f1829f09 100644 --- a/libs/langgraph/src/runtime/create-session.ts +++ b/libs/langgraph/src/runtime/create-session.ts @@ -17,6 +17,9 @@ import { FetchStreamTransport } from '../lib/transport/fetch-stream.transport'; import { initialMessageState, reduceMessages } from './message-reducer'; import { createPublication } from './publication'; import { projectHistory } from './history-projection'; +import type { LangGraphSnapshot, LangGraphValues } from './langgraph-snapshot'; +import { projectHistoryValues, projectValues } from './values-projection'; +import { ownMessage } from './ownership'; import { createSafeRequestError } from './operation-errors'; import { failureProjection, @@ -59,7 +62,8 @@ export type LangGraphSession< string, ToolContract > -> = AgentSession & { +> = Omit, 'getSnapshot'> & { + getSnapshot(): LangGraphSnapshot; load?(options?: { readonly signal?: AbortSignal }): Promise; }; @@ -133,8 +137,10 @@ export function createSession( status: 'idle', messages: [], toolCalls: [], + values: undefined, }); let state = initialMessageState(); + let values: LangGraphValues | undefined; let owner: Attempt | undefined; let recoveryAttempt: Attempt | undefined; let disposed = false; @@ -148,6 +154,7 @@ export function createSession( function publish(status: 'idle' | 'running' | 'error', error?: AgentError) { publication.publish({ status, + values, messages: state.messages, toolCalls: typedTools ? state.toolCalls.filter( @@ -329,15 +336,14 @@ export function createSession( closeLoad(reading); } - function reconcile( - attempt: Attempt, - history: ThreadState[] - ): CompleteOutcome | undefined { + function reconcile(attempt: Attempt, history: ThreadState[]) { + const previousState = state; + const previousValues = values; const latest = history[0]; if (!latest) return undefined; - const values = record(latest.values); - const messages = Array.isArray(values?.['messages']) - ? values['messages'] + const checkpointValues = record(latest.values); + const messages = Array.isArray(checkpointValues?.['messages']) + ? checkpointValues['messages'] : []; // Inert construction gives us no server baseline. Only our unique submitted // user ID can correlate this checkpoint to this request, including when the @@ -363,7 +369,7 @@ export function createSession( : turn; const paused = nextUser < 0 && - (hasPause(values) || + (hasPause(checkpointValues) || latest.tasks?.some((task) => (task.interrupts?.length ?? 0) > 0)); const committed = (nextUser >= 0 || latest.next.length === 0) && @@ -378,13 +384,17 @@ export function createSession( ); }); if (!paused && !committed) return undefined; - const projected = projectStream(state, attempt.projection, { + const projected = projectStream(previousState, attempt.projection, { type: 'values', - data: { ...values, messages: [messages[anchor], ...turn] }, + data: { ...checkpointValues, messages: [messages[anchor], ...turn] }, }); - attempt.projection = projected.projection; - state = finalizeProjection(projected.state, projected.projection); - return paused ? 'paused' : 'success'; + const projectedValues = projectHistoryValues(previousValues, history); + return { + state: finalizeProjection(projected.state, projected.projection), + projection: projected.projection, + values: projectedValues, + outcome: paused ? ('paused' as const) : ('success' as const), + }; } async function execute(attempt: Attempt): Promise { @@ -425,7 +435,12 @@ export function createSession( return; } const projected = projectStream(state, attempt.projection, event); + const projectedValues = projectValues(values, event); + // Both projections may invoke transport-owned getters. Commit neither + // candidate if projection failed or a getter changed the owner. + if (!owns(attempt)) return; state = projected.state; + values = projectedValues; attempt.projection = projected.projection; publish('running'); // publish drains observer commands before returning. Never dispatch or @@ -445,7 +460,14 @@ export function createSession( attempt.controller.signal ); if (!owns(attempt)) return; - outcome = reconcile(attempt, history) ?? outcome; + const recovered = reconcile(attempt, history); + if (!owns(attempt)) return; + if (recovered) { + state = recovered.state; + values = recovered.values; + attempt.projection = recovered.projection; + outcome = recovered.outcome; + } } catch { if (!owns(attempt)) return; } @@ -599,10 +621,12 @@ export function createSession( ? { registeredTools: new Set(definitions.keys()) } : undefined ); + const projectedValues = projectHistoryValues(values, history); // Even a plain projection can invoke getters supplied by a transport. // Such a getter can submit/stop/dispose; never commit its stale result. if (!ownsLoad(read)) return; state = projected; + values = projectedValues; authoredTools.clear(); loading = undefined; read.resolve(); @@ -695,26 +719,32 @@ export function createSession( ); await publication.command(() => { if (disposed || captured.revision !== revision || !history) return; - const outcome = reconcile(captured.attempt, history); - if (!outcome) return; - recoveryAttempt = undefined; - checkController = undefined; + const recovered = reconcile(captured.attempt, history); + if (!recovered) return; // Interrupted deliveries are already complete: recovered canonical // history carries the conclusive success stamp in this aggregate. - state = { - ...state, - messages: state.messages.map((message) => + const recoveredState = { + ...recovered.state, + messages: recovered.state.messages.map((message) => message.delivery.generation === captured.attempt.generation - ? { + ? ownMessage({ ...message, delivery: completeDelivery( captured.attempt.generation, - outcome + recovered.outcome ), - } + }) : message ), }; + // Projection and delivery ownership must finish before clearing this + // check: a raw getter can synchronously start a replacement operation. + if (disposed || captured.revision !== revision) return; + recoveryAttempt = undefined; + checkController = undefined; + state = recoveredState; + values = recovered.values; + captured.attempt.projection = recovered.projection; publish('idle'); }); } finally { diff --git a/libs/langgraph/src/runtime/langgraph-snapshot.ts b/libs/langgraph/src/runtime/langgraph-snapshot.ts new file mode 100644 index 000000000..7cd792813 --- /dev/null +++ b/libs/langgraph/src/runtime/langgraph-snapshot.ts @@ -0,0 +1,18 @@ +import type { + AgentSnapshot, + PlainValue, + ToolContract, +} from '@threadplane/core'; + +/** Observed backend application data, not a validated application schema. */ +export type LangGraphValues = Readonly>; + +/** Backend-private extension; the core snapshot remains backend-independent. */ +export type LangGraphSnapshot< + TTools extends { [K in keyof TTools]: ToolContract } = Record< + string, + ToolContract + > +> = AgentSnapshot & { + readonly values: LangGraphValues | undefined; +}; diff --git a/libs/langgraph/src/runtime/ownership.ts b/libs/langgraph/src/runtime/ownership.ts index 2e111998c..4da6b5d4f 100644 --- a/libs/langgraph/src/runtime/ownership.ts +++ b/libs/langgraph/src/runtime/ownership.ts @@ -8,6 +8,7 @@ import { type PlainValue, type ToolCall, } from '@threadplane/core'; +import type { LangGraphSnapshot, LangGraphValues } from './langgraph-snapshot'; // Only objects projected here are trusted. Object.isFrozen on external input is // insufficient: its children may still be mutable. The weak set retains no data. @@ -58,6 +59,46 @@ export function ownValue( return result; } +/** Full-state ingress can reuse equal owned branches while replacing changed + * data. Both inputs pass the same ownership boundary before any reuse. */ +export function ownValueWithSharing( + value: PlainValue, + previous?: PlainValue +): PlainValue { + return shareOwnedValue(ownValue(value), ownValue(previous)); +} + +function shareOwnedValue(next: PlainValue, previous: PlainValue): PlainValue { + if (Object.is(next, previous)) return previous; + if ( + next === null || + previous === null || + typeof next !== 'object' || + typeof previous !== 'object' || + Array.isArray(next) !== Array.isArray(previous) + ) + return next; + const prior = previous as Record; + let equal = + Object.keys(next).length === Object.keys(previous).length && + (!Array.isArray(next) || + next.length === (previous as readonly PlainValue[]).length); + let shared = false; + const child = (value: PlainValue, key: string) => { + const exists = Object.hasOwn(previous, key); + const projected = exists ? shareOwnedValue(value, prior[key]) : value; + if (!exists || !Object.is(projected, prior[key])) equal = false; + if (!Object.is(projected, value)) shared = true; + return projected; + }; + const result = Array.isArray(next) + ? next.map((value, index) => child(value, String(index))) + : Object.fromEntries( + Object.entries(next).map(([key, value]) => [key, child(value, key)]) + ); + return equal ? previous : shared ? freeze(result) : next; +} + function equalValue(a: PlainValue, b: PlainValue): boolean { if (Object.is(a, b)) return true; if ( @@ -231,3 +272,16 @@ export function ownSnapshot( return previous; return freeze({ status: input.status, messages, toolCalls, error }); } + +/** One backend aggregate, composing the core fields and owned application data. */ +export function ownLangGraphSnapshot( + input: LangGraphSnapshot, + previous?: LangGraphSnapshot +): LangGraphSnapshot { + const core = ownSnapshot(input, previous); + const values = ownValueWithSharing(input.values, previous?.values) as + | LangGraphValues + | undefined; + if (core === previous && values === previous?.values) return previous; + return freeze({ ...core, values }); +} diff --git a/libs/langgraph/src/runtime/publication.spec.ts b/libs/langgraph/src/runtime/publication.spec.ts index f0483eda8..27a0cd7b8 100644 --- a/libs/langgraph/src/runtime/publication.spec.ts +++ b/libs/langgraph/src/runtime/publication.spec.ts @@ -1,11 +1,13 @@ -import { streamingDelivery, type AgentSnapshot } from '@threadplane/core'; +import { streamingDelivery } from '@threadplane/core'; import { describe, expect, it } from 'vitest'; import { createPublication } from './publication'; import { deferred } from './testing/deferred'; +import type { LangGraphSnapshot } from './langgraph-snapshot'; -function snapshot(content = 'one'): AgentSnapshot { +function snapshot(content = 'one'): LangGraphSnapshot { return { status: 'idle', + values: undefined, messages: [ { id: 'm', @@ -28,6 +30,46 @@ function snapshot(content = 'one'): AgentSnapshot { } describe('private snapshot publication', () => { + it('leaves the aggregate unchanged if application values ownership throws', () => { + const publication = createPublication(snapshot()); + const before = publication.getSnapshot(); + expect(() => + publication.publish({ + ...snapshot('candidate'), + values: { bad: new Date(0) }, + } as unknown as LangGraphSnapshot) + ).toThrow(TypeError); + expect(publication.getSnapshot()).toBe(before); + }); + + it('owns application values and suppresses equal queued aggregate publications', () => { + const data = { stable: { x: 1 }, count: 1 }; + const publication = createPublication({ ...snapshot(), values: data }); + const first = publication.getSnapshot() as LangGraphSnapshot; + expect(first.values).toEqual(data); + expect(first.values).not.toBe(data); + data.stable.x = 9; + expect(first.values?.['stable']).toEqual({ x: 1 }); + const seen: LangGraphSnapshot[] = []; + for (let index = 0; index < 2; index++) + publication.subscribe(() => { + if (publication.getSnapshot().status === 'running') { + const values = { stable: { x: 1 }, count: 2 }; + publication.publish({ ...snapshot('two'), values }); + values.count = 99; + } + }); + publication.subscribe(() => + seen.push(publication.getSnapshot() as LangGraphSnapshot) + ); + publication.publish({ ...first, status: 'running' }); + expect(seen).toHaveLength(2); + expect(seen[1].values).toEqual({ stable: { x: 1 }, count: 2 }); + expect(seen[1].values?.['stable']).toBe(first.values?.['stable']); + expect(seen[1].messages[0].content).toBe('two'); + expect(Object.isFrozen(seen[1].values)).toBe(true); + }); + it('does not turn unsupported SDK instances into apparently portable tool data', () => { const input = snapshot(); const invalid = { @@ -35,7 +77,7 @@ describe('private snapshot publication', () => { toolCalls: [{ ...input.toolCalls[0], args: new Date() }], }; expect(() => - createPublication(invalid as unknown as AgentSnapshot) + createPublication(invalid as unknown as LangGraphSnapshot) ).toThrow(TypeError); const publication = createPublication({ ...input, @@ -47,7 +89,7 @@ describe('private snapshot publication', () => { toolCalls: [ { id: 't', name: 'search', status: 'running', args: new Date() }, ], - } as unknown as AgentSnapshot) + } as unknown as LangGraphSnapshot) ).toThrow(TypeError); }); it('caches repeated reads per session and observes only actual changes', () => { diff --git a/libs/langgraph/src/runtime/publication.ts b/libs/langgraph/src/runtime/publication.ts index 8e1fe8349..ca70b1c2a 100644 --- a/libs/langgraph/src/runtime/publication.ts +++ b/libs/langgraph/src/runtime/publication.ts @@ -1,14 +1,14 @@ -import type { AgentSnapshot } from '@threadplane/core'; -import { ownSnapshot } from './ownership'; +import type { LangGraphSnapshot } from './langgraph-snapshot'; +import { ownLangGraphSnapshot } from './ownership'; /** Backend-private publication. Listener failures are reported once to the * optional callback and otherwise ignored; reporter failures are contained too. * Neither kind of failure changes execution state or rejects a command. */ export function createPublication( - initial: AgentSnapshot, + initial: LangGraphSnapshot, reportListenerError: (error: unknown) => void = () => undefined ) { - let current = ownSnapshot(initial); + let current = ownLangGraphSnapshot(initial); const listeners = new Set<{ notify: () => void }>(); const pending: (() => void)[] = []; let flushing = false; @@ -32,11 +32,11 @@ export function createPublication( } } - function publish(input: AgentSnapshot): void { + function publish(input: LangGraphSnapshot): void { // Capture external ingress now, including when a listener queues a publish. - const captured = ownSnapshot(input, current); + const captured = ownLangGraphSnapshot(input, current); schedule(() => { - const next = ownSnapshot(captured, current); + const next = ownLangGraphSnapshot(captured, current); if (next === current) return; current = next; notifying = true; diff --git a/libs/langgraph/src/runtime/stream-projection.ts b/libs/langgraph/src/runtime/stream-projection.ts index 5de14d72b..78f0ad475 100644 --- a/libs/langgraph/src/runtime/stream-projection.ts +++ b/libs/langgraph/src/runtime/stream-projection.ts @@ -53,7 +53,7 @@ export function projectStream( projection = { ...projection, paused: true }; } if (!terminal && !messageEvent) return { state, projection }; - const mode = event.messageMetadata ? 'delta' : 'snapshot'; + const mode = messageEvent && event.messageMetadata ? 'delta' : 'snapshot'; const incoming = Array.isArray(messages) ? messages.map(record).filter((m): m is Record => !!m) : []; diff --git a/libs/langgraph/src/runtime/testing/binding-fixture.ts b/libs/langgraph/src/runtime/testing/binding-fixture.ts index 97f4f9226..4c58a44ca 100644 --- a/libs/langgraph/src/runtime/testing/binding-fixture.ts +++ b/libs/langgraph/src/runtime/testing/binding-fixture.ts @@ -21,16 +21,29 @@ export function bindingFixture() { let handlerSignal: AbortSignal | undefined; const history: { reads: number; value: ThreadState[] } = { reads: 0, - value: [{ - values: { messages: [ - { id: 'saved-user', type: 'human', content: 'Saved question' }, - { id: 'saved-answer', type: 'ai', content: 'Saved answer' }, - ] }, - next: [], tasks: [], metadata: {}, - checkpoint: { thread_id: 'binding-thread', checkpoint_ns: '', checkpoint_id: 'saved', checkpoint_map: {} }, - parent_checkpoint: null, - created_at: '2026-09-21T00:00:00Z', - }], + value: [ + { + values: { + counter: 1, + stable: { items: ['saved'] }, + messages: [ + { id: 'saved-user', type: 'human', content: 'Saved question' }, + { id: 'saved-answer', type: 'ai', content: 'Saved answer' }, + ], + }, + next: [], + tasks: [], + metadata: {}, + checkpoint: { + thread_id: 'binding-thread', + checkpoint_ns: '', + checkpoint_id: 'saved', + checkpoint_map: {}, + }, + parent_checkpoint: null, + created_at: '2026-09-21T00:00:00Z', + }, + ], }; const stream: AgentTransport['stream'] = (_a, _t, _p, signal) => { const controlled = controlledTransport({ signal }); @@ -41,7 +54,13 @@ export function bindingFixture() { const runtime = createSession({ assistantId: 'binding-agent', threadId: 'binding-thread', - transport: { stream, getHistory: async () => { history.reads++; return history.value; } }, + transport: { + stream, + getHistory: async () => { + history.reads++; + return history.value; + }, + }, tools: { weather: { description: 'Weather', @@ -146,7 +165,10 @@ export const delta = (content: string, id = 'answer'): StreamEvent => ({ export const finalText = (content: string): StreamEvent => ({ type: 'values', - data: { messages: [{ type: 'ai', id: 'answer', content }] }, + data: { + stage: 'complete', + messages: [{ type: 'ai', id: 'answer', content }], + }, }); export const weatherCall: StreamEvent = { diff --git a/libs/langgraph/src/runtime/transport.integration.spec.ts b/libs/langgraph/src/runtime/transport.integration.spec.ts index 0cc9fe49e..d477ad73f 100644 --- a/libs/langgraph/src/runtime/transport.integration.spec.ts +++ b/libs/langgraph/src/runtime/transport.integration.spec.ts @@ -70,6 +70,85 @@ describe('neutral real SDK transport', () => { vi.unstubAllGlobals(); }); + it('keeps SDK routing authoritative while exposing colliding application keys', async () => { + const root = { + type: 'domain', + namespace: ['application'], + stage: 'root', + messages: [{ type: 'ai', id: 'answer', content: 'Root' }], + }; + const child = { + type: 'values', + namespace: [], + stage: 'child', + messages: [{ type: 'ai', id: 'child', content: 'Child' }], + }; + const request = vi.fn(async (url) => + String(url).endsWith('/history') + ? new Response('[]') + : fragmentedResponse( + `event: values\ndata: ${JSON.stringify( + root + )}\n\nevent: values|child\ndata: ${JSON.stringify(child)}\n\n` + ) + ); + vi.stubGlobal('fetch', request); + const session = createSession({ + assistantId: 'a', + threadId: 't', + apiUrl: 'https://runtime.example', + }); + try { + expect(await session.submit('Go')).toBe('success'); + expect(session.getSnapshot()).toMatchObject({ + values: { type: 'domain', namespace: ['application'], stage: 'root' }, + }); + expect(session.getSnapshot().messages.at(-1)?.content).toBe('Root'); + expect(request).toHaveBeenCalledTimes(1); + } finally { + await session.dispose(); + } + }); + + it('does not treat application messageMetadata as message-chunk routing', async () => { + const request = vi.fn(async () => + fragmentedResponse( + ['First', 'Second'] + .map( + (content) => + `event: values\ndata: ${JSON.stringify({ + messageMetadata: { domain: true }, + count: content, + messages: [{ type: 'ai', id: 'answer', content }], + })}\n\n` + ) + .join('') + ) + ); + vi.stubGlobal('fetch', request); + const session = createSession({ + assistantId: 'a', + threadId: 't', + apiUrl: 'https://runtime.example', + }); + const text: string[] = []; + session.subscribe(() => { + const last = session.getSnapshot().messages.at(-1); + if (last?.role === 'assistant') text.push(last.content); + }); + try { + expect(await session.submit('Go')).toBe('success'); + expect(text).toContain('Second'); + expect(text).not.toContain('FirstSecond'); + expect(session.getSnapshot()).toMatchObject({ + values: { messageMetadata: { domain: true }, count: 'Second' }, + }); + expect(request).toHaveBeenCalledTimes(1); + } finally { + await session.dispose(); + } + }); + it('loads decoded history through the real SDK endpoint without issuing runs, writes or tools', async () => { const request = vi.fn( async () => diff --git a/libs/langgraph/src/runtime/values-projection.spec.ts b/libs/langgraph/src/runtime/values-projection.spec.ts new file mode 100644 index 000000000..4807f24da --- /dev/null +++ b/libs/langgraph/src/runtime/values-projection.spec.ts @@ -0,0 +1,278 @@ +import type { ThreadState } from '@langchain/langgraph-sdk'; +import { describe, expect, it } from 'vitest'; +import { projectHistoryValues, projectValues } from './values-projection'; +import type { StreamEvent } from './transport.types'; + +const values = (data?: unknown): StreamEvent => ({ type: 'values', data }); +const checkpoint = (data?: unknown): StreamEvent => ({ + type: 'checkpoints', + data, +}); +const history = (value: unknown) => [{ values: value }] as ThreadState[]; + +describe('application values projection', () => { + it('distinguishes unobserved values from an observed empty root', () => { + expect(projectValues(undefined, values())).toBeUndefined(); + const empty = projectValues(undefined, values({})); + expect(empty).toEqual({}); + expect(Object.isFrozen(empty)).toBe(true); + expect(projectValues(empty, values({ messages: [] }))).toBe(empty); + }); + + it('replaces root state, correcting values and deleting omitted keys', () => { + const previous = projectValues( + undefined, + values({ text: 'Long answer', removed: 1, list: [1, 2] }) + ); + const next = projectValues( + previous, + values({ + text: '', + list: [], + zero: 0, + no: false, + nil: null, + absent: undefined, + }) + ); + expect(next).toEqual({ + text: '', + list: [], + zero: 0, + no: false, + nil: null, + absent: undefined, + }); + expect(Object.hasOwn(next ?? {}, 'removed')).toBe(false); + expect(Object.hasOwn(next ?? {}, 'absent')).toBe(true); + expect(previous).toEqual({ text: 'Long answer', removed: 1, list: [1, 2] }); + }); + + it('excludes only reserved fields in authoritative checkpoints and history', () => { + const input = { + messages: ['wire'], + __interrupt__: [], + tools: [], + __other__: true, + nested: { messages: 'keep', __interrupt__: 'keep' }, + }; + const expected = { + tools: [], + __other__: true, + nested: { messages: 'keep', __interrupt__: 'keep' }, + }; + const next = projectValues(undefined, checkpoint({ values: input })); + expect(next).toEqual(expected); + expect(projectHistoryValues(undefined, history(input))).toEqual(expected); + expect(input.messages).toEqual(['wire']); + }); + + for (const interrupt of [[], [{ value: 'pause' }], undefined]) { + it(`ignores live interrupt control envelopes (${JSON.stringify( + interrupt + )})`, () => { + const previous = projectValues(undefined, values({ saved: 'state' })); + expect( + projectValues(previous, values({ __interrupt__: interrupt })) + ).toBe(previous); + expect( + projectValues( + previous, + values({ __interrupt__: interrupt, saved: 'ignore control payload' }) + ) + ).toBe(previous); + expect( + projectValues(undefined, values({ __interrupt__: interrupt })) + ).toBeUndefined(); + }); + } + + it('requires explicit checkpoint data.values and replaces from that record', () => { + const previous = projectValues(undefined, values({ keep: 1 })); + expect(projectValues(previous, checkpoint({ keep: 2 }))).toBe(previous); + expect(projectValues(previous, checkpoint({ values: {} }))).toEqual({}); + expect( + projectValues(previous, checkpoint({ values: { next: 2 }, ignored: 1 })) + ).toEqual({ next: 2 }); + }); + + for (const data of [undefined, null, false, 0, '', []]) { + it(`ignores non-record live state but clears non-record history (${JSON.stringify( + data + )})`, () => { + const previous = projectValues(undefined, values({ keep: true })); + expect(projectValues(previous, values(data))).toBe(previous); + expect(projectValues(previous, checkpoint({ values: data }))).toBe( + previous + ); + expect(projectHistoryValues(previous, history(data))).toBeUndefined(); + }); + } + + it('uses only the latest history checkpoint and treats missing history as authoritative', () => { + const previous = projectValues(undefined, values({ stale: true })); + const next = projectHistoryValues(previous, [ + ...history({ fresh: 2 }), + ...history({ old: 1 }), + ]); + expect(next).toEqual({ fresh: 2 }); + expect(projectHistoryValues(next, [])).toBeUndefined(); + expect(projectHistoryValues(next, [{}] as ThreadState[])).toBeUndefined(); + expect(projectHistoryValues(next, history({}))).toEqual({}); + }); + + for (const event of [ + { type: 'values', namespace: ['child'], data: { wrong: true } }, + { + type: 'checkpoints', + namespace: ['child'], + data: { values: { wrong: true } }, + }, + { type: 'values|child', data: { wrong: true } }, + { type: 'checkpoints|child', data: { values: { wrong: true } } }, + { type: 'updates', data: { node: { wrong: true } } }, + { type: 'custom', data: { wrong: true } }, + { type: 'messages', data: { wrong: true } }, + { type: 'messages/complete', data: { wrong: true } }, + ] satisfies StreamEvent[]) { + it(`ignores ${event.type} namespace=${ + 'namespace' in event ? event.namespace : '' + }`, () => { + const previous = projectValues(undefined, values({ keep: 1 })); + expect(projectValues(previous, event)).toBe(previous); + expect(projectValues(undefined, event)).toBeUndefined(); + }); + } + + it('retains equal root identity across key order and history reads', () => { + const previous = projectValues(undefined, { + ...values({ a: { b: [1, { c: 2 }] }, count: 3 }), + namespace: [], + }); + expect( + projectValues(previous, values({ count: 3, a: { b: [1, { c: 2 }] } })) + ).toBe(previous); + expect( + projectHistoryValues( + previous, + history({ count: 3, messages: [], a: { b: [1, { c: 2 }] } }) + ) + ).toBe(previous); + }); + + it('shares unchanged nested branches within changed records and arrays', () => { + const previous = projectValues( + undefined, + values({ + stable: { x: 1 }, + changed: { child: { same: true }, value: 1 }, + items: [{ same: 1 }, { change: 1 }], + }) + ); + const next = projectValues( + previous, + values({ + stable: { x: 1 }, + changed: { child: { same: true }, value: 2 }, + items: [{ same: 1 }, { change: 2 }], + }) + ); + expect(next).not.toBe(previous); + expect(next?.['stable']).toBe(previous?.['stable']); + const oldChanged = previous?.['changed'] as Record; + const newChanged = next?.['changed'] as Record; + expect(newChanged).not.toBe(oldChanged); + expect(newChanged['child']).toBe(oldChanged['child']); + expect((next?.['items'] as unknown[])[0]).toBe( + (previous?.['items'] as unknown[])[0] + ); + expect((next?.['items'] as unknown[])[1]).not.toBe( + (previous?.['items'] as unknown[])[1] + ); + }); + + it('owns nested plain data without freezing or retaining the caller objects', () => { + const source = { nested: { entries: [{ count: 1 }] } }; + const next = projectValues(undefined, values(source)); + expect(next).not.toBe(source); + expect(next?.['nested']).not.toBe(source.nested); + expect(Object.isFrozen(source.nested)).toBe(false); + source.nested.entries[0].count = 2; + source.nested.entries.push({ count: 3 }); + expect(next).toEqual({ nested: { entries: [{ count: 1 }] } }); + const nested = next?.['nested'] as { entries: { count: number }[] }; + expect(Object.isFrozen(nested)).toBe(true); + expect(Object.isFrozen(nested.entries)).toBe(true); + expect(Object.isFrozen(nested.entries[0])).toBe(true); + }); + + it('does not trust shallow-frozen external roots or previous values', () => { + const nested = { count: 1 }; + const source = Object.freeze({ nested }); + const next = projectValues(source, values(source)); + expect(next).not.toBe(source); + expect(next?.['nested']).not.toBe(nested); + nested.count = 9; + expect(next).toEqual({ nested: { count: 1 } }); + }); + + it('preserves null-prototype records and own prototype-named data safely', () => { + const source = Object.assign(Object.create(null), { constructor: 'data' }); + source.__proto__ = { safe: true }; + const next = projectValues(undefined, values(source)); + expect(Object.getPrototypeOf(next)).toBe(Object.prototype); + expect(Object.hasOwn(next ?? {}, '__proto__')).toBe(true); + expect(next?.['__proto__']).toEqual({ safe: true }); + expect(next?.['constructor']).toBe('data'); + }); + + it('does not traverse excluded reserved contents or live interrupt getters', () => { + const source = { + keep: true, + get messages() { + throw new Error('reserved'); + }, + get __interrupt__() { + throw new Error('reserved'); + }, + }; + expect(projectValues(undefined, values(source))).toBeUndefined(); + expect(projectValues(undefined, checkpoint({ values: source }))).toEqual({ + keep: true, + }); + expect(projectHistoryValues(undefined, history(source))).toEqual({ + keep: true, + }); + }); + + it('rejects cyclic application values through the existing ownership boundary', () => { + const source: Record = {}; + source['self'] = source; + expect(() => projectValues(undefined, values(source))).toThrow(TypeError); + expect(() => + projectHistoryValues(undefined, history({ nested: source })) + ).toThrow(TypeError); + expect(source['self']).toBe(source); + expect(Object.isFrozen(source)).toBe(false); + }); + + for (const unsupported of [new Date(0), new Map(), () => undefined, 1n]) { + it(`rejects unsupported nested application data (${typeof unsupported})`, () => { + expect(() => projectValues(undefined, values({ unsupported }))).toThrow( + TypeError + ); + expect(() => + projectHistoryValues(undefined, history({ unsupported })) + ).toThrow(TypeError); + }); + } + + it('rejects unsupported object roots instead of fabricating empty state', () => { + expect(() => projectValues(undefined, values(new Date(0)))).toThrow( + TypeError + ); + expect(() => projectHistoryValues(undefined, history(new Map()))).toThrow( + TypeError + ); + }); +}); diff --git a/libs/langgraph/src/runtime/values-projection.ts b/libs/langgraph/src/runtime/values-projection.ts new file mode 100644 index 000000000..311efd3c9 --- /dev/null +++ b/libs/langgraph/src/runtime/values-projection.ts @@ -0,0 +1,52 @@ +import type { ThreadState } from '@langchain/langgraph-sdk'; +import type { PlainValue } from '@threadplane/core'; +import type { LangGraphValues } from './langgraph-snapshot'; +import { ownValue, ownValueWithSharing } from './ownership'; +import type { StreamEvent } from './transport.types'; +import { record } from './wire-message'; + +type Values = LangGraphValues | undefined; + +function projectRecord( + previous: Values, + value: Record +): LangGraphValues { + const prototype = Object.getPrototypeOf(value); + if (prototype !== Object.prototype && prototype !== null) { + // Reject SDK/class instances through the existing plain-data boundary, + // before selecting fields could disguise them as ordinary records. + ownValue(value as PlainValue); + } + const selected = Object.fromEntries( + Object.keys(value) + .filter((key) => key !== 'messages' && key !== '__interrupt__') + .map((key) => [key, value[key]]) + ); + return ownValueWithSharing( + selected as PlainValue, + previous + ) as LangGraphValues; +} + +/** Only root full-state records replace application values. Node updates are + * not root snapshots, and live interrupt envelopes carry control metadata only. */ +export function projectValues(previous: Values, event: StreamEvent): Values { + if ((event.namespace?.length ?? 0) > 0 || event.type.includes('|')) + return previous; + if (event.type !== 'values' && event.type !== 'checkpoints') return previous; + const data = record(event['data']); + if (event.type === 'values' && data && Object.hasOwn(data, '__interrupt__')) + return previous; + const value = event.type === 'checkpoints' ? record(data?.['values']) : data; + return value ? projectRecord(previous, value) : previous; +} + +/** A successful history replacement is authoritative, including absent values. + * Checkpoint interrupt metadata does not turn the checkpoint into an envelope. */ +export function projectHistoryValues( + previous: Values, + history: readonly ThreadState[] +): Values { + const value = record(history[0]?.values); + return value ? projectRecord(previous, value) : undefined; +} diff --git a/libs/langgraph/src/runtime/values.spec.ts b/libs/langgraph/src/runtime/values.spec.ts new file mode 100644 index 000000000..c7cec6031 --- /dev/null +++ b/libs/langgraph/src/runtime/values.spec.ts @@ -0,0 +1,619 @@ +import type { ThreadState } from '@langchain/langgraph-sdk'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { createSession } from './create-session'; +import type { LangGraphSnapshot } from './langgraph-snapshot'; +import { controlledTransport } from './testing/controlled-transport'; +import { deferred } from './testing/deferred'; +import type { AgentTransport, StreamEvent } from './transport.types'; + +const ai = (content: string, id = 'answer') => ({ id, type: 'ai', content }); +const full = (data: unknown): StreamEvent => ({ type: 'values', data }); +const checkpoint = (values: unknown, next: string[] = []): ThreadState => ({ + values: values as ThreadState['values'], + next, + tasks: [], + metadata: null, + checkpoint: { + thread_id: 't', + checkpoint_id: 'c', + checkpoint_ns: '', + checkpoint_map: {}, + }, + parent_checkpoint: null, + created_at: null, +}); +const fixtures: { + session: { dispose(): Promise }; + streams: ReturnType>[]; +}[] = []; +function fixture() { + const streams: ReturnType>[] = []; + const acknowledgements: ReturnType>[] = []; + const starts = Array.from({ length: 8 }, () => deferred()); + const stream = vi.fn((_a, _t, _p, signal) => { + const index = streams.length; + const controlled = controlledTransport({ signal }); + streams.push(controlled); + starts[index].resolve(); + const iterator: AsyncIterableIterator = { + [Symbol.asyncIterator]() { + return iterator; + }, + next() { + acknowledgements[index]?.resolve(); + return controlled.stream.next(); + }, + return() { + acknowledgements[index]?.resolve(); + return controlled.stream.return(); + }, + }; + return iterator; + }); + const history = vi.fn>( + async () => [] + ); + const write = vi.fn>( + async () => undefined + ); + const handler = vi.fn((args: { input: string }) => args.input); + const claim = vi.fn(async () => 'claimed' as const); + const record = vi.fn(async () => undefined); + const session = createSession({ + assistantId: 'a', + threadId: 't', + transport: { stream, getHistory: history, updateState: write }, + executionStore: { claim, record }, + tools: { work: { description: 'Work', handler } }, + }); + const f = { + session, + streams, + stream, + history, + write, + handler, + claim, + record, + snapshot: () => session.getSnapshot() as LangGraphSnapshot, + started: (index = 0) => starts[index].promise, + async emit(event: StreamEvent, index = 0) { + acknowledgements[index] = deferred(); + streams[index].release(event); + await acknowledgements[index].promise; + }, + turn(values: Record = {}, index = 0) { + const user = (stream.mock.calls[index][2] as { messages: unknown[] }) + .messages[0]; + const data = Object.create( + Object.getPrototypeOf(values), + Object.getOwnPropertyDescriptors(values) + ); + Object.defineProperty(data, 'messages', { + value: [user, ai('Recovered')], + enumerable: true, + configurable: true, + }); + return checkpoint(data); + }, + }; + fixtures.push(f); + return f; +} +async function seed(f: ReturnType) { + f.history.mockResolvedValueOnce([ + checkpoint({ + messages: [ai('Saved', 'saved')], + count: 1, + stable: { nested: ['owned'] }, + }), + ]); + await f.session.load?.(); +} +afterEach(async () => { + await Promise.all( + fixtures.splice(0).map(async (f) => { + await f.session.dispose(); + f.streams.forEach((stream) => stream.finish()); + }) + ); +}); + +describe('session application values', () => { + it('starts unobserved and publishes values-only changes in one coherent aggregate', async () => { + const f = fixture(); + expect(f.snapshot()).toHaveProperty('values', undefined); + const run = f.session.submit('Go'); + await f.started(); + const observed: LangGraphSnapshot[] = []; + f.session.subscribe(() => observed.push(f.snapshot())); + await f.emit( + full({ + messages: [ai('First')], + count: 1, + removed: true, + stable: { x: 1 }, + }) + ); + expect(observed).toHaveLength(1); + expect(observed[0].values).toEqual({ + count: 1, + removed: true, + stable: { x: 1 }, + }); + expect(observed[0].messages.at(-1)?.content).toBe('First'); + const first = f.snapshot(); + await f.emit(full({ count: 2, stable: { x: 1 } })); + expect(observed).toHaveLength(2); + expect(f.snapshot().messages).toBe(first.messages); + expect(f.snapshot().values).toEqual({ count: 2, stable: { x: 1 } }); + expect(f.snapshot().values?.['stable']).toBe(first.values?.['stable']); + const same = f.snapshot(); + await f.emit(full({ count: 2, stable: { x: 1 } })); + expect(f.snapshot()).toBe(same); + expect(observed).toHaveLength(2); + f.streams[0].finish(); + expect(await run).toBe('success'); + }); + + it('retains values through tokens, ignored envelopes, stop, failure and disposal', async () => { + const f = fixture(); + await seed(f); + const values = f.snapshot().values; + const run = f.session.submit('Go'); + await f.started(); + expect(f.snapshot().values).toBe(values); + for (const event of [ + { + type: 'messages', + messages: [{ type: 'AIMessageChunk', id: 'partial', content: 'Token' }], + messageMetadata: {}, + }, + full({ __interrupt__: [] }), + { type: 'updates', data: { node: { changed: true } } }, + { type: 'custom', data: { changed: true } }, + { type: 'values', namespace: ['child'], data: { changed: true } }, + { type: 'values|child', data: { changed: true } }, + ] satisfies StreamEvent[]) { + await f.emit(event); + expect(f.snapshot().values).toBe(values); + } + await f.session.stop(); + expect(await run).toBe('aborted'); + const failed = f.session.submit('Fail'); + await f.started(1); + await f.emit({ type: 'error', data: { message: 'Unavailable' } }, 1); + expect(await failed).toBe('error'); + expect(f.snapshot().values).toBe(values); + await f.session.dispose(); + expect(f.snapshot().values).toBe(values); + }); + + it('uses root checkpoints authoritatively even with interrupt metadata', async () => { + const f = fixture(); + await seed(f); + const run = f.session.submit('Go'); + await f.started(); + await f.emit({ + type: 'checkpoints', + data: { + values: { count: 2, __interrupt__: [], messages: [ai('Checkpoint')] }, + }, + }); + expect(f.snapshot().values).toEqual({ count: 2 }); + f.streams[0].finish(); + expect(await run).toBe('success'); + }); + + for (const [label, invalid] of [ + ['nested instance', () => ({ bad: new Date(0) })], + [ + 'root instance', + () => Object.assign(new Date(0), { messages: [ai('Invalid')] }), + ], + [ + 'cycle', + () => { + const value: Record = {}; + value['self'] = value; + return value; + }, + ], + [ + 'throwing getter', + () => ({ + get bad() { + throw new Error('PRIVATE'); + }, + }), + ], + ] as const) { + it(`rejects ${label} without committing candidate messages, tools or values`, async () => { + const f = fixture(); + await seed(f); + const run = f.session.submit('Go'); + await f.started(); + const prior = f.snapshot(); + const input = invalid(); + Object.defineProperty(input, 'messages', { + value: [ + { + ...ai('Invalid', 'rejected-message'), + tool_calls: [ + { id: 'bad-call', name: 'work', args: { input: 'bad' } }, + ], + }, + ], + enumerable: true, + configurable: true, + }); + await f.emit(full(input)); + expect(f.snapshot().messages.map((message) => message.content)).toEqual( + prior.messages.map((message) => message.content) + ); + expect(f.snapshot().values).toBe(prior.values); + expect(f.snapshot().toolCalls).toBe(prior.toolCalls); + expect(f.handler).not.toHaveBeenCalled(); + expect(f.claim).not.toHaveBeenCalled(); + expect(f.record).not.toHaveBeenCalled(); + expect(f.write).not.toHaveBeenCalled(); + expect(f.stream).toHaveBeenCalledTimes(1); + await run; + // Reconciliation must not resurrect the rejected event's canonical + // message or tool eligibility from a partially committed projection. + f.history.mockImplementationOnce(async () => [ + f.turn({ recovered: true }), + ]); + await f.session.checkStatus?.(); + expect( + f + .snapshot() + .messages.some((message) => message.id === 'rejected-message') + ).toBe(false); + expect(f.snapshot().toolCalls).toEqual([]); + expect(f.handler).not.toHaveBeenCalled(); + }); + } + + it('does not commit values when message ownership fails first', async () => { + const f = fixture(); + await seed(f); + const prior = f.snapshot(); + const run = f.session.submit('Go'); + await f.started(); + await f.emit( + full({ + count: 9, + messages: [ + { + ...ai('Bad'), + tool_calls: [{ id: 'bad', name: 'work', args: new Date(0) }], + }, + ], + }) + ); + await run; + expect(f.snapshot().values).toBe(prior.values); + expect( + f.snapshot().messages.some((message) => message.content === 'Bad') + ).toBe(false); + }); + + for (const action of ['stop', 'submit', 'dispose'] as const) { + it(`discards live candidates when a values getter invokes ${action}`, async () => { + const f = fixture(); + await seed(f); + const prior = f.snapshot().values; + const run = f.session.submit('Old'); + await f.started(); + let fired = false; + let nested: Promise | undefined; + const data = { + messages: [ai('Stale')], + get count() { + if (!fired) { + fired = true; + nested = + action === 'submit' + ? f.session.submit('New') + : f.session[action](); + } + return 9; + }, + }; + await f.emit(full(data)); + expect(fired).toBe(true); + await run; + expect(f.snapshot().values).toBe(prior); + expect( + f.snapshot().messages.some((message) => message.content === 'Stale') + ).toBe(false); + if (action === 'submit') { + await f.started(1); + await f.session.stop(); + } + await nested; + expect(f.stream).toHaveBeenCalledTimes(action === 'submit' ? 2 : 1); + }); + } + + it('loads coherent state, preserves equal identity, retains failed reads and clears empty history', async () => { + const f = fixture(); + await seed(f); + const before = f.snapshot(); + const changed = vi.fn(); + f.session.subscribe(changed); + await seed(f); + expect(f.snapshot()).toBe(before); + expect(changed).not.toHaveBeenCalled(); + f.history.mockRejectedValueOnce(new Error('PRIVATE')); + await expect(f.session.load?.()).rejects.toThrow(); + expect(f.snapshot()).toBe(before); + f.history.mockResolvedValueOnce([ + checkpoint({ count: new Map(), messages: [ai('Invalid load')] }), + ]); + await expect(f.session.load?.()).rejects.toThrow(); + expect(f.snapshot()).toBe(before); + const pending = deferred(); + f.history.mockReturnValueOnce(pending.promise); + const stale = f.session.load?.(); + await Promise.resolve(); + await f.session.stop(); + await stale; + pending.resolve([checkpoint({ count: 99, messages: [ai('Stale load')] })]); + await Promise.resolve(); + await Promise.resolve(); + expect(f.snapshot()).toBe(before); + await f.session.load?.(); + expect(f.snapshot().values).toBeUndefined(); + expect(f.snapshot().messages).toEqual([]); + expect(changed).toHaveBeenCalledTimes(1); + }); + + it('guards load candidates after a values getter submits a new request', async () => { + const f = fixture(); + await seed(f); + const prior = f.snapshot().values; + let run: Promise | undefined; + f.history.mockResolvedValueOnce([ + checkpoint({ + messages: [ai('Stale load')], + get count() { + run ??= f.session.submit('New'); + return 9; + }, + }), + ]); + await f.session.load?.(); + expect(run).toBeDefined(); + await f.started(); + expect(f.snapshot().values).toBe(prior); + expect( + f.snapshot().messages.some((message) => message.content === 'Stale load') + ).toBe(false); + await f.session.stop(); + await run; + }); + + for (const mode of ['close', 'check'] as const) { + it(`recovers correlated ${mode} history atomically with checkpoint values`, async () => { + const f = fixture(); + await seed(f); + const run = f.session.submit('Go'); + await f.started(); + if (mode === 'check') { + f.streams[0].finish(); + expect(await run).toBe('interrupted'); + } + f.history.mockImplementationOnce(async () => [ + f.turn({ count: 3, __interrupt__: [] }), + ]); + const seen: LangGraphSnapshot[] = []; + f.session.subscribe(() => seen.push(f.snapshot())); + if (mode === 'close') { + f.streams[0].finish(); + expect(await run).toBe('success'); + } else await f.session.checkStatus?.(); + expect(f.snapshot().values).toEqual({ count: 3 }); + expect(f.snapshot().messages.at(-1)?.content).toBe('Recovered'); + expect( + seen.every( + (snapshot) => + snapshot.messages.at(-1)?.content !== 'Recovered' || + snapshot.values?.['count'] === 3 + ) + ).toBe(true); + }); + } + + it('does not change values from unrelated or inconclusive recovery checkpoints', async () => { + const f = fixture(); + await seed(f); + const prior = f.snapshot().values; + const run = f.session.submit('Go'); + await f.started(); + f.history.mockResolvedValueOnce([ + checkpoint({ count: 8, messages: [ai('Unrelated')] }), + ]); + f.streams[0].finish(); + expect(await run).toBe('interrupted'); + const interrupted = f.snapshot(); + const inconclusive = f.turn({ count: 8 }); + inconclusive.next = ['still-running']; + f.history.mockResolvedValueOnce([inconclusive]); + await f.session.checkStatus?.(); + expect(f.snapshot()).toBe(interrupted); + expect(f.snapshot().values).toBe(prior); + }); + + for (const action of ['stop', 'submit', 'dispose'] as const) { + it(`discards close-time recovery candidates after a getter invokes ${action}`, async () => { + const f = fixture(); + await seed(f); + const prior = f.snapshot().values; + const run = f.session.submit('Old'); + await f.started(); + let nested: Promise | undefined; + let fired = false; + f.history.mockImplementationOnce(async () => [ + f.turn({ + get count() { + if (!fired) { + fired = true; + nested = + action === 'submit' + ? f.session.submit('New') + : f.session[action](); + } + return 9; + }, + }), + ]); + f.streams[0].finish(); + expect(await run).toBe(action === 'submit' ? 'interrupted' : 'aborted'); + expect(fired).toBe(true); + expect(f.snapshot().values).toBe(prior); + expect( + f.snapshot().messages.some((message) => message.content === 'Recovered') + ).toBe(false); + if (action === 'submit') { + await f.started(1); + await f.session.stop(); + } + await nested; + }); + } + + for (const mode of ['close', 'check'] as const) { + it(`preserves prior candidates when ${mode} recovery values ownership fails`, async () => { + const f = fixture(); + await seed(f); + const run = f.session.submit('Old'); + await f.started(); + if (mode === 'check') { + f.streams[0].finish(); + expect(await run).toBe('interrupted'); + } + const before = f.snapshot(); + f.history.mockImplementationOnce(async () => [ + f.turn({ invalid: new Date(0) }), + ]); + if (mode === 'close') { + f.streams[0].finish(); + expect(await run).toBe('interrupted'); + } else await expect(f.session.checkStatus?.()).rejects.toThrow(TypeError); + expect(f.snapshot().values).toBe(before.values); + expect(f.snapshot().messages.map((message) => message.content)).toEqual( + before.messages.map((message) => message.content) + ); + if (mode === 'check') expect(f.snapshot()).toBe(before); + f.history.mockImplementationOnce(async () => [ + f.turn({ recovered: true }), + ]); + await f.session.checkStatus?.(); + expect(f.snapshot().values).toEqual({ recovered: true }); + }); + } + + it('does not acknowledge a staged handoff on invalid values or unrelated continuation recovery', async () => { + const f = fixture(); + await seed(f); + const run = f.session.submit('Work'); + await f.started(); + await f.emit( + full({ + messages: [ + { + ...ai('Tool'), + tool_calls: [ + { id: 'work-call', name: 'work', args: { input: 'result' } }, + ], + }, + ], + }) + ); + f.streams[0].finish(); + await f.started(1); + // The earlier tool-producing checkpoint lacks this continuation's exact + // ToolMessage handoff, so its values cannot recover the continuation. + f.history.mockImplementationOnce(async () => [ + f.turn({ wrong: 'earlier step' }), + ]); + f.streams[1].finish(); + expect(await run).toBe('interrupted'); + expect(f.snapshot().values).toEqual({}); + const handoff = (f.stream.mock.calls[1][2] as { messages: unknown[] }) + .messages[0]; + const invalid = f.session.submit('Retry'); + await f.started(2); + await f.emit(full({ messages: [ai('Invalid')], bad: new Date(0) }), 2); + await invalid; + const retry = f.session.submit('Again'); + await f.started(3); + expect( + (f.stream.mock.calls[2][2] as { messages: unknown[] }).messages[0] + ).toEqual(handoff); + expect( + (f.stream.mock.calls[3][2] as { messages: unknown[] }).messages[0] + ).toEqual(handoff); + expect(f.handler).toHaveBeenCalledTimes(1); + expect(f.write).not.toHaveBeenCalled(); + await f.session.stop(); + await retry; + }); + + for (const action of ['submit', 'checkStatus'] as const) { + it(`an older recovery getter cannot clear a replacement ${action} owner`, async () => { + const f = fixture(); + await seed(f); + const run = f.session.submit('Old'); + await f.started(); + f.streams[0].finish(); + expect(await run).toBe('interrupted'); + const prior = f.snapshot(); + let fired = false; + let nested: Promise | undefined; + const replacement = deferred(); + f.history.mockImplementationOnce(async () => [ + f.turn({ + get count() { + if (!fired) { + fired = true; + nested = + action === 'submit' + ? f.session.submit('New') + : f.session.checkStatus?.(); + } + return 9; + }, + }), + ]); + f.history.mockReturnValueOnce(replacement.promise); + await f.session.checkStatus?.(); + expect(f.snapshot().values).toBe(prior.values); + expect( + f.snapshot().messages.some((message) => message.content === 'Recovered') + ).toBe(false); + if (action === 'submit') { + await f.started(1); + await f.session.stop(); + } else { + replacement.resolve([f.turn({ count: 4 })]); + } + await nested; + if (action === 'checkStatus') + expect(f.snapshot().values).toEqual({ count: 4 }); + }); + } + + it('keeps independent values for two sessions', async () => { + const first = fixture(); + const second = fixture(); + await seed(first); + expect(second.snapshot()).toHaveProperty('values', undefined); + second.history.mockResolvedValueOnce([checkpoint({ other: 2 })]); + await second.session.load?.(); + expect(second.snapshot().values).toEqual({ other: 2 }); + expect(first.snapshot().values?.['count']).toBe(1); + }); +}); diff --git a/libs/langgraph/src/runtime/values.type-test.ts b/libs/langgraph/src/runtime/values.type-test.ts new file mode 100644 index 000000000..0d69e1413 --- /dev/null +++ b/libs/langgraph/src/runtime/values.type-test.ts @@ -0,0 +1,42 @@ +import type { AgentSession, PlainValue } from '@threadplane/core'; +import { createSession } from './create-session'; +import type { LangGraphSnapshot } from './langgraph-snapshot'; + +const session = createSession({ + assistantId: 'a', + threadId: 't', + tools: { + weather: { + description: 'Weather', + handler: (args: { city: string }) => ({ temperature: args.city.length }), + }, + }, +}); +const observer: AgentSession = session; +const snapshot = session.getSnapshot(); +const exact: LangGraphSnapshot<{ + weather: { args: { city: string }; result: { temperature: number } }; +}> = snapshot; +const values: Readonly> | undefined = + snapshot.values; +// @ts-expect-error The backend does not infer an application schema. +const assumed: number = snapshot.values?.['counter']; +// @ts-expect-error Values fields cannot be assigned. +snapshot.values = {}; +if (snapshot.values) { + // @ts-expect-error The broad application record is readonly. + snapshot.values['counter'] = 1; + const nested = snapshot.values['nested']; + if (nested && typeof nested === 'object' && !Array.isArray(nested)) { + // @ts-expect-error Plain nested data is readonly too. + nested['mutable'] = true; + } +} +for (const call of snapshot.toolCalls) + if (call.status === 'complete') { + const temperature: number = call.result.temperature; + // @ts-expect-error Authored result inference is retained by the replacement getter. + const bad: string = call.result.temperature; + void [temperature, bad]; + } +void [observer, exact, values, assumed]; diff --git a/libs/react/README.md b/libs/react/README.md index f85f533bc..e08cc2a16 100644 --- a/libs/react/README.md +++ b/libs/react/README.md @@ -1,9 +1,10 @@ # @threadplane/react -Private, unpublished React binding for app-owned `AgentSession` values from -`@threadplane/core`. The root exports `useAgent(session)`, which returns the -current `AgentSnapshot` through React's `useSyncExternalStore`. Tool names, -arguments, and results retain the session's declared types. +Private, unpublished React binding for app-owned sessions. The root exports +`useAgent(session)`, which accepts `getSnapshot()` and `subscribe(notify)` methods +and returns their concrete `TSnapshot` through React's `useSyncExternalStore`. +The snapshot must extend the core `AgentSnapshot`; its additional fields and tool +names, arguments, and results retain their inferred types. ```tsx import { useAgent } from '@threadplane/react'; @@ -21,6 +22,15 @@ it does not stop pending work or dispose the session. Multiple components can observe the same session. The app calls `session.submit(text)`, `session.stop()`, and `session.dispose()` and owns the session's lifetime. Replacing the session prop transfers the subscription without disposing the previous session. +Snapshot reads and subscription calls preserve the session method receiver. + +Keep the concrete session type when observing backend-specific fields. The private +LangGraph fixture exposes a broad readonly `values` map on its snapshot: `undefined` +means no current application-values map is observed, while `{}` is an observed +empty map. The hook preserves that field without inferring an application schema, +validating values, or issuing extra reads. Values and messages arrive in the same +immutable snapshot. This does not make the private backend factory public or add +state-writing support. The root retains `use client`. Server rendering and hydration are not supported by this binding. No backend constructor is exported here; the current real diff --git a/libs/react/src/use-agent.spec.tsx b/libs/react/src/use-agent.spec.tsx index b2b6cbe64..8503f1bf5 100644 --- a/libs/react/src/use-agent.spec.tsx +++ b/libs/react/src/use-agent.spec.tsx @@ -34,31 +34,68 @@ describe('useAgent borrowed session', () => { function History() { const snapshot = useAgent(f.session); renders++; - return {snapshot.messages.map((message) => message.content).join('\n')}; + return ( + <> + + {snapshot.messages.map((message) => message.content).join('\n')} + + + {JSON.stringify(snapshot.values) ?? 'unobserved'} + + + ); } const view = render(, { reactStrictMode: true }); expect(f.session.load).toBeTypeOf('function'); expect(f.history.reads).toBe(0); - await act(async () => { await f.session.load(); }); - expect(view.getByRole('status').textContent).toBe('Saved question\nSaved answer'); + expect(view.getByTestId('values').textContent).toBe('unobserved'); + await act(async () => { + await f.session.load(); + }); + expect(view.getByTestId('history-messages').textContent).toBe( + 'Saved question\nSaved answer' + ); + expect(JSON.parse(view.getByTestId('values').textContent ?? '')).toEqual({ + counter: 1, + stable: { items: ['saved'] }, + }); const snapshot = f.session.getSnapshot(); const beforeRefresh = renders; - await act(async () => { await f.session.load(); }); + await act(async () => { + await f.session.load(); + }); expect(f.session.getSnapshot()).toBe(snapshot); expect(renders).toBe(beforeRefresh); expect(f.history.reads).toBe(2); + const saved = f.history.value[0]; + f.history.value = [{ ...saved, values: { ...saved.values, counter: 2 } }]; + await act(async () => { + await f.session.load(); + }); + const refreshed = f.session.getSnapshot(); + expect(refreshed.values?.['stable']).toBe(snapshot.values?.['stable']); + expect(JSON.parse(view.getByTestId('values').textContent ?? '')).toEqual({ + counter: 2, + stable: { items: ['saved'] }, + }); view.unmount(); const reattached = render(, { reactStrictMode: true }); - expect(reattached.getByRole('status').textContent).toBe('Saved question\nSaved answer'); - expect(f.history.reads).toBe(2); + expect(reattached.getByTestId('history-messages').textContent).toBe( + 'Saved question\nSaved answer' + ); + expect(f.session.getSnapshot()).toBe(refreshed); + expect(f.history.reads).toBe(3); reattached.unmount(); f.history.value = []; await f.session.load(); expect(f.session.getSnapshot().messages).toEqual([]); - expect(f.history.reads).toBe(3); + expect(f.session.getSnapshot().values).toBeUndefined(); + expect(f.history.reads).toBe(4); expect(f.handlerCalls).toBe(0); expect(f.streams).toHaveLength(0); - expect(f.session.submitCalls + f.session.stopCalls + f.session.disposeCalls).toBe(0); + expect( + f.session.submitCalls + f.session.stopCalls + f.session.disposeCalls + ).toBe(0); }); it('renders streamed text, tool results, errors and stop outcomes through native controls', async () => { @@ -120,6 +157,7 @@ describe('useAgent borrowed session', () => { expect(await run).toBe('success'); }); expect(view.getByTestId('messages').textContent).toContain('Visible final'); + expect(f.session.getSnapshot().values).toEqual({ stage: 'complete' }); expect(view.getByTestId('status').textContent).toBe('idle'); expect(view.getByTestId('delivery').textContent).toContain('success'); diff --git a/libs/react/src/use-agent.ts b/libs/react/src/use-agent.ts index b2c909df4..f17f8caa4 100644 --- a/libs/react/src/use-agent.ts +++ b/libs/react/src/use-agent.ts @@ -1,16 +1,13 @@ 'use client'; import { useCallback, useSyncExternalStore } from 'react'; -import type { - AgentSession, - AgentSnapshot, - ToolContract, -} from '@threadplane/core'; +import type { AgentSnapshot } from '@threadplane/core'; /** Observe an app-owned session. Unmount releases only this subscription. */ -export function useAgent( - session: AgentSession -): AgentSnapshot { +export function useAgent(session: { + getSnapshot(): TSnapshot; + subscribe(notify: () => void): () => void; +}): TSnapshot { const getSnapshot = useCallback(() => session.getSnapshot(), [session]); const subscribe = useCallback( (notify: () => void) => session.subscribe(notify), diff --git a/libs/react/src/use-agent.type-test.ts b/libs/react/src/use-agent.type-test.ts index 829c00ab2..4d35b3156 100644 --- a/libs/react/src/use-agent.type-test.ts +++ b/libs/react/src/use-agent.type-test.ts @@ -6,6 +6,42 @@ interface Tools { count: { args: { values: readonly string[] }; result: number }; } +interface ConcreteSnapshot extends AgentSnapshot { + readonly backend: 'concrete'; + readonly values: { + readonly counter: number; + readonly items: readonly string[]; + }; +} + +class ConcreteObserver { + constructor(readonly snapshot: ConcreteSnapshot) {} + getSnapshot() { + return this.snapshot; + } + subscribe(notify: () => void) { + void notify; + return () => undefined; + } +} + +export function useConcreteSession(session: ConcreteObserver) { + const snapshot = useAgent(session); + const exact: ConcreteSnapshot = snapshot; + const count: number = snapshot.values.counter; + const backend: 'concrete' = snapshot.backend; + // @ts-expect-error The concrete values field remains readonly. + snapshot.values = { counter: 2, items: [] }; + // @ts-expect-error Concrete fields remain readonly. + snapshot.values.counter = 2; + // @ts-expect-error Nested concrete arrays remain readonly. + snapshot.values.items.push('mutable'); + // @ts-expect-error Exact concrete fields cannot widen to any. + const invalid: string = snapshot.values.counter; + void [count, backend, invalid]; + return exact; +} + export function useTypedSession(session: AgentSession) { const snapshot = useAgent(session); const exact: AgentSnapshot = snapshot; diff --git a/scripts/react-parity/baseline.json b/scripts/react-parity/baseline.json index 29d6b5653..cca17dc56 100644 --- a/scripts/react-parity/baseline.json +++ b/scripts/react-parity/baseline.json @@ -1,15 +1,18 @@ { "schemaVersion": 1, - "baselineHead": "e1da2bd10d0f009924eeb2ea67203da92d71cd0b", + "baselineHead": "192f94e9b57b77e1858e2e4337f0d34bcd6a9c64", "sourceState": { "modified": [ + "libs/langgraph/src/lib/transport/fetch-stream.transport.ts", "libs/langgraph/src/runtime/create-session.ts", + "libs/langgraph/src/runtime/ownership.ts", + "libs/langgraph/src/runtime/publication.ts", "libs/langgraph/src/runtime/stream-projection.ts", "libs/langgraph/src/runtime/testing/binding-fixture.ts" ], "untracked": [ - "libs/langgraph/src/runtime/history-projection.ts", - "libs/langgraph/src/runtime/wire-message.ts" + "libs/langgraph/src/runtime/langgraph-snapshot.ts", + "libs/langgraph/src/runtime/values-projection.ts" ] }, "scope": { @@ -12541,7 +12544,7 @@ "id": "source:libs/langgraph/src/lib/transport/fetch-stream.transport.ts", "kind": "source", "path": "libs/langgraph/src/lib/transport/fetch-stream.transport.ts", - "sha256": "78e712be228f3354bd63f30b13093695701b730c2e8415322d126f2a84e04876" + "sha256": "c32e0512ede7c3b1d1ae945af4ec3e4deed0a4d7207ffc256536105bb7ecfd39" }, { "id": "source:libs/langgraph/src/lib/transport/mock-stream.transport.ts", @@ -12565,7 +12568,7 @@ "id": "source:libs/langgraph/src/runtime/create-session.ts", "kind": "source", "path": "libs/langgraph/src/runtime/create-session.ts", - "sha256": "80db69d29a2fb196d149d10d925f12c7b988695a5e5bbd0c2b6508b15d42bc6e" + "sha256": "5341e188ac9663a3b5d058602e91dd1ba5c0df2983d264d92de0049a792b22ce" }, { "id": "source:libs/langgraph/src/runtime/function-tools.ts", @@ -12579,6 +12582,12 @@ "path": "libs/langgraph/src/runtime/history-projection.ts", "sha256": "f7065276cc2d2122d8cec64311592210b521a9d76cc35550ba94d85d8cbd4742" }, + { + "id": "source:libs/langgraph/src/runtime/langgraph-snapshot.ts", + "kind": "source", + "path": "libs/langgraph/src/runtime/langgraph-snapshot.ts", + "sha256": "2c5f72c9c9b5259426bb94a7a1e44e1e81c401e4c9ec7d6724adfd2fb1b7670f" + }, { "id": "source:libs/langgraph/src/runtime/message-reducer.ts", "kind": "source", @@ -12595,25 +12604,25 @@ "id": "source:libs/langgraph/src/runtime/ownership.ts", "kind": "source", "path": "libs/langgraph/src/runtime/ownership.ts", - "sha256": "5cbf266756c5558997bd187944ccba4b84b9c291653f441e8a0ae6e195dcc22f" + "sha256": "eb7692b96d5113ba690c275624280142df905c78c69b6e5ebdf4f788b6fa7811" }, { "id": "source:libs/langgraph/src/runtime/publication.ts", "kind": "source", "path": "libs/langgraph/src/runtime/publication.ts", - "sha256": "00c5f784139690c001475bd7ef63a5446a4e5d77a99f57353a03d22fbc3569f3" + "sha256": "7d290721417d504313f893ee7e90154b85e9eee27d05f6cce816d592495791e5" }, { "id": "source:libs/langgraph/src/runtime/stream-projection.ts", "kind": "source", "path": "libs/langgraph/src/runtime/stream-projection.ts", - "sha256": "4e2d299abf3d0c5452366515eb9af4a227aecf8b7681496b5f5c6da83f411caf" + "sha256": "ef917d98131c0311846247550ca653d8340f2dcaa97858a5d9a4acaa6bffc1e4" }, { "id": "source:libs/langgraph/src/runtime/testing/binding-fixture.ts", "kind": "source", "path": "libs/langgraph/src/runtime/testing/binding-fixture.ts", - "sha256": "41f4df4d94bfedee7f0bfa5b822bd58c07a7725eba2115ed2156665d1ff733e7" + "sha256": "9299cc57319e881aebe684f6ad5daffd8549aef7e62e83c104f94e7f3a6f5c9c" }, { "id": "source:libs/langgraph/src/runtime/testing/controlled-transport.ts", @@ -12633,6 +12642,12 @@ "path": "libs/langgraph/src/runtime/transport.types.ts", "sha256": "ca5d20d673bec0ca60d3af59b50b948e0dbd93b0cc26efe59d962f666a6c12cb" }, + { + "id": "source:libs/langgraph/src/runtime/values-projection.ts", + "kind": "source", + "path": "libs/langgraph/src/runtime/values-projection.ts", + "sha256": "8c640b8386f2e8cd127188c9961276ec80df820ac05c4b12184fa7954452e0a1" + }, { "id": "source:libs/langgraph/src/runtime/wire-message.ts", "kind": "source", diff --git a/scripts/react-parity/dispositions.json b/scripts/react-parity/dispositions.json index 5eba73d1b..3386f476e 100644 --- a/scripts/react-parity/dispositions.json +++ b/scripts/react-parity/dispositions.json @@ -11355,7 +11355,7 @@ ], "treatment": "shared", "status": "in-progress", - "note": "Only framework-free transport/error dependencies were extracted for the private runtime proof; existing Angular behavior and the remaining assigned migration work are retained." + "note": "SDK event normalization now protects protocol type and namespace from raw application fields. This narrow routing correction retains legacy transport ownership; broader T08 migration remains open." }, { "id": "source:libs/langgraph/src/lib/transport/mock-stream.transport.ts", @@ -11394,7 +11394,7 @@ "treatment": "internal", "status": "in-progress", "reason": "Private staged LangGraph runtime subset; fixture-only and not a neutral public LangGraph root or tarball.", - "note": "Adds explicit optional fixed-thread history loading with cancellation, stale-read and unresolved-work admission guards. This is partial T10 coverage; broader task capabilities and public package cutover remain open." + "note": "Explicit history and root live values now publish application data atomically with messages, including conclusive correlated recovery. Read-only fixed-thread T09/T10 subsets only; broader migration and public cutover remain open." }, { "id": "source:libs/langgraph/src/runtime/function-tools.ts", @@ -11417,6 +11417,17 @@ "reason": "Private staged LangGraph runtime subset; fixture-only and not a neutral public LangGraph root or tarball.", "note": "Authoritative latest-checkpoint transcript projection only: replacement, deletion, reordering, identity sharing and observed wire tools. No full history, branch, pagination, state or interrupt-resume migration claim." }, + { + "id": "source:libs/langgraph/src/runtime/langgraph-snapshot.ts", + "taskIds": [ + "T09", + "T10" + ], + "treatment": "internal", + "status": "in-progress", + "reason": "Private staged LangGraph runtime subset; fixture-only and not a neutral public LangGraph root or tarball.", + "note": "Private broad readonly application-values snapshot extension; undefined distinguishes no current observation from an observed empty record. No application-schema inference or public root export." + }, { "id": "source:libs/langgraph/src/runtime/message-reducer.ts", "taskIds": [ @@ -11446,7 +11457,7 @@ "treatment": "internal", "status": "in-progress", "reason": "Private staged LangGraph runtime subset; fixture-only and not a neutral public LangGraph root or tarball.", - "note": "Bounded runtime proof only; broader task capabilities and final public package cutover remain open." + "note": "Owns immutable plain application values and shares unchanged nested branches; token-only publication reuses owned graphs. Broader execution/publication migration remains open." }, { "id": "source:libs/langgraph/src/runtime/publication.ts", @@ -11456,7 +11467,7 @@ "treatment": "internal", "status": "in-progress", "reason": "Private staged LangGraph runtime subset; fixture-only and not a neutral public LangGraph root or tarball.", - "note": "Bounded runtime proof only; broader task capabilities and final public package cutover remain open." + "note": "Retains concrete backend snapshot extensions and publishes values/messages as one immutable aggregate with no-op identity. No public shared store or whole-task completion claim." }, { "id": "source:libs/langgraph/src/runtime/stream-projection.ts", @@ -11466,7 +11477,7 @@ "treatment": "internal", "status": "in-progress", "reason": "Private staged LangGraph runtime subset; fixture-only and not a neutral public LangGraph root or tarball.", - "note": "Shared wire decoding moved to a private helper for live and history projection. Existing T09 ownership is unchanged; broader event/reducer migration remains open." + "note": "Uses messageMetadata for delta text semantics only on actual message events while preserving a same-named application value. Broader T09 migration remains open." }, { "id": "source:libs/langgraph/src/runtime/testing/binding-fixture.ts", @@ -11477,7 +11488,7 @@ "treatment": "internal", "status": "in-progress", "reason": "Private controlled runtime/binding test helper; excluded from public exports and production packages.", - "note": "Controlled history fixtures now prove explicit load, equal refresh and empty replacement through both native bindings; broader behavioral and native migration coverage remains open." + "note": "Controlled native fixtures cover four history reads, including values-only replacement with unchanged messages, alongside equal/empty refresh. Broader T05/T06 coverage remains open." }, { "id": "source:libs/langgraph/src/runtime/testing/controlled-transport.ts", @@ -11511,6 +11522,17 @@ "reason": "Private staged LangGraph runtime subset; fixture-only and not a neutral public LangGraph root or tarball.", "note": "Bounded runtime proof only; broader task capabilities and final public package cutover remain open." }, + { + "id": "source:libs/langgraph/src/runtime/values-projection.ts", + "taskIds": [ + "T09", + "T10" + ], + "treatment": "internal", + "status": "in-progress", + "reason": "Private staged LangGraph runtime subset; fixture-only and not a neutral public LangGraph root or tarball.", + "note": "Read-only root values/checkpoints and latest-history projection with replacement/deletion, identity sharing and namespace/control exclusions. State writes and broader T09/T10 migration remain open." + }, { "id": "source:libs/langgraph/src/runtime/wire-message.ts", "taskIds": [ diff --git a/scripts/react-parity/runtime-consumer.mjs b/scripts/react-parity/runtime-consumer.mjs index 4f1e9f56b..cfdadb3d1 100644 --- a/scripts/react-parity/runtime-consumer.mjs +++ b/scripts/react-parity/runtime-consumer.mjs @@ -26,7 +26,7 @@ const toolCall = { type: 'ai', id: 'assistant-tool', content: '', tool_calls: [{ const sse = (event, data) => `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`; const textTrace = readFileSync(new URL('../../fixtures/react-parity/traces/langgraph-text-state.sse', import.meta.url), 'utf8'); const savedHistory = [{ - values: { messages: [ + values: { stage: 'saved', profile: { name: 'Saved user' }, messages: [ { id: 'saved-human', type: 'human', content: 'Saved question' }, { id: 'saved-tools', type: 'ai', content: 'Saved tool request', tool_calls: [ { id: 'saved-weather', name: 'weather', args: { city: 'Paris' }, type: 'tool_call' }, @@ -57,7 +57,11 @@ export function runtimeResponse(body) { assert.equal(message.type, 'human'); assert.equal(typeof message.id, 'string'); assert.deepEqual(message, { id: message.id, type: 'human', content: message.content }, 'exact user message fields'); - if (message.content === 'Send') return textTrace.replaceAll('message-parity', `answer-${message.id}`); + if (message.content === 'Send') return textTrace.replaceAll('message-parity', `answer-${message.id}`) + + sse('values|child', { type: 'values', namespace: [], stage: 'child' }) + + sse('updates', { writer: { stage: 'node-update' } }) + + sse('custom', { stage: 'custom' }) + + sse('values', { __interrupt__: [], stage: 'control-envelope' }); if (message.content === 'Tool') return sse('values', { messages: [message, toolCall] }); if (message.content === 'Error') return sse('error', { error: 'FixtureFailure', message: 'PRIVATE backend diagnostic' }); if (message.content === 'Hold') return null; @@ -67,8 +71,26 @@ export function runtimeResponse(body) { export function installedTypeSource(template, kind) { if (kind === 'core') return template; const binding = kind === 'react' ? 'useAgent' : 'observeAgent'; - return template.replace('/* BINDING_IMPORT */', `import { ${binding} } from '@threadplane/${kind}';`) - .replace('export function assertSnapshot(snapshot: AgentSnapshot) {', `export function assertSnapshot(session: AgentSession) {\n const snapshot = ${binding}(session)${kind === 'angular' ? '()' : ''};\n const exact: AgentSnapshot = snapshot;\n void exact;`) + const entry = kind === 'angular' ? './src/runtime-entry.js' : './runtime-entry.js'; + return template.replace('/* BINDING_IMPORT */', `import { ${binding} } from '@threadplane/${kind}';\nimport type { createFixtureSession } from '${entry}';`) + .replace('export function assertSnapshot(snapshot: AgentSnapshot) {', `export function assertSnapshot(session: ReturnType) {\n const snapshot = ${binding}(session)${kind === 'angular' ? '()' : ''};\n const exact: AgentSnapshot = snapshot;\n void exact;`) + .replace('/* BACKEND_VALUES */', `const direct = session.getSnapshot(); + const directValues: Readonly> | undefined = direct.values; + const values: Readonly> | undefined = snapshot.values; + // @ts-expect-error No application schema is inferred from the broad values map. + const assumedCounter: number = snapshot.values?.['counter']; + // @ts-expect-error Concrete backend fields remain readonly. + snapshot.values = {}; + if (snapshot.values) { + // @ts-expect-error Application records remain readonly. + snapshot.values['counter'] = 2; + const nested = snapshot.values['profile']; + if (nested && typeof nested === 'object' && !Array.isArray(nested)) { + // @ts-expect-error Nested application records remain readonly. + nested['name'] = 'mutable'; + } + } + void [directValues, values, assumedCounter];`) .replace(' assertSnapshot(snapshot);', ' void snapshot;'); } @@ -158,7 +180,7 @@ export async function serveRuntimeConsumer(directory) { held.add(response); response.once('close', () => { held.delete(response); notifyAborted(); }); // Actual incremental network bytes, not a whole-response route.fulfill. - response.write(sse('messages', [{ type: 'AIMessageChunk', id: 'held-answer', content: 'Held partial' }, { langgraph_node: 'assistant' }])); + response.write(sse('values', { stage: 'held', transient: true }) + sse('messages', [{ type: 'AIMessageChunk', id: 'held-answer', content: 'Held partial' }, { langgraph_node: 'assistant' }])); notifyHeld(); return; } @@ -205,6 +227,7 @@ export async function runRuntimeScenarios(directory, kind) { browser = await chromium.launch({ headless: true }); context = await browser.newContext(); const page = await context.newPage(); + const expectValues = (value) => expect(page.getByTestId('values')).toHaveText(value === undefined ? 'unobserved' : JSON.stringify(value)); page.on('pageerror', (error) => pageErrors.push(error.message)); page.on('request', (request) => { if (!request.url().startsWith(`${server.url}/`)) unexpected.push(request.url()); @@ -214,6 +237,7 @@ export async function runRuntimeScenarios(directory, kind) { await expect(page.getByTestId('owner')).toHaveText('mounted'); await expect(page.getByTestId('handler-calls')).toHaveText('0'); await expect(page.getByTestId('submissions')).toHaveText('0'); + await expectValues(undefined); assert.equal(server.requests.length, 0, 'mount/observation performs no I/O'); assert.equal(server.historyRequests.length, 0, 'mount/observation performs no history reads'); completed.push('inert mount'); @@ -221,6 +245,7 @@ export async function runRuntimeScenarios(directory, kind) { await page.getByRole('button', { name: 'Load', exact: true }).click(); await expect(page.getByTestId('loads-finished')).toHaveText('1'); await expect(page.getByTestId('load-error')).toHaveText(''); + await expectValues({ stage: 'saved', profile: { name: 'Saved user' } }); await expect(page.getByTestId('text')).toHaveText('Saved tool request\nSaved final answer'); await expect(page.getByTestId('transcript')).toContainText('Saved question'); await expect(page.getByTestId('transcript')).toContainText('Raw historical weather result'); @@ -235,6 +260,7 @@ export async function runRuntimeScenarios(directory, kind) { await page.getByRole('button', { name: 'Load', exact: true }).click(); await expect(page.getByTestId('loads-finished')).toHaveText('2'); await expect(page.getByTestId('load-error')).toHaveText(''); + await expectValues({ stage: 'saved', profile: { name: 'Saved user' } }); await expect(page.getByTestId('text')).toHaveText('Saved tool request\nSaved final answer'); await expect(page.getByTestId('handler-calls')).toHaveText('0'); assert.equal(server.historyRequests.length, 2); @@ -244,6 +270,7 @@ export async function runRuntimeScenarios(directory, kind) { await page.getByRole('button', { name: 'Load', exact: true }).click(); await expect(page.getByTestId('loads-finished')).toHaveText('3'); await expect(page.getByTestId('load-error')).toHaveText(''); + await expectValues(undefined); await expect(page.getByTestId('text')).toHaveText(''); await expect(page.getByTestId('transcript')).toHaveText(''); await expect(page.getByTestId('tool')).toHaveText('[]'); @@ -257,6 +284,7 @@ export async function runRuntimeScenarios(directory, kind) { await expect(page.getByTestId('delivery')).toHaveText('complete:success'); await expect(page.getByTestId('status')).toHaveText('idle'); assert.equal(server.requests.length, 1); + await expectValues({ stage: 'complete' }); completed.push('text success'); const beforeTool = server.requests.length; @@ -266,6 +294,7 @@ export async function runRuntimeScenarios(directory, kind) { await expect(page.getByTestId('status')).toHaveText('idle'); assert.deepEqual(JSON.parse(await page.getByTestId('tool').innerText()), [{ id: 'call-weather', name: 'weather', args: { city: 'Paris' }, status: 'complete', result: { city: 'Paris', temperature: 20 } }]); assert.equal(server.requests.length - beforeTool, 2, 'tool has exactly one run and one result continuation'); + await expectValues({}); completed.push('tool roundtrip'); await page.getByRole('button', { name: 'Error', exact: true }).click(); @@ -273,6 +302,7 @@ export async function runRuntimeScenarios(directory, kind) { await expect(page.getByTestId('error')).not.toHaveText(''); await expect(page.getByTestId('error')).not.toContainText('PRIVATE'); assert.equal(server.requests.length, 4, 'failed run is not retried'); + await expectValues({}); completed.push('visible protected error'); await page.getByRole('button', { name: 'Hold', exact: true }).click(); @@ -280,11 +310,13 @@ export async function runRuntimeScenarios(directory, kind) { await expect(page.getByTestId('text')).toContainText('Held partial'); await expect(page.getByTestId('delivery')).toHaveText('streaming'); await expect(page.getByTestId('status')).toHaveText('running'); + await expectValues({ stage: 'held', transient: true }); await page.getByRole('button', { name: 'Stop', exact: true }).click(); await handshake(server.holdAborted, 'native request abort'); await expect(page.getByTestId('status')).toHaveText('idle'); await expect(page.getByTestId('delivery')).toHaveText('complete:aborted'); assert.equal(server.requests.length, 5); + await expectValues({ stage: 'held', transient: true }); completed.push('incremental DOM update and stop abort'); await page.getByRole('button', { name: 'Send', exact: true }).click(); @@ -293,6 +325,7 @@ export async function runRuntimeScenarios(directory, kind) { await expect(page.getByTestId('text')).toContainText('Hello 🌍.'); await expect(page.getByTestId('handler-calls')).toHaveText('1'); assert.equal(server.requests.length, 6); + await expectValues({ stage: 'complete' }); await expect(page.getByTestId('submissions')).toHaveText('5'); completed.push('reuse after stop'); diff --git a/scripts/react-parity/runtime-consumer.spec.mjs b/scripts/react-parity/runtime-consumer.spec.mjs index eb66c0a96..d3a2ac2d4 100644 --- a/scripts/react-parity/runtime-consumer.spec.mjs +++ b/scripts/react-parity/runtime-consumer.spec.mjs @@ -83,6 +83,8 @@ test('history uses the exact SDK body and counts reads separately from runs', as const first = await read(); assert.equal(first.status, 200); const saved = await first.json(); + assert.equal(saved[0].values.stage, 'saved'); + assert.deepEqual(saved[0].values.profile, { name: 'Saved user' }); assert.deepEqual(saved[0].values.messages.at(-1).content, [{ type: 'text', text: 'Saved final answer' }]); assert.deepEqual(await (await read()).json(), saved); assert.deepEqual(await (await read()).json(), []); @@ -92,6 +94,16 @@ test('history uses the exact SDK body and counts reads separately from runs', as } finally { await server.close(); } }); +test('text fixture retains root application state while exercising ignored child and control data', () => { + const body = { ...heldBody, input: { ...heldBody.input, messages: [{ id: 'user', type: 'human', content: 'Send' }] } }; + const trace = runtime.runtimeResponse(body); + assert.match(trace, /"stage":"complete"/); + assert.match(trace, /event: values\|child/); + assert.match(trace, /event: updates/); + assert.match(trace, /event: custom/); + assert.match(trace, /"__interrupt__":\[\]/); +}); + for (const [label, route, method, body] of [ ['wrong thread', '/api/threads/other/history', 'POST', { limit: 10 }], ['wrong method', '/api/threads/fixture-thread/history', 'GET', undefined], diff --git a/scripts/react-parity/verify-angular-package.mjs b/scripts/react-parity/verify-angular-package.mjs index 0281dc2ed..732bb6413 100644 --- a/scripts/react-parity/verify-angular-package.mjs +++ b/scripts/react-parity/verify-angular-package.mjs @@ -42,8 +42,8 @@ export async function verifyAngularPackage(root = process.cwd()) { prepareInstalledTypes(root, consumer, 'angular'); const contracts = join(consumer, 'installed-types.ts'); writeFileSync(contracts, readFileSync(contracts, 'utf8') + '\n' + specifiers.map((specifier, index) => `import type * as entry${index} from ${JSON.stringify(specifier)};\nexport type Entry${index} = typeof entry${index};`).join('\n')); - runConsumer(process.execPath, [join(consumer, 'node_modules/typescript/bin/tsc'), '-p', 'tsconfig.contracts.json'], consumer); await prepareRuntimeConsumer(root, consumer, 'angular'); + runConsumer(process.execPath, [join(consumer, 'node_modules/typescript/bin/tsc'), '-p', 'tsconfig.contracts.json'], consumer); console.log(runConsumer(process.execPath, angularBuildCommand(consumer), consumer)); const stats = JSON.parse(readFileSync(join(consumer, 'dist/consumer/stats.json'), 'utf8')); assertParserFreeInputs(stats.inputs);