From cd56b9647a20367616f39b2dc0f53c82356a3dea Mon Sep 17 00:00:00 2001 From: Brian Love Date: Mon, 21 Sep 2026 19:13:17 -0700 Subject: [PATCH] feat: load persisted LangGraph history in shared sessions --- fixtures/react-parity/README.md | 34 +- fixtures/react-parity/runtime/README.md | 40 +- fixtures/react-parity/runtime/angular-app.ts | 16 +- fixtures/react-parity/runtime/evidence.json | 455 ++++++------ fixtures/react-parity/runtime/react-app.tsx | 15 +- .../react-parity/runtime/runtime-entry.ts | 9 +- fixtures/react-parity/runtime/scenarios.ts | 1 + libs/angular/src/observe-agent.spec.ts | 35 + libs/langgraph/src/runtime/create-session.ts | 217 +++++- .../src/runtime/history-projection.spec.ts | 415 +++++++++++ .../src/runtime/history-projection.ts | 179 +++++ libs/langgraph/src/runtime/history.spec.ts | 672 ++++++++++++++++++ .../src/runtime/history.type-test.ts | 36 + .../src/runtime/stream-projection.ts | 51 +- .../src/runtime/testing/binding-fixture.ts | 22 +- .../src/runtime/transport.integration.spec.ts | 83 +++ libs/langgraph/src/runtime/wire-message.ts | 51 ++ libs/react/src/use-agent.spec.tsx | 33 + scripts/react-parity/baseline.json | 50 +- scripts/react-parity/dispositions.json | 27 +- scripts/react-parity/runtime-consumer.mjs | 67 +- .../react-parity/runtime-consumer.spec.mjs | 34 + 22 files changed, 2161 insertions(+), 381 deletions(-) create mode 100644 libs/langgraph/src/runtime/history-projection.spec.ts create mode 100644 libs/langgraph/src/runtime/history-projection.ts create mode 100644 libs/langgraph/src/runtime/history.spec.ts create mode 100644 libs/langgraph/src/runtime/history.type-test.ts create mode 100644 libs/langgraph/src/runtime/wire-message.ts diff --git a/fixtures/react-parity/README.md b/fixtures/react-parity/README.md index 69e96ca0a..0be99c106 100644 --- a/fixtures/react-parity/README.md +++ b/fixtures/react-parity/README.md @@ -10,13 +10,20 @@ The existing Angular packages and release group remain the production path. The runtime owns immutable snapshots, request generations, stop/dispose, protected errors, read-only reconciliation after uncertain failures and fixed function-tool -execution. Angular and -React borrow the app-owned session and observe it through their native lifecycles. -The installed consumers run seven scenarios each: inert mount, text, weather tool +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. +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 roundtrip, protected error, held partial text and Stop, reuse after Stop, and 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. 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 @@ -26,11 +33,13 @@ 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,451 records**: the historical 1,438 plus eight private +The current inventory has **1,453 records**: the historical 1,438 plus ten 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. Fourteen existing export records changed declaration/import text; -there are no legacy export-name additions or removals. Existing task assignments +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 +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 inventory scope and are checked by their own tests and package gates. @@ -153,7 +162,7 @@ type-checks the tarballs outside workspace aliases with `skipLibCheck: false`. Its core-only consumer checks all three core exports and rejects extra dependencies. The separate Angular check packs the one Angular APF entry and proves CLI compilation/linking with `skipLibCheck: false`. Both frameworks now run installed -production browser apps with the seven shared scenarios. Inferred native contract +production browser apps with the ten shared scenarios. Inferred native contract probes reject invalid tool names/arguments/results and deep mutations. The private runtime's narrow declaration is compiler-generated against installed core declarations, never hand-written; the staged SDK bundle is fixture-only. Both inspect consumer @@ -207,14 +216,16 @@ burst streams and repeated agent/thread disposal. ## Maintenance and release -The foundation candidate is `codex/react-support-baseline`; the runtime candidate is +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 +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` (`8daea78d35bfa27513474bd624d0e9495af3cfab`) and retains its released lockfile. Creating that local branch does not establish an operated release lane: a maintainer -must own the backport/publication workflow before it is used. No Angular facade -currently depends on a new package. Version/tag enforcement, remote maintenance +must own the backport/publication workflow before it is used. The legacy Angular +package roots retain their existing production path. Version/tag enforcement, remote maintenance policy and a tested backport/rollback remain T37 work. The existing release group is unchanged, and none of the new private packages is publishable. @@ -225,7 +236,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. Renderer reuse and SSR are deferred gates, alongside the broader T01–T39 map. +it. Explicit fixed-thread history loading now covers a further subset of T10. +Renderer reuse and SSR are deferred gates, alongside the broader T01–T39 map. This bounded runtime proof does not establish complete migration parity. | Task | Scope | diff --git a/fixtures/react-parity/runtime/README.md b/fixtures/react-parity/runtime/README.md index 708706f0a..474a8b0d3 100644 --- a/fixtures/react-parity/runtime/README.md +++ b/fixtures/react-parity/runtime/README.md @@ -33,7 +33,8 @@ readonly types directly on each binding's inferred snapshot. 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. Vite bundles the private backend and SDK into temporary ESM, externalizing +type with an optional fixture `load` capability. 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, transitive private declarations, workspace aliases, or core/framework source is @@ -47,25 +48,50 @@ 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 native fixtures expose Send, Tool, Error, Hold and Stop buttons plus text, -status, error, tool result, delivery, submission and handler count outputs. A single app-owned +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, +including deleted/reordered messages and shorter or empty corrections. Equal +reads preserve snapshot identity; unchanged explicit message IDs retain shared +immutable objects. Snapshots stay unchanged while loading and on read failure; +failures reject with protected diagnostics. Cancellation, supersession, stop and +disposal settle promptly even if a transport ignores abort, and stale reads cannot +publish. Loading is refused while execution, uncertain recovery, staged tool +results or asynchronous tool settlement/write work remains unresolved. + +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, +interrupt resume, SSR, or a public LangGraph package cutover. Core public contracts +and the native binding implementations are unchanged. + +The native fixtures expose Load, Send, Tool, Error, Hold and Stop buttons plus text, +transcript, 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. React uses a Vite production build. Angular uses the existing consumer template's installed Angular CLI application builder and real APF linking, with output in `dist/consumer/browser` and input evidence from `dist/consumer/stats.json`. -Both built apps run the same seven browser scenarios in installed Playwright -Chromium: inert mount, successful text, a real local tool handler and exact +Both built apps run the same ten browser scenarios in installed Playwright +Chromium: inert mount, explicit history load, equal history refresh, empty history +replacement, successful text, a real local tool handler and exact two-request result continuation, protected visible server error, held streaming DOM updates and Stop, reuse after Stop, then unmount/dispose/post-disposal submission. Five submissions through the component controls make exactly six run requests (including one tool continuation) and call the handler once. The separate post-disposal submit attempt resolves aborted without making a request. -Request bodies check the catalog and actual serialized ToolMessage payload. +Three explicit Load clicks make exactly three history reads with `{ limit: 10 }` +and no run requests or handler calls. Every completed load must leave its visible +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. A small in-process HTTP fixture serves only built artifacts and the expected -LangGraph run route on dynamic port 0. The held response writes an actual SSE +LangGraph run/history routes on dynamic port 0. The held response writes an actual SSE assistant chunk and stays open. The test observes partial DOM text and streaming delivery before pressing Stop, then awaits the server response-close handshake and aborted delivery. This proves incremental DOM updates and native request abort in diff --git a/fixtures/react-parity/runtime/angular-app.ts b/fixtures/react-parity/runtime/angular-app.ts index 0a198d710..c15697590 100644 --- a/fixtures/react-parity/runtime/angular-app.ts +++ b/fixtures/react-parity/runtime/angular-app.ts @@ -1,4 +1,4 @@ -import { Component } from '@angular/core'; +import { Component, signal } from '@angular/core'; import { bootstrapApplication } from '@angular/platform-browser'; import { observeAgent } from '@threadplane/angular'; import { createFixtureSession } from './runtime-entry.js'; @@ -14,6 +14,7 @@ const submit = (input: string) => { submissions += 1; return session.submit(inpu standalone: true, template: `
+ @@ -21,6 +22,9 @@ const submit = (input: string) => { submissions += 1; return session.submit(inpu {{ snapshot().status }} {{ view().text }} + {{ view().transcript }} + {{ loadsFinished() }} + {{ loadError() }} {{ view().error }} {{ view().tool }} {{ view().delivery }} @@ -30,6 +34,16 @@ const submit = (input: string) => { submissions += 1; return session.submit(inpu `, }) class App { + readonly canLoad = !!session.load; + readonly loadsFinished = signal(0); + readonly loadError = signal(''); + async load() { + if (!session.load) return; + this.loadError.set(''); + try { await session.load(); } + catch { this.loadError.set('History unavailable'); } + finally { this.loadsFinished.update((count) => count + 1); } + } readonly snapshot = observeAgent(session); readonly view = () => display(this.snapshot()); readonly handlerCalls = () => handlerCalls; diff --git a/fixtures/react-parity/runtime/evidence.json b/fixtures/react-parity/runtime/evidence.json index ba813d757..2bf822a74 100644 --- a/fixtures/react-parity/runtime/evidence.json +++ b/fixtures/react-parity/runtime/evidence.json @@ -1,12 +1,14 @@ { "schemaVersion": 1, "status": "verified-local", + "increment": "Explicit fixed-thread history loading (partial T10)", "observedOn": "2026-09-21", + "recordedAt": "2026-09-22T02:09:48.896Z", "source": { - "branch": "codex/shared-runtime-quality", - "baseCommit": "bdcc22ed31aa94f420077e046e88e1481088d453", - "verificationHead": "bdcc22ed31aa94f420077e046e88e1481088d453", - "workingTree": "Verified uncommitted runtime, native binding, fixture and CI changes on the foundation HEAD. The fingerprint identifies the actual selected working-tree bytes, including tests and configuration; this is not a future commit SHA.", + "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.", "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": [ @@ -36,108 +38,36 @@ "excludedPaths": [ "fixtures/react-parity/runtime/evidence.json" ], - "fileCount": 765, - "sha256": "2385a2cf52c60c6e717c03d4b28f399d6228b068315d46947519848c49496669", - "selection": "git ls-files -z --cached --others --exclude-standard -- ; keep existing files and remove excludedPaths. This includes tracked and non-ignored untracked files. Evidence itself is excluded to avoid self-reference; the six research/planning reports are outside all selected paths.", + "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.", "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": { "modified": [ - ".github/workflows/ci.yml", "fixtures/react-parity/README.md", - "fixtures/react-parity/consumers/angular/src/main.ts", - "libs/angular/README.md", - "libs/angular/package.json", - "libs/angular/src/public-api.ts", - "libs/angular/tsconfig.spec.json", - "libs/angular/vite.config.mts", - "libs/core/README.md", - "libs/core/package.json", - "libs/core/src/index.ts", - "libs/core/src/tools/index.ts", - "libs/core/tsconfig.json", - "libs/core/vite.config.mts", - "libs/langgraph/eslint.config.mjs", - "libs/langgraph/project.json", - "libs/langgraph/src/lib/agent.types.ts", - "libs/langgraph/src/lib/client/create-langgraph-client.ts", - "libs/langgraph/src/lib/runtime-operation-reporter.ts", - "libs/langgraph/src/lib/transport/fetch-stream.transport.spec.ts", - "libs/langgraph/src/lib/transport/fetch-stream.transport.ts", - "libs/langgraph/src/lib/transport/transport.interface.ts", - "libs/langgraph/tsconfig.lib.json", - "libs/langgraph/tsconfig.lib.prod.json", - "libs/langgraph/vite.config.mts", - "libs/react/README.md", - "libs/react/package.json", - "libs/react/src/index.ts", - "libs/react/tsconfig.spec.json", - "libs/react/vite.config.mts", - "package-lock.json", - "scripts/ci-scope.spec.mjs", - "scripts/ci-workflow.spec.mjs", - "scripts/react-parity/baseline.json", - "scripts/react-parity/dispositions.json", - "scripts/react-parity/package-policy.mjs", - "scripts/react-parity/verify-angular-package.mjs", - "scripts/react-parity/verify-angular-package.spec.mjs", - "scripts/react-parity/verify-boundaries.mjs", - "scripts/react-parity/verify-boundaries.spec.mjs", - "scripts/react-parity/verify-packages.mjs", - "scripts/react-parity/verify-packages.spec.mjs" - ], - "untracked": [ "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", - "fixtures/react-parity/runtime/vite.config.mts", "libs/angular/src/observe-agent.spec.ts", - "libs/angular/src/observe-agent.ts", - "libs/angular/src/observe-agent.type-test.ts", - "libs/angular/src/test-setup.ts", - "libs/core/src/contracts/agent-session.ts", - "libs/core/src/contracts/agent-snapshot.ts", - "libs/core/src/contracts/contracts.spec.ts", - "libs/core/src/contracts/contracts.type-test.ts", - "libs/core/src/contracts/delivery.ts", - "libs/core/src/contracts/error.ts", - "libs/core/src/contracts/message.ts", - "libs/core/src/contracts/tool.ts", - "libs/core/src/tools/execution-context.ts", - "libs/core/src/tools/function-tool.ts", - "libs/core/src/tools/function-tool.type-test.ts", "libs/langgraph/src/runtime/create-session.ts", - "libs/langgraph/src/runtime/function-tools.ts", - "libs/langgraph/src/runtime/function-tools.type-test.ts", - "libs/langgraph/src/runtime/harness.spec.ts", - "libs/langgraph/src/runtime/message-reducer.spec.ts", - "libs/langgraph/src/runtime/message-reducer.ts", - "libs/langgraph/src/runtime/operation-errors.ts", - "libs/langgraph/src/runtime/ownership.ts", - "libs/langgraph/src/runtime/publication.spec.ts", - "libs/langgraph/src/runtime/publication.ts", - "libs/langgraph/src/runtime/recovery.spec.ts", - "libs/langgraph/src/runtime/session-lifecycle.spec.ts", - "libs/langgraph/src/runtime/stream-projection.spec.ts", "libs/langgraph/src/runtime/stream-projection.ts", "libs/langgraph/src/runtime/testing/binding-fixture.ts", - "libs/langgraph/src/runtime/testing/controlled-transport.ts", - "libs/langgraph/src/runtime/testing/deferred.ts", - "libs/langgraph/src/runtime/tool-execution.spec.ts", - "libs/langgraph/src/runtime/tool-settlement.spec.ts", "libs/langgraph/src/runtime/transport.integration.spec.ts", - "libs/langgraph/src/runtime/transport.types.ts", - "libs/langgraph/tsconfig.runtime-tests.json", - "libs/langgraph/vite.runtime.config.mts", "libs/react/src/use-agent.spec.tsx", - "libs/react/src/use-agent.ts", - "libs/react/src/use-agent.type-test.ts", - "scripts/react-parity/runtime-config.spec.mjs", + "scripts/react-parity/baseline.json", + "scripts/react-parity/dispositions.json", "scripts/react-parity/runtime-consumer.mjs", "scripts/react-parity/runtime-consumer.spec.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" ] } }, @@ -164,123 +94,18 @@ "@playwright/test": "1.58.2", "nx": "22.5.1" }, - "ciNode22ExecutedLocally": false, "browser": "Google Chrome for Testing 145.0.7632.6 (Playwright Chromium)", - "browserInstallation": "Existing local Playwright browser installation; no local Linux --with-deps installation is claimed. CI installs Chromium with --with-deps." - }, - "acceptanceMatrix": { - "status": "Both production-built installed browser verifiers passed against this source fingerprint.", - "frameworks": [ - "Angular installed APF production app", - "React installed Vite production app" - ], - "scenariosPerFramework": 7, - "scenarios": [ - "inert mount", - "text success", - "weather tool roundtrip", - "protected visible error", - "held partial DOM update and Stop/native response close", - "reuse after Stop", - "unmount/app-owned disposal/post-disposal submit" - ], - "notificationAssertions": { - "source": "libs/langgraph/src/runtime/publication.spec.ts and session-lifecycle.spec.ts", - "inertSubscription": 0, - "duplicateSnapshot": 0, - "changedSnapshot": 1, - "submitThenStop": { - "unsubscribedListener": 1, - "remainingListener": 2 - }, - "limit": "Behavioral counts, not render/paint or performance measurements." - }, - "observedPerFramework": { - "componentSubmissions": 5, - "toolContinuations": 1, - "wireRequests": 6, - "toolHandlerCalls": 1, - "postDisposalSubmissions": 1, - "postDisposalRequests": 0, - "pageErrors": 0, - "unexpectedRequests": 0, - "text": "Hello 🌍.", - "toolResult": { - "city": "Paris", - "temperature": 20 - }, - "toolAnswerContains": "20 degrees", - "heldTextContains": "Held partial", - "stopDelivery": "complete:aborted", - "postDisposalOutcome": "aborted" - } + "browserInstallation": "Existing local Playwright browser; no local Linux --with-deps installation is claimed.", + "ciNode22ExecutedLocally": false }, - "cleanup": { - "assertions": "Unmount removes component controls; explicit app disposal resolves; post-disposal submit resolves aborted without extra I/O. Held SSE waits for the native response-close handshake.", - "resources": "Browser context, browser, held responses and server connections close in finally; verifier temporary install/bundle directories are removed in finally, including assertion failures.", - "finalBrowserCleanupVerified": true, - "processes": "Both verifier processes exited 0 after awaited finally cleanup of contexts, browsers and HTTP servers.", - "temporaryDirectories": [ - { - "path": "/var/folders/_b/0t5_pyt94n7dlqkv1gmt29300000gn/T/threadplane-consumer-FdU7ri", - "existsAfterExit": false - }, - { - "path": "/var/folders/_b/0t5_pyt94n7dlqkv1gmt29300000gn/T/threadplane-angular-consumer-XpH0tb", - "existsAfterExit": false - } - ], - "runtimeBundleDirectoriesRemaining": [] - }, - "inventory": { - "historicalRows": 1438, - "currentRows": 1451, - "newProductionRuntimeSources": 8, - "newTestingHelpers": 3, - "newRuntimeVitestConfigSources": 1, - "newRuntimeTypeConfigAssets": 1, - "legacyExports": 550, - "uniqueLocalDefinitions": 514, - "legacyExportRecordsWithReviewedTextChanges": 14, - "legacyExportNameChanges": 0, - "existingAssignmentsPreserved": true, - "scope": "Existing 16-library inventory only. Core/native package public contracts are checked separately. T01-T39 assignments are retained; only touched bounded subsets are in-progress." - }, - "ci": { - "pullRequestBases": "unrestricted, including stacked branches", - "pushBranches": [ - "main" - ], - "deploymentGuards": "Existing main-ref and push-event guards preserved.", - "runtimeEnvironment": "Isolated Node Vitest config, no Angular setup/plugins.", - "chromium": "npx playwright install --with-deps chromium precedes both packed browser calls.", - "builtScan": "Production legacy builds precede emitted boundary scan." - }, - "limits": [ - "Legacy LangGraph root and tarball remain Angular.", - "Private staged runtime is fixture-only; no neutral LangGraph tarball or public runtime entry is claimed.", - "Public core/native contracts compile against installed declarations; the narrow private fixture declaration is compiler-generated against installed core, never hand-written.", - "Native inferred type probes cover heterogeneous tool names, arguments, results and deep readonly behavior.", - "HTTP/SSE fixture proves incremental DOM updates and native request abort, not compositor paint.", - "Production React StrictMode has no development replay; native unit tests cover replay separately.", - "No full product parity, renderer reuse, SSR, calibrated latency, heap retention or performance claim.", - "CI Node22 and other platforms were not executed locally.", - "Test-only imports add Nx dependencies on legacy builds although production package graphs remain isolated.", - "No production backend, publication, release, deployment or remote data was exercised." - ], - "warnings": [ - "Nx reports NO_COLOR/FORCE_COLOR overlap; legacy native builds report stale Browserslist data, ng-packagr export-condition overrides and keepLifecycleScripts notices.", - "Vite reports that use client directives are ignored in the client-only production consumer bundle. No SSR claim is made." - ], - "recordedAt": "2026-09-21T23:00:37.935Z", "executed": [ { "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": 407, + "testsPassed": 412, "testsFailed": 0, - "log": "/tmp/r08-final-focused.log", - "logSha256": "c2d520eccb4f0002282c2f043ac3b0f08dbcc756d95a559ac4d5448ef6f94bcf" + "log": "/tmp/h04-final-focused.log", + "logSha256": "8d929ce23bb0c125cdbb859166c4a9fef70b8304ae4de7fa0bc12ee035c77fe8" }, { "command": "NX_DAEMON=false npx nx run-many -t lint test type-tests build --projects=core,content,angular,react --parallel=2 --skip-nx-cache", @@ -289,45 +114,53 @@ "targetConfigurations": 16, "behaviorTests": { "core": 2, - "angular": 6, - "react": 6 + "angular": 7, + "react": 7 }, - "content": "Empty scaffold: test target passes with passWithNoTests; no content behavior test count claimed.", - "log": "/tmp/r08-final-foundations.log", - "logSha256": "0738c2cdf2575fa1a6269d3c96ef95a88663d80450a1042b9ae5dfe689b4bc91" + "content": "Empty scaffold with passWithNoTests; no content behavior count claimed.", + "log": "/tmp/h04-final-foundations.log", + "logSha256": "499d4f3a8ed8fdd6af1c26252aa418a9f70b667ce6fb28c39f5f4793d654d18c" }, { "command": "NX_DAEMON=false npx nx run langgraph:runtime-quality --skip-nx-cache", "exitCode": 0, - "testFiles": 9, - "testsPassed": 156, - "log": "/tmp/r08-final-runtime-quality.log", - "logSha256": "9675327833ac0e4987c58d423da570da2bc476e487eec58838d61ef613c693f1" + "testFiles": 11, + "testsPassed": 207, + "log": "/tmp/h04-final-runtime.log", + "logSha256": "6b18035d54c771dfba77a48230512e9001b268075bc834e7f5b13dc5d4fd2e1c" }, { "command": "NX_DAEMON=false npx nx run langgraph:runtime-type-tests --skip-nx-cache", "exitCode": 0, - "log": "/tmp/r08-final-runtime-types.log", + "log": "/tmp/h04-final-runtime-types.log", "logSha256": "6e5c64695f3aad8fd1b474649ebb26c81204d54e45d589424d65e2b0843c921d" }, + { + "command": "NX_DAEMON=false npx nx lint langgraph --skip-nx-cache", + "exitCode": 0, + "errors": 0, + "existingWarnings": 68, + "log": "/tmp/h04-final-langgraph-lint.log", + "logSha256": "1fc98b0ca0707512f2ebf1084ed37e8e979f82be6b3cfb7a7194e17795710432" + }, { "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/r08-final-production-builds.log", - "logSha256": "eeb87b96a645703d0511f4c26d7360397ad07cfbb6a02042ca78bb4e6cf427ff" + "log": "/tmp/h04-final-production-builds.log", + "logSha256": "b757c69ada7ddf18e2ef60f8b702aec960243099cea7b66ab3af8d54c445d453" }, { "command": "node scripts/react-parity/verify-boundaries.mjs", "exitCode": 0, - "log": "/tmp/r08-final-source-boundaries.log", + "log": "/tmp/h04-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/r08-final-built-boundaries.log", + "log": "/tmp/h04-final-built-boundaries.log", "logSha256": "c983a97d17ff7ced7aa9113afa031f464103f566f1bc94eec8160558ec2553f7" }, { @@ -336,49 +169,170 @@ "privatePlainTarballs": 3, "esmTypeExports": 9, "isolatedCoreExports": 3, - "browserScenariosPassed": 7, - "log": "/tmp/r08-final-packages.log", - "logSha256": "91125a29c21f67710c85520d48339c49ef3854b0e7691adfb64747ff397ad319" + "browserScenariosPassed": 10, + "log": "/tmp/h04-final-packages.log", + "logSha256": "8e749830c928e0102cb2b74c34361667d62022c2767fe1d36d11acd338213943" }, { "command": "node scripts/react-parity/verify-angular-package.mjs", "exitCode": 0, "angularAPFExports": 1, - "browserScenariosPassed": 7, - "log": "/tmp/r08-final-angular-package.log", - "logSha256": "5d2c29a1435ed80ae0f1981319ef77eba62e56c35ab51d175a9bb0bdaf137c28" + "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" }, { "command": "node scripts/react-parity/inventory.mjs --write-baseline; node scripts/react-parity/inventory.mjs --check", "exitCode": 0, - "inventoryRows": 1451, - "reviewedChange": "Final refresh changed only hashes for create-session.ts, message-reducer.ts and stream-projection.ts; no IDs/counts/assignments changed.", - "log": "/tmp/r08-final-inventory.log", - "logSha256": "5b8f74cc9f500da010e2933303fbeab9e4b06813a3cf5a95c97416a1621acd93" + "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" }, { "command": "git diff --check", "exitCode": 0, - "log": "/tmp/r08-final-diff-check.log", + "log": "/tmp/h04-final-diff-check.log", "logSha256": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" } ], "reusedUnchangedLegacyChecks": [ { - "command": "NX_DAEMON=false npx nx test langgraph --skip-nx-cache", - "exitCode": 0, - "log": "/tmp/runtime-final-legacy-langgraph.log", - "provenance": "Same-session parent run, with successful output inspected. Final review fixes changed private runtime files only; the Angular suite excludes src/runtime. These unchanged legacy results are reused, not represented as a new post-fix run. No test count inferred.", - "logSha256": "eb2085807b18946b5e3340b37475f8ae706f86471c392635748fe83b04f5b42e" - }, - { - "command": "NX_DAEMON=false npx nx run langgraph:type-tests --skip-nx-cache", + "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/runtime-final-legacy-types.log", - "provenance": "Same-session parent run, with successful output inspected. Final review fixes changed private runtime files only; the Angular suite excludes src/runtime. These unchanged legacy results are reused, not represented as a new post-fix run. No test count inferred.", - "logSha256": "48a8dac2f4cc97169640bd20382f995f07e2cb40e125fac98b926d8ba39103b8" + "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." + }, + "acceptanceMatrix": { + "status": "Both production-built installed browser verifiers passed.", + "frameworks": [ + "Angular installed APF production app", + "React installed Vite production app" + ], + "scenariosPerFramework": 10, + "scenarios": [ + "inert mount", + "explicit history load", + "equal history refresh", + "empty history replacement", + "text success", + "weather tool roundtrip", + "protected visible error", + "held partial DOM update and Stop/native response close", + "reuse after Stop", + "unmount/app-owned disposal/post-disposal submit" + ], + "observedPerFramework": { + "historyReads": 3, + "historyRequestBodies": [ + { + "limit": 10 + }, + { + "limit": 10 + }, + { + "limit": 10 + } + ], + "historyRunRequests": 0, + "historyHandlerCalls": 0, + "componentSubmissions": 5, + "toolContinuations": 1, + "runRequests": 6, + "toolHandlerCalls": 1, + "postDisposalSubmissions": 1, + "postDisposalRequests": 0, + "pageErrors": 0, + "unexpectedRequests": 0, + "loadedAssistantText": "Saved tool request\nSaved final answer", + "historicalResultText": "Raw historical weather result", + "loadedPendingTool": { + "id": "saved-count", + "name": "count", + "args": { + "values": [ + "saved" + ] + }, + "status": "pending" + }, + "equalRefreshLoadError": "", + "emptyReplacementTranscript": "", + "emptyReplacementToolCalls": [], + "text": "Hello 🌍.", + "toolResult": { + "city": "Paris", + "temperature": 20 + }, + "toolAnswerContains": "20 degrees", + "heldTextContains": "Held partial", + "stopDelivery": "complete:aborted", + "postDisposalOutcome": "aborted" + }, + "notificationAssertions": { + "source": "libs/langgraph/src/runtime/history.spec.ts", + "pendingLoad": 0, + "successfulLoadCumulative": 1, + "equalRefreshCumulative": 1, + "emptyReplacementCumulative": 2, + "failedRefreshAdditional": 0, + "limit": "Unit-test behavioral counts, not render/paint or 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.", + "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", + "existsAfterExit": false + }, + { + "path": "/var/folders/_b/0t5_pyt94n7dlqkv1gmt29300000gn/T/threadplane-angular-consumer-3OGMOu", + "existsAfterExit": false + } + ], + "runtimeBundleDirectoriesRemaining": [] + }, + "inventory": { + "historicalFoundationRows": 1438, + "previousRuntimeRows": 1451, + "currentRows": 1453, + "newHistorySourceFiles": [ + "libs/langgraph/src/runtime/history-projection.ts", + "libs/langgraph/src/runtime/wire-message.ts" + ], + "sourceFiles": 475, + "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." + }, "runtimePolicy": { "ownedSDKDefaultMaxRetries": 0, "explicitOptIn": "Positive clientOptions.maxRetries enables SDK retries. Custom transports own their retry policy; no global no-retry guarantee is claimed.", @@ -396,8 +350,8 @@ "fileBytes": 51239870, "lockLocationsIncludingOptionalPlatforms": 74, "productionAppModulesTransformed": 34, - "productionAppJavaScriptRawReported": "408.65 kB", - "productionAppJavaScriptGzipReported": "123.14 kB", + "productionAppJavaScriptRawReported": "412.73 kB", + "productionAppJavaScriptGzipReported": "124.43 kB", "developmentRootImportProbe": { "inputs": 5, "bytes": 47778, @@ -410,16 +364,31 @@ "lockLocationsIncludingOptionalPlatforms": 516, "bundleInputs": 257, "contentParserInputs": 0, - "productionAppRawReported": "309.75 kB", - "productionAppEstimatedTransferReported": "81.94 kB", + "productionAppRawReported": "314.20 kB", + "productionAppEstimatedTransferReported": "83.30 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." }, - "notRun": [ - "CI Node22 or Linux environment", - "Angular20/22 consumer lanes", - "SSR, renderer reuse, full parity or calibrated performance benchmarks", - "Remote backend/deployment smoke, release or publication" + "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." + }, + "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.", + "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.", + "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.", + "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." ] } diff --git a/fixtures/react-parity/runtime/react-app.tsx b/fixtures/react-parity/runtime/react-app.tsx index 56f8857c8..8abbd55bf 100644 --- a/fixtures/react-parity/runtime/react-app.tsx +++ b/fixtures/react-parity/runtime/react-app.tsx @@ -1,4 +1,4 @@ -import { StrictMode } from 'react'; +import { StrictMode, useState } from 'react'; import { createRoot } from 'react-dom/client'; import { useAgent } from '@threadplane/react'; import { createFixtureSession } from './runtime-entry.js'; @@ -12,8 +12,18 @@ const submit = (input: string) => { submissions += 1; return session.submit(inpu function App() { const snapshot = useAgent(session); + const [loadsFinished, setLoadsFinished] = useState(0); + const [loadError, setLoadError] = useState(''); + const load = async () => { + if (!session.load) return; + setLoadError(''); + try { await session.load(); } + catch { setLoadError('History unavailable'); } + finally { setLoadsFinished((count) => count + 1); } + }; const view = display(snapshot); return
+ @@ -21,6 +31,9 @@ function App() { {snapshot.status} {view.text} + {view.transcript} + {loadsFinished} + {loadError} {view.error} {view.tool} {view.delivery} diff --git a/fixtures/react-parity/runtime/runtime-entry.ts b/fixtures/react-parity/runtime/runtime-entry.ts index e56e8ee7d..644edd326 100644 --- a/fixtures/react-parity/runtime/runtime-entry.ts +++ b/fixtures/react-parity/runtime/runtime-entry.ts @@ -8,7 +8,9 @@ export function createFixtureSession( endpoint: string, threadId: string, onHandler: () => void = () => undefined -): AgentSession { +): AgentSession & { + load?: (options?: { signal?: AbortSignal }) => Promise; +} { return createSession({ assistantId: 'fixture-assistant', threadId, @@ -24,7 +26,10 @@ export function createFixtureSession( }, count: { description: 'Count values', - handler: ({ values }: { values: readonly string[] }) => values.length, + handler: ({ values }: { values: readonly string[] }) => { + onHandler(); + return values.length; + }, }, }, }); diff --git a/fixtures/react-parity/runtime/scenarios.ts b/fixtures/react-parity/runtime/scenarios.ts index 24e1733e7..c74d4256f 100644 --- a/fixtures/react-parity/runtime/scenarios.ts +++ b/fixtures/react-parity/runtime/scenarios.ts @@ -10,6 +10,7 @@ export function display(snapshot: AgentSnapshot) { const delivery = assistant.at(-1)?.delivery; return { text: assistant.map((message) => message.content).join('\n'), + transcript: snapshot.messages.map((message) => message.content).join('\n'), error: snapshot.error?.message ?? '', tool: JSON.stringify(snapshot.toolCalls), delivery: delivery?.phase === 'complete' ? `complete:${delivery.outcome}` : delivery?.phase ?? '', diff --git a/libs/angular/src/observe-agent.spec.ts b/libs/angular/src/observe-agent.spec.ts index 5bc28f767..94c2a81ef 100644 --- a/libs/angular/src/observe-agent.spec.ts +++ b/libs/angular/src/observe-agent.spec.ts @@ -83,6 +83,41 @@ afterEach(async () => { }); describe('observeAgent borrowed session', () => { + it('observes explicit history loads without owning reads, refreshes, or teardown', async () => { + const f = fixture(); + TestBed.configureTestingModule({ + imports: [Chat], + providers: [{ provide: SESSION, useValue: f.session }], + }); + const view = TestBed.createComponent(Chat); + view.detectChanges(); + expect(f.session.load).toBeTypeOf('function'); + expect(f.history.reads).toBe(0); + let notifications = 0; + 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'); + 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); + release(); + view.destroy(); + const reattached = observe(f.session); + expect(reattached.snapshot()).toBe(snapshot); + expect(f.history.reads).toBe(2); + reattached.destroy(); + f.history.value = []; + await f.session.load(); + expect(f.session.getSnapshot().messages).toEqual([]); + expect(f.history.reads).toBe(3); + expect(f.handlerCalls).toBe(0); + expect(f.streams).toHaveLength(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 () => { const f = fixture(); TestBed.configureTestingModule({ diff --git a/libs/langgraph/src/runtime/create-session.ts b/libs/langgraph/src/runtime/create-session.ts index 3e8ef75fb..72c0bd6c3 100644 --- a/libs/langgraph/src/runtime/create-session.ts +++ b/libs/langgraph/src/runtime/create-session.ts @@ -4,6 +4,7 @@ import { type AgentSession, type CompleteOutcome, type ToolCall, + type ToolContract, } from '@threadplane/core'; import type { CheckedTools, @@ -15,6 +16,8 @@ import type { ThreadState } from '@langchain/langgraph-sdk'; import { FetchStreamTransport } from '../lib/transport/fetch-stream.transport'; import { initialMessageState, reduceMessages } from './message-reducer'; import { createPublication } from './publication'; +import { projectHistory } from './history-projection'; +import { createSafeRequestError } from './operation-errors'; import { failureProjection, finalizeProjection, @@ -50,6 +53,24 @@ export interface SessionOptions { readonly executionStore?: ToolExecutionStore; } +/** Backend-private capability; core sessions and borrowed observers stay minimal. */ +export type LangGraphSession< + TTools extends { [K in keyof TTools]: ToolContract } = Record< + string, + ToolContract + > +> = AgentSession & { + load?(options?: { readonly signal?: AbortSignal }): Promise; +}; + +interface HistoryRead { + readonly controller: AbortController; + readonly result: Promise; + readonly resolve: () => void; + readonly reject: (error: Error) => void; + unlink?: () => void; +} + interface Attempt { readonly controller: AbortController; readonly generation: string; @@ -73,15 +94,15 @@ interface Attempt { * during a run; it only reads history and never creates another logical run. */ export function createSession>( options: SessionOptions & { readonly tools: T & CheckedTools } -): AgentSession>; +): LangGraphSession>; export function createSession( options: SessionOptions & { readonly tools?: undefined } -): AgentSession; +): LangGraphSession; export function createSession( options: SessionOptions & { readonly tools?: Record; } -): AgentSession { +): LangGraphSession { const { assistantId, threadId } = options; const { definitions, catalog } = captureTools(options.tools); const typedTools = options.tools !== undefined; @@ -91,6 +112,9 @@ export function createSession( }; const buffer = createToolBuffer(); const resolvedTools = new Set(); + // Execution dedupe survives transcript replacement. Authored result provenance + // belongs only to the current transcript; a wire string cannot restore it. + const authoredTools = new Set(); const transport = options.transport ?? new FetchStreamTransport(options.apiUrl ?? '', undefined, { @@ -100,7 +124,11 @@ export function createSession( const protectedTransport = transport instanceof FetchStreamTransport && transport.protectsOperationErrors; - const canCheck = typeof transport.getHistory === 'function'; + const getHistory = + typeof transport.getHistory === 'function' + ? transport.getHistory.bind(transport) + : undefined; + const canCheck = !!getHistory; const publication = createPublication({ status: 'idle', messages: [], @@ -112,6 +140,9 @@ export function createSession( let disposed = false; let revision = 0; let checkController: AbortController | undefined; + let loading: HistoryRead | undefined; + let pendingToolSettlements = 0; + let pendingToolWrites = 0; const owns = (attempt: Attempt) => owner === attempt && !disposed; function publish(status: 'idle' | 'running' | 'error', error?: AgentError) { @@ -122,7 +153,7 @@ export function createSession( ? state.toolCalls.filter( (call) => definitions.has(call.name) && - (resolvedTools.has(call.id) || + (authoredTools.has(call.id) || !state.messages.some( (message) => message.role === 'tool' && message.toolCallId === call.id @@ -138,6 +169,22 @@ export function createSession( checkController = undefined; return previous; } + const ownsLoad = (read: HistoryRead) => loading === read && !disposed; + function detachLoad() { + const previous = loading; + loading = undefined; + previous?.resolve(); + return previous; + } + function closeLoad(read: HistoryRead | undefined, abort = true) { + if (!read) return; + const unlink = read.unlink; + read.unlink = undefined; + // Ownership must already be committed: abort/remove-listener hooks may + // synchronously start another command. Cleanup never owns its replacement. + if (abort) read.controller.abort(); + unlink?.(); + } function close(attempt: Attempt, abort = false, final = true) { const unlink = final ? attempt.unlink : undefined; if (final) attempt.unlink = undefined; @@ -177,6 +224,7 @@ export function createSession( const current = state.toolCalls.find((entry) => entry.id === call.id); if (current?.status !== 'running') continue; resolvedTools.add(call.id); + authoredTools.add(call.id); state = reduceMessages(state, { type: 'tool', toolCall: resultCall(call, cancelledResult(call.id)), @@ -198,8 +246,17 @@ export function createSession( throw new Error( 'Persisting terminal tool results requires transport.updateState().' ); - await transport.updateState(threadId, { messages: batch.messages }, signal); - batch.acknowledge(); + pendingToolWrites += 1; + try { + await transport.updateState( + threadId, + { messages: batch.messages }, + signal + ); + batch.acknowledge(); + } finally { + pendingToolWrites -= 1; + } } async function executeTools(attempt: Attempt, groups: number) { @@ -219,32 +276,42 @@ export function createSession( toolCall: { ...call, status: 'running' }, }); } + pendingToolSettlements += calls.length; publish('running'); await Promise.all( calls.map(async (call) => { - const definition = definitions.get(call.name); - if (!definition) return; - const result = await executeTool( - definition, - call, - attempt.controller.signal, - { threadId, toolCallId: call.id }, - store, - groups >= 10 - ); - resolvedTools.add(call.id); - buffer.stage(call.id, result); - if (owns(attempt)) { - state = reduceMessages(state, { - type: 'tool', - toolCall: resultCall(call, result), - }); - publish('running'); - } - if (!owns(attempt)) { - // Required durable cleanup may finish after stop/dispose. It can only - // persist results; it has no route back to publication or run creation. - void flushTools(new AbortController().signal).catch(() => undefined); + try { + const definition = definitions.get(call.name); + if (!definition) return; + const result = await executeTool( + definition, + call, + attempt.controller.signal, + { threadId, toolCallId: call.id }, + store, + groups >= 10 + ); + resolvedTools.add(call.id); + authoredTools.add(call.id); + buffer.stage(call.id, result); + if (owns(attempt)) { + state = reduceMessages(state, { + type: 'tool', + toolCall: resultCall(call, result), + }); + publish('running'); + } + if (!owns(attempt)) { + // Required durable cleanup may finish after stop/dispose. It can only + // persist results; it has no route back to publication or run creation. + try { + await flushTools(new AbortController().signal); + } catch { + /* The staged result remains available for explicit handoff. */ + } + } + } finally { + pendingToolSettlements -= 1; } }) ); @@ -253,11 +320,13 @@ export function createSession( ); } function stopExecution() { + const reading = detachLoad(); const checking = invalidateCheck(); const attempt = detach('aborted'); if (attempt) publish('idle'); checking?.abort(); if (attempt) close(attempt, true); + closeLoad(reading); } function reconcile( @@ -451,6 +520,7 @@ export function createSession( let attempt: Attempt | undefined; const beginning = publication.command(() => { if (disposed || submitOptions?.signal?.aborted) return; + const reading = detachLoad(); const checking = invalidateCheck(); const previous = detach('interrupted'); const generation = crypto.randomUUID(); @@ -505,6 +575,7 @@ export function createSession( publish('running'); checking?.abort(); if (previous) close(previous, true); + closeLoad(reading); }); // Even nested observer commands finish draining before this continuation can // issue I/O. A stop/dispose from the running publication can prevent it. @@ -515,6 +586,88 @@ export function createSession( }); } + async function readHistory(read: HistoryRead) { + if (!ownsLoad(read) || !getHistory) return; + try { + const history = await getHistory(threadId, read.controller.signal); + await publication.command(() => { + if (!ownsLoad(read)) return; + const projected = projectHistory( + state, + history, + typedTools + ? { registeredTools: new Set(definitions.keys()) } + : undefined + ); + // 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; + authoredTools.clear(); + loading = undefined; + read.resolve(); + publish('idle'); + closeLoad(read, false); + }); + } catch { + await publication.command(() => { + if (!ownsLoad(read)) return; + loading = undefined; + read.reject(createSafeRequestError()); + closeLoad(read); + }); + } + } + + function load(options?: { readonly signal?: AbortSignal }): Promise { + let read: HistoryRead | undefined; + const beginning = publication.command(() => { + if (disposed || options?.signal?.aborted) return; + if ( + owner || + recoveryAttempt || + pendingToolSettlements || + pendingToolWrites || + buffer.snapshot().messages.length + ) + throw new Error( + 'History cannot replace an active request, recovery, or unsettled tool results.' + ); + const previous = detachLoad(); + let resolve!: HistoryRead['resolve']; + let reject!: HistoryRead['reject']; + const result = new Promise((done, failed) => { + resolve = done; + reject = failed; + }); + const created: HistoryRead = { + controller: new AbortController(), + result, + resolve, + reject, + }; + read = created; + loading = created; + const external = options?.signal; + if (external) { + const abort = () => { + void publication.command(() => { + if (ownsLoad(created)) closeLoad(detachLoad()); + }); + }; + created.unlink = () => external.removeEventListener('abort', abort); + external.addEventListener('abort', abort, { once: true }); + if (external.aborted) abort(); + } + closeLoad(previous); + }); + return beginning.then(() => { + if (!read) return; + void readHistory(read); + return read.result; + }); + } + async function checkStatus() { let checking: | { revision: number; attempt: Attempt; controller: AbortController } @@ -579,11 +732,12 @@ export function createSession( if (disposed) return; stopExecution(); }), - ...(canCheck ? { checkStatus } : {}), + ...(canCheck ? { checkStatus, load } : {}), dispose: () => publication.command(() => { if (disposed) return; disposed = true; + const reading = detachLoad(); const checking = invalidateCheck(); const attempt = detach('aborted'); recoveryAttempt = undefined; @@ -591,6 +745,7 @@ export function createSession( publication.clearListeners(); checking?.abort(); if (attempt) close(attempt, true); + closeLoad(reading); }), }; } diff --git a/libs/langgraph/src/runtime/history-projection.spec.ts b/libs/langgraph/src/runtime/history-projection.spec.ts new file mode 100644 index 000000000..5e2cfd77f --- /dev/null +++ b/libs/langgraph/src/runtime/history-projection.spec.ts @@ -0,0 +1,415 @@ +import type { ThreadState } from '@langchain/langgraph-sdk'; +import { describe, expect, it } from 'vitest'; +import { completeDelivery, staticDelivery } from '@threadplane/core'; +import { initialMessageState, reduceMessages } from './message-reducer'; +import { projectHistory } from './history-projection'; + +const human = (id: string, content = 'Question') => ({ + type: 'human', + id, + content, +}); +const ai = (id: string, content = 'Answer', tool_calls?: unknown[]) => ({ + type: 'ai', + id, + content, + ...(tool_calls === undefined ? {} : { tool_calls }), +}); +const call = (id: string, name = 'work', args: unknown = { id }) => ({ + id, + name, + args, +}); +const result = (id: string, tool_call_id: string, content = 'Saved') => ({ + type: 'tool', + id, + tool_call_id, + content, +}); +function checkpoint( + messages: unknown[], + overrides: Partial = {} +): ThreadState { + return { + values: { messages } as ThreadState['values'], + next: [], + tasks: [], + checkpoint: { + thread_id: 'thread', + checkpoint_id: 'latest', + checkpoint_ns: '', + checkpoint_map: {}, + }, + metadata: null, + created_at: null, + parent_checkpoint: null, + ...overrides, + }; +} +const initial = () => initialMessageState(); + +describe('pure authoritative history projection', () => { + it('reads only the latest checkpoint, without merging older messages or pause evidence', () => { + const state = projectHistory(initial(), [ + checkpoint([human('u'), ai('a', 'Latest')]), + checkpoint([ai('old')], { values: { __interrupt__: ['old'] } }), + ]); + expect(state.messages.map((message) => message.id)).toEqual(['u', 'a']); + expect(state.messages[1].content).toBe('Latest'); + expect(state.messages[1].delivery).toEqual(staticDelivery('a')); + }); + + it('treats empty history and a latest checkpoint without messages as authoritative empty', () => { + const prior = projectHistory(initial(), [ + checkpoint([ai('a', 'Old', [call('c')])]), + ]); + for (const history of [[], [checkpoint([], { values: {} })]]) { + const empty = projectHistory(prior, history); + expect(empty).toEqual(initial()); + expect(projectHistory(empty, history)).toBe(empty); + } + }); + + it('replaces, removes, reorders and corrects shorter or empty content while sharing unchanged IDs', () => { + const before = projectHistory(initial(), [ + checkpoint([human('u'), ai('a', 'Long answer'), ai('keep'), ai('gone')]), + ]); + const shorter = projectHistory(before, [ + checkpoint([ai('keep'), ai('a', 'Long'), human('u')]), + ]); + expect(shorter.messages.map((message) => message.id)).toEqual([ + 'keep', + 'a', + 'u', + ]); + expect(shorter.messages[0]).toBe(before.messages[2]); + expect(shorter.messages[2]).toBe(before.messages[0]); + expect(shorter.messages[1].content).toBe('Long'); + const empty = projectHistory(shorter, [checkpoint([ai('a', '')])]); + expect(empty.messages[0].content).toBe(''); + expect(before.messages[1].content).toBe('Long answer'); + }); + + it('keeps equal state, arrays, messages and nested tool identities across separately allocated reads', () => { + const history = [ + checkpoint([ai('a', '', [call('c', 'work', { nested: ['x'] })])]), + ]; + const first = projectHistory(initial(), history); + const second = projectHistory(first, structuredClone(history)); + expect(first.toolCalls).toHaveLength(1); + expect(second).toBe(first); + expect(second.messages).toBe(first.messages); + expect(second.toolCalls).toBe(first.toolCalls); + expect(second.toolCalls[0].args).toBe(first.toolCalls[0].args); + expect(second.messages[0].delivery).toEqual(staticDelivery('a')); + }); + + it('resets live canonical and alias bookkeeping and adopts static delivery', () => { + const live = reduceMessages(initial(), { + type: 'message', + mode: 'canonical', + message: { + id: 'a', + role: 'assistant', + content: 'Answer', + delivery: completeDelivery('attempt', 'success'), + }, + }); + const withAliases = { + ...live, + aliases: [{ from: 'draft', to: 'a', generation: 'attempt' }], + }; + const projected = projectHistory(withAliases, [checkpoint([ai('a')])]); + expect(projected.canonical).toEqual([]); + expect(projected.aliases).toEqual([]); + expect(projected.messages[0].delivery).toEqual(staticDelivery('a')); + }); + + it('uses last explicit duplicate occurrence and never publishes duplicate identities or superseded tools', () => { + const state = projectHistory(initial(), [ + checkpoint([ + ai('same', 'Old', [call('old')]), + human('u'), + ai('same', 'Final', [call('new')]), + ]), + ]); + expect(state.messages.map((message) => message.id)).toEqual(['u', 'same']); + expect(state.messages[1].content).toBe('Final'); + expect(state.toolCalls.map((entry) => entry.id)).toEqual(['new']); + }); + + it('reserves all explicit IDs before assigning deterministic index fallbacks, including unknown roles', () => { + const history = [ + checkpoint([ + { type: 'human', content: 'No id' }, + ai('history-message-0'), + { type: 'unsupported', id: 'history-message-0-1' }, + { type: 'ai', content: 'No id either' }, + ai('history-message-3'), + ]), + ]; + const first = projectHistory(initial(), history); + expect(first.messages.map((message) => message.id)).toEqual([ + 'history-message-0-2', + 'history-message-0', + 'history-message-3-1', + 'history-message-3', + ]); + expect(new Set(first.messages.map((message) => message.id)).size).toBe(4); + expect(projectHistory(first, structuredClone(history))).toBe(first); + }); + + it('normalizes wire role aliases and text blocks using the text-only stream vocabulary', () => { + const state = projectHistory(initial(), [ + checkpoint([ + { + type: 'HumanMessage', + id: 'u', + content: [ + { type: 'text', text: 'Hi' }, + { type: 'image', url: 'ignored' }, + { type: 'text', text: '!' }, + ], + }, + { + type: 'AIMessage', + id: 'a', + content: [{ type: 'text', text: 'Hello' }], + }, + { role: 'system', id: 's', content: 'Rules' }, + { + type: 'ToolMessage', + id: 't', + name: 'work', + tool_call_id: 'c', + content: 'Wire result', + }, + { + role: 'assistant', + id: 'empty', + content: { text: 'Not a supported block list' }, + }, + null, + 'ignored', + { type: 'unsupported', id: 'ignored', content: 'Ignored' }, + ]), + ]); + expect( + state.messages.map(({ role, content }) => ({ role, content })) + ).toEqual([ + { role: 'user', content: 'Hi!' }, + { role: 'assistant', content: 'Hello' }, + { role: 'system', content: 'Rules' }, + { role: 'tool', content: 'Wire result' }, + { role: 'assistant', content: '' }, + ]); + expect(state.messages[3]).toMatchObject({ name: 'work', toolCallId: 'c' }); + }); + + it('matches wire results by exact call ID independent of message order and retains pending calls', () => { + const state = projectHistory(initial(), [ + checkpoint([ + result('r', 'c1', '{"saved":true}'), + ai('a', '', [call('c10'), call('c1')]), + ]), + ]); + expect(state.toolCalls).toMatchObject([ + { id: 'c10', status: 'pending' }, + { id: 'c1', status: 'complete', result: '{"saved":true}' }, + ]); + expect(state.messages[0]).toMatchObject({ + role: 'tool', + toolCallId: 'c1', + content: '{"saved":true}', + }); + }); + + it('shares unchanged tools by ID across reorder and removes absent tool metadata and results', () => { + const before = projectHistory(initial(), [ + checkpoint([ai('a', '', [call('c1'), call('c2')])]), + ]); + const reordered = projectHistory(before, [ + checkpoint([ai('a', '', [call('c2'), call('c1')])]), + ]); + expect(reordered.toolCalls).toHaveLength(2); + expect(reordered.toolCalls[0]).toBe(before.toolCalls[1]); + expect(reordered.toolCalls[1]).toBe(before.toolCalls[0]); + for (const raw of [ai('a'), ai('a', '', [])]) { + const removed = projectHistory(reordered, [checkpoint([raw])]); + expect(removed.toolCalls).toEqual([]); + expect(removed.messages[0].toolCallIds ?? []).toEqual([]); + } + }); + + it('owns nested arguments and readonly message data without freezing or mutating the caller', () => { + const args = { nested: { values: ['original'] } }; + const history = [checkpoint([ai('a', '', [call('c', 'work', args)])])]; + const before = structuredClone(history); + const state = projectHistory(initial(), history); + expect(state.messages).toHaveLength(1); + expect(state.toolCalls).toHaveLength(1); + expect(history).toEqual(before); + expect(Object.isFrozen(args.nested.values)).toBe(false); + expect(Object.isFrozen(state)).toBe(true); + expect(Object.isFrozen(state.messages)).toBe(true); + expect(Object.isFrozen(state.messages[0])).toBe(true); + expect(Object.isFrozen(state.messages[0].toolCallIds)).toBe(true); + expect(Object.isFrozen(state.toolCalls[0])).toBe(true); + expect( + Object.isFrozen((state.toolCalls[0].args as typeof args).nested.values) + ).toBe(true); + args.nested.values[0] = 'changed'; + expect(state.toolCalls[0].args).toEqual({ + nested: { values: ['original'] }, + }); + }); + + it('does not expose chunk arguments as finalized historical calls', () => { + const state = projectHistory(initial(), [ + checkpoint([ + { + ...ai('a', 'Partial', [call('fragment', 'work', '{')]), + type: 'AIMessageChunk', + }, + ]), + ]); + expect(state.messages).toHaveLength(1); + expect(state.messages[0].content).toBe('Partial'); + expect(state.toolCalls).toEqual([]); + expect(state.messages[0].toolCallIds).toBeUndefined(); + }); + + it.each(['values', 'tasks'] as const)( + 'pauses only the last assistant in the latest turn from explicit %s evidence', + (source) => { + const messages = [ + human('old-u'), + ai('old-a'), + human('new-u'), + ai('step'), + ai('latest'), + ]; + const task = { + id: 't', + name: 'ask', + error: null, + interrupts: [{ value: 'Continue?' }], + checkpoint: null, + state: null, + result: null, + }; + const history = [ + checkpoint( + messages, + source === 'values' + ? { values: { messages, __interrupt__: ['Continue?'] } } + : { tasks: [task] } + ), + ]; + const state = projectHistory(initial(), history); + expect(state.messages.map((message) => message.delivery)).toEqual( + messages.map((message) => + message.id === 'latest' + ? completeDelivery('latest', 'paused') + : staticDelivery(message.id) + ) + ); + expect(projectHistory(state, structuredClone(history))).toBe(state); + } + ); + + it('never marks an older assistant paused when the latest user has no assistant response', () => { + const messages = [human('old-u'), ai('old-a'), human('new-u')]; + const state = projectHistory(initial(), [ + checkpoint(messages, { + values: { messages, __interrupt__: ['Waiting'] }, + }), + ]); + expect(state.messages.map((message) => message.delivery)).toEqual( + messages.map((message) => staticDelivery(message.id)) + ); + }); + + it('treats next-node work without explicit interrupts as static history, not interruption', () => { + const state = projectHistory(initial(), [ + checkpoint([human('u'), ai('a')], { next: ['tools'] }), + ]); + expect(state.messages).toHaveLength(2); + expect(state.messages[1].delivery).toEqual(staticDelivery('a')); + }); + + it('loads an ordinary wire task that omits interrupt evidence', () => { + const history = [ + checkpoint([human('u'), ai('a')], { + next: ['tools'], + tasks: [ + { id: 'task', name: 'tools' }, + ] as unknown as ThreadState['tasks'], + }), + ]; + expect(() => projectHistory(initial(), history)).not.toThrow(); + expect(projectHistory(initial(), history).messages[1].delivery).toEqual( + staticDelivery('a') + ); + }); + + it('exposes only registered pending calls to typed catalogs and retains all ToolMessages', () => { + const history = [ + checkpoint([ + ai('a', '', [ + call('pending'), + call('settled'), + call('remote', 'server'), + ]), + result('wire', 'settled', '{"value":42}'), + ]), + ]; + const broad = projectHistory(initial(), history); + expect(broad.toolCalls.map((entry) => entry.id)).toEqual([ + 'pending', + 'settled', + 'remote', + ]); + const typed = projectHistory(broad, history, { + registeredTools: new Set(['work']), + }); + expect(typed.toolCalls).toMatchObject([ + { id: 'pending', status: 'pending' }, + ]); + expect(typed.messages).toBe(broad.messages); + expect(typed.messages[1].content).toBe('{"value":42}'); + expect( + projectHistory(typed, history, { registeredTools: new Set() }).toolCalls + ).toEqual([]); + }); + + it('does not revive a locally resolved typed result from an authoritative remote wire string', () => { + const first = projectHistory(initial(), [ + checkpoint([ai('a', '', [call('c')])]), + ]); + const local = reduceMessages(first, { + type: 'tool', + toolCall: { + id: 'c', + name: 'work', + args: { id: 'c' }, + status: 'complete', + result: { authored: true }, + }, + }); + const history = [ + checkpoint([ + ai('a', '', [call('c')]), + result('wire', 'c', 'Server authored text'), + ]), + ]; + const typed = projectHistory(local, history, { + registeredTools: new Set(['work']), + }); + expect(typed.toolCalls).toEqual([]); + expect(typed.messages[1].content).toBe('Server authored text'); + expect(projectHistory(local, history).toolCalls).toMatchObject([ + { id: 'c', status: 'complete', result: 'Server authored text' }, + ]); + }); +}); diff --git a/libs/langgraph/src/runtime/history-projection.ts b/libs/langgraph/src/runtime/history-projection.ts new file mode 100644 index 000000000..003d2b692 --- /dev/null +++ b/libs/langgraph/src/runtime/history-projection.ts @@ -0,0 +1,179 @@ +import type { ThreadState } from '@langchain/langgraph-sdk'; +import { + completeDelivery, + staticDelivery, + type Message, + type PlainValue, + type ToolCall, +} from '@threadplane/core'; +import type { MessageState } from './message-reducer'; +import { + ownMessage, + ownToolCall, + sameMessage, + sameToolCall, +} from './ownership'; +import { hasPause, record, roleOf, textContent } from './wire-message'; + +export interface HistoryProjectionOptions { + /** Omit for broad wire observation. A supplied catalog exposes only its + * pending calls: a persisted ToolMessage string is not an authored result. */ + readonly registeredTools?: ReadonlySet; +} + +function sameEntries(left: readonly unknown[], right: readonly unknown[]) { + return ( + left.length === right.length && + left.every((value, index) => value === right[index]) + ); +} + +/** Replace the transcript from the latest checkpoint, independently of any run. + * Explicit duplicate message IDs use their last occurrence and last position. + * ID-less messages use checkpoint indices, avoiding every explicit ID first; + * their identity is stable on equal reads, not guaranteed across reorder. + * Nothing executes here, and prior locally settled results are not evidence for + * the types of newly loaded wire results. Streaming bookkeeping is discarded. */ +export function projectHistory( + previous: MessageState, + history: readonly ThreadState[], + options: HistoryProjectionOptions = {} +): MessageState { + const latest = history[0]; + const values = record(latest?.values); + const rawMessages: unknown[] = Array.isArray(values?.['messages']) + ? values['messages'] + : []; + const raw = rawMessages.map(record); + const explicitIds = new Map(); + raw.forEach((message, index) => { + if (typeof message?.['id'] === 'string') + explicitIds.set(message['id'], index); + }); + const reservedIds = new Set(explicitIds.keys()); + const previousMessages = new Map( + previous.messages.map((message) => [message.id, message]) + ); + const previousTools = new Map( + previous.toolCalls.map((tool) => [tool.id, tool]) + ); + const projectedMessages: Message[] = []; + const calls = new Map(); + + raw.forEach((message, index) => { + if (!message) return; + const role = roleOf(message); + if (!role) return; + let id: string; + if (typeof message['id'] === 'string') { + id = message['id']; + if (explicitIds.get(id) !== index) return; + } else { + const base = `history-message-${index}`; + id = base; + let suffix = 0; + while (reservedIds.has(id)) id = `${base}-${++suffix}`; + reservedIds.add(id); + } + const finalizedCalls = + role === 'assistant' && + message['type'] !== 'AIMessageChunk' && + Array.isArray(message['tool_calls']); + const messageCalls = finalizedCalls + ? (message['tool_calls'] as unknown[]) + .map(record) + .filter((call): call is Record => !!call) + : []; + projectedMessages.push({ + id, + role, + content: textContent(message['content']), + delivery: staticDelivery(id), + ...(typeof message['name'] === 'string' ? { name: message['name'] } : {}), + ...(typeof message['tool_call_id'] === 'string' + ? { toolCallId: message['tool_call_id'] } + : {}), + ...(finalizedCalls + ? { + toolCallIds: messageCalls.flatMap((call) => + typeof call['id'] === 'string' ? [call['id']] : [] + ), + } + : {}), + }); + for (const call of messageCalls) { + if (typeof call['id'] !== 'string' || typeof call['name'] !== 'string') + continue; + calls.set(call['id'], { + id: call['id'], + name: call['name'], + args: call['args'] as PlainValue, + status: 'pending', + }); + } + }); + + if ( + hasPause(values) || + latest?.tasks?.some((task) => (task.interrupts?.length ?? 0) > 0) + ) { + // A pause belongs to the current turn. Do not reach past its last user to + // borrow an older assistant when the latest request has no response yet. + for (let index = projectedMessages.length - 1; index >= 0; index -= 1) { + const message = projectedMessages[index]; + if (message.role === 'user') break; + if (message.role === 'assistant') { + projectedMessages[index] = { + ...message, + delivery: completeDelivery(message.id, 'paused'), + }; + break; + } + } + } + + const results = new Map(); + for (const message of projectedMessages) + if (message.role === 'tool' && message.toolCallId !== undefined) + results.set(message.toolCallId, message.content); + const projectedTools: ToolCall[] = []; + for (const call of calls.values()) { + if ( + options.registeredTools && + (!options.registeredTools.has(call.name) || results.has(call.id)) + ) + continue; + const projected: ToolCall = results.has(call.id) + ? { ...call, status: 'complete', result: results.get(call.id) } + : call; + const prior = previousTools.get(call.id); + projectedTools.push( + ownToolCall(prior && sameToolCall(prior, projected) ? prior : projected) + ); + } + const ownedMessages = projectedMessages.map((message) => { + const prior = previousMessages.get(message.id); + return ownMessage(prior && sameMessage(prior, message) ? prior : message); + }); + const messages = sameEntries(ownedMessages, previous.messages) + ? previous.messages + : Object.freeze(ownedMessages); + const toolCalls = sameEntries(projectedTools, previous.toolCalls) + ? previous.toolCalls + : Object.freeze(projectedTools); + if ( + messages === previous.messages && + toolCalls === previous.toolCalls && + previous.canonical.length === 0 && + previous.aliases.length === 0 + ) + return previous; + return Object.freeze({ + messages, + toolCalls, + canonical: previous.canonical.length + ? Object.freeze([]) + : previous.canonical, + aliases: previous.aliases.length ? Object.freeze([]) : previous.aliases, + }); +} diff --git a/libs/langgraph/src/runtime/history.spec.ts b/libs/langgraph/src/runtime/history.spec.ts new file mode 100644 index 000000000..02cd150e8 --- /dev/null +++ b/libs/langgraph/src/runtime/history.spec.ts @@ -0,0 +1,672 @@ +import type { ThreadState } from '@langchain/langgraph-sdk'; +import type { AgentSession } from '@threadplane/core'; +import type { ToolExecutionStore } from '@threadplane/core/tools'; +import { describe, expect, it, vi } from 'vitest'; +import { createSession } from './create-session'; +import type { AgentTransport, StreamEvent } from './transport.types'; +import { deferred, type Deferred } from './testing/deferred'; +import { controlledTransport } from './testing/controlled-transport'; + +type Loadable = AgentSession & { + load?(options?: { signal?: AbortSignal }): Promise; +}; +function load( + session: Loadable, + options?: { signal?: AbortSignal } +): Promise { + expect(session.load).toBeTypeOf('function'); + return session.load?.(options) as Promise; +} +function history( + content = 'Persisted', + messages: unknown[] = [ + { type: 'human', id: 'saved-user', content: 'Question' }, + { type: 'ai', id: 'saved-answer', content }, + ] +): ThreadState[] { + return [ + { + values: { messages } as ThreadState['values'], + next: [], + tasks: [], + checkpoint: { + thread_id: 'thread', + checkpoint_id: 'checkpoint', + checkpoint_ns: '', + checkpoint_map: {}, + }, + metadata: null, + created_at: null, + parent_checkpoint: null, + }, + ]; +} +const toolMessage = { + type: 'ai', + id: 'tool-step', + content: '', + tool_calls: [{ id: 'call', name: 'work', args: {} }], +}; +const answer: StreamEvent = { + type: 'values', + data: { messages: [{ type: 'ai', id: 'answer', content: 'Done' }] }, +}; +function fixture() { + const reads: { result: Deferred; signal: AbortSignal }[] = []; + const started = Array.from({ length: 6 }, () => deferred()); + const getHistory = vi.fn>( + (_thread, signal) => { + const result = deferred(); + reads.push({ result, signal }); + started[reads.length - 1].resolve(); + return result.promise; + } + ); + const stream = vi.fn(async function* () { + yield answer; + }); + const updateState = vi.fn>( + async () => undefined + ); + const session = createSession({ + assistantId: 'agent', + threadId: 'thread', + transport: { stream, getHistory, updateState }, + }); + return { + session, + reads, + getHistory, + stream, + updateState, + started: (index = 0) => started[index].promise, + }; +} + +describe('explicit owned history loading', () => { + it('exposes only the optional transport capability, with inert construction and observation', async () => { + const f = fixture(); + const notify = vi.fn(); + const before = f.session.getSnapshot(); + const off = f.session.subscribe(notify); + expect(f.session.getSnapshot()).toBe(before); + expect(f.getHistory).not.toHaveBeenCalled(); + expect(f.stream).not.toHaveBeenCalled(); + expect(f.updateState).not.toHaveBeenCalled(); + expect(notify).not.toHaveBeenCalled(); + expect((f.session as Loadable).load).toBeTypeOf('function'); + const unsupported = createSession({ + assistantId: 'a', + threadId: 't', + transport: { stream: f.stream }, + }); + expect('load' in unsupported).toBe(false); + off(); + await f.session.dispose(); + await unsupported.dispose(); + }); + + it('reads once with an owned signal, atomically replaces on success, shares equal reads and clears empty history', async () => { + const f = fixture(); + const external = new AbortController(); + const notify = vi.fn(); + f.session.subscribe(notify); + const before = f.session.getSnapshot(); + const pending = load(f.session, { signal: external.signal }); + await f.started(); + expect(f.getHistory).toHaveBeenCalledWith('thread', f.reads[0].signal); + expect(f.reads[0].signal).not.toBe(external.signal); + expect(f.session.getSnapshot()).toBe(before); + expect(notify).not.toHaveBeenCalled(); + f.reads[0].result.resolve(history()); + await pending; + const loaded = f.session.getSnapshot(); + expect(loaded.messages.map((message) => message.content)).toEqual([ + 'Question', + 'Persisted', + ]); + expect(loaded.status).toBe('idle'); + expect(notify).toHaveBeenCalledTimes(1); + const equal = load(f.session); + await f.started(1); + f.reads[1].result.resolve(history()); + await equal; + expect(f.session.getSnapshot()).toBe(loaded); + expect(notify).toHaveBeenCalledTimes(1); + const empty = load(f.session); + await f.started(2); + f.reads[2].result.resolve([]); + await empty; + expect(f.session.getSnapshot().messages).toEqual([]); + expect(notify).toHaveBeenCalledTimes(2); + expect(f.stream).not.toHaveBeenCalled(); + expect(f.updateState).not.toHaveBeenCalled(); + await f.session.dispose(); + }); + + it('keeps loaded snapshot on failure, rejects sanitized diagnostics and permits explicit retry', async () => { + const f = fixture(); + const first = load(f.session); + await f.started(); + f.reads[0].result.resolve(history()); + await first; + const before = f.session.getSnapshot(); + const notify = vi.fn(); + f.session.subscribe(notify); + const failed = load(f.session); + const rejected = expect(failed).rejects.toMatchObject({ + name: 'LangGraphRequestError', + message: 'The LangGraph request failed.', + }); + await f.started(1); + f.reads[1].result.reject({ + get message() { + throw new Error('Never inspect secrets'); + }, + body: 'secret', + cause: 'secret', + }); + await rejected; + expect(f.reads[1].signal.aborted).toBe(true); + expect(f.session.getSnapshot()).toBe(before); + expect(notify).not.toHaveBeenCalled(); + const retry = load(f.session); + await f.started(2); + f.reads[2].result.resolve(history('Retried')); + await retry; + expect(f.session.getSnapshot().messages[1].content).toBe('Retried'); + await f.session.dispose(); + }); + + it('clears a prior non-recovery error only after successful history projection', async () => { + const f = fixture(); + f.stream.mockImplementation(async function* () { + yield { type: 'error', data: { status: 401, message: 'Login' } }; + }); + await expect(f.session.submit('Fail')).resolves.toBe('error'); + const before = f.session.getSnapshot(); + expect(before.error?.recovery).toBe('none'); + const pending = load(f.session); + await f.started(); + expect(f.session.getSnapshot()).toBe(before); + f.reads[0].result.resolve(history()); + await pending; + expect(f.session.getSnapshot()).toMatchObject({ + status: 'idle', + error: undefined, + }); + await f.session.dispose(); + }); + + it.each(['resolve', 'reject'] as const)( + 'supersedes an ignored-abort read and ignores its late %s', + async (completion) => { + const f = fixture(); + const first = load(f.session); + await f.started(); + const second = load(f.session); + await first; + await f.started(1); + expect(f.reads[0].signal.aborted).toBe(true); + f.reads[1].result.resolve(history('Newest')); + await second; + const committed = f.session.getSnapshot(); + if (completion === 'resolve') f.reads[0].result.resolve(history('Stale')); + else f.reads[0].result.reject(new Error('secret stale rejection')); + await f.reads[0].result.promise.catch(() => undefined); + await Promise.resolve(); + expect(f.session.getSnapshot()).toBe(committed); + await f.session.dispose(); + } + ); + + it.each( + (['stop', 'dispose', 'submit', 'abort'] as const).flatMap((command) => + (['resolve', 'reject'] as const).map((completion) => ({ + command, + completion, + })) + ) + )( + 'settles promptly on $command and ignores late $completion', + async ({ command, completion }) => { + const f = fixture(); + const controller = new AbortController(); + const pending = load(f.session, { signal: controller.signal }); + await f.started(); + if (command === 'abort') controller.abort(); + else if (command === 'submit') await f.session.submit('New request'); + else await f.session[command](); + await pending; + expect(f.reads[0].signal.aborted).toBe(true); + const current = f.session.getSnapshot(); + if (completion === 'resolve') f.reads[0].result.resolve(history('Stale')); + else f.reads[0].result.reject(new Error('Late failure with raw secrets')); + await f.reads[0].result.promise.catch(() => undefined); + await Promise.resolve(); + expect(f.session.getSnapshot()).toBe(current); + await f.session.dispose(); + } + ); + + it('does no I/O for pre-aborted or disposed loads and detaches caller abort listeners on success', async () => { + const f = fixture(); + const aborted = new AbortController(); + aborted.abort(); + await load(f.session, { signal: aborted.signal }); + expect(f.getHistory).not.toHaveBeenCalled(); + const external = new AbortController(); + const remove = vi.spyOn(external.signal, 'removeEventListener'); + const pending = load(f.session, { signal: external.signal }); + await f.started(); + f.reads[0].result.resolve(history()); + await pending; + expect(remove).toHaveBeenCalledWith('abort', expect.any(Function)); + const current = f.session.getSnapshot(); + external.abort(); + expect(f.session.getSnapshot()).toBe(current); + await f.session.dispose(); + await load(f.session); + expect(f.getHistory).toHaveBeenCalledTimes(1); + }); + + it('rejects active execution and recovery admission before requesting history', async () => { + const f = fixture(); + const running = controlledTransport(); + f.stream.mockReturnValue(running.stream); + const submit = f.session.submit('Running'); + await expect(load(f.session)).rejects.toThrow(); + expect(f.getHistory).not.toHaveBeenCalled(); + await f.session.stop(); + await submit; + f.stream.mockImplementation(async function* () { + yield { type: 'error', data: { status: 500 } }; + }); + await f.session.submit('Uncertain'); + expect(f.session.getSnapshot().error?.recovery).toBe('check'); + await expect(load(f.session)).rejects.toThrow(); + expect(f.getHistory).not.toHaveBeenCalled(); + await f.session.dispose(); + }); + + it('isolates concurrent reads in two sessions', async () => { + const a = fixture(); + const b = fixture(); + const first = load(a.session); + const second = load(b.session); + await a.started(); + await b.started(); + await a.session.stop(); + await first; + expect(b.reads[0].signal.aborted).toBe(false); + b.reads[0].result.resolve(history('B')); + await second; + a.reads[0].result.resolve(history('A')); + await a.reads[0].result.promise; + expect(a.session.getSnapshot().messages).toEqual([]); + expect(b.session.getSnapshot().messages[1].content).toBe('B'); + await a.session.dispose(); + await b.session.dispose(); + }); + + it('commits a new read owner before aborting the old resource and prevents stale reentrant I/O', async () => { + const f = fixture(); + const first = load(f.session); + await f.started(); + f.reads[0].signal.addEventListener('abort', () => { + void f.session.stop(); + }); + const superseding = load(f.session); + await first; + await superseding; + expect(f.getHistory).toHaveBeenCalledTimes(1); + f.reads[0].result.resolve(history('Old')); + await f.reads[0].result.promise; + expect(f.session.getSnapshot().messages).toEqual([]); + await f.session.dispose(); + }); + + it('allows a reentrant transport stop without publishing its eventual history', async () => { + const f = fixture(); + const result = deferred(); + f.getHistory.mockImplementation(() => { + void f.session.stop(); + return result.promise; + }); + await load(f.session); + result.resolve(history('Never')); + await result.promise; + await Promise.resolve(); + expect(f.session.getSnapshot().messages).toEqual([]); + await f.session.dispose(); + }); + + it('lets an old abort callback start a newer read without dispatching the superseded middle read', async () => { + const f = fixture(); + const first = load(f.session); + await f.started(); + let newest: Promise | undefined; + f.reads[0].signal.addEventListener('abort', () => { + newest = load(f.session); + }); + const middle = load(f.session); + await first; + await middle; + await f.started(1); + expect(f.getHistory).toHaveBeenCalledTimes(2); + f.reads[1].result.resolve(history('Newest')); + await newest; + f.reads[0].result.reject(new Error('Old failure')); + await f.reads[0].result.promise.catch(() => undefined); + expect(f.session.getSnapshot().messages[1].content).toBe('Newest'); + await f.session.dispose(); + }); + + it('sanitizes a projection failure and preserves the exact previous snapshot', async () => { + const f = fixture(); + const before = f.session.getSnapshot(); + const pending = load(f.session); + const rejected = expect(pending).rejects.toMatchObject({ + name: 'LangGraphRequestError', + message: 'The LangGraph request failed.', + }); + await f.started(); + const checkpoint = history()[0]; + Object.defineProperty(checkpoint, 'values', { + get() { + throw new Error('Private checkpoint contents'); + }, + }); + f.reads[0].result.resolve([checkpoint]); + await rejected; + expect(f.session.getSnapshot()).toBe(before); + await f.session.dispose(); + }); + + it('aborts a failed read only after detaching it so its abort callback can own a replacement', async () => { + const f = fixture(); + const failed = load(f.session); + const rejection = expect(failed).rejects.toMatchObject({ + name: 'LangGraphRequestError', + }); + await f.started(); + let replacement: Promise | undefined; + f.reads[0].signal.addEventListener('abort', () => { + replacement = load(f.session); + }); + f.reads[0].result.reject(new Error('Failed read')); + await rejection; + expect(replacement).toBeDefined(); + await f.started(1); + expect(f.reads[1].signal.aborted).toBe(false); + f.reads[1].result.resolve(history('Replacement')); + await replacement; + expect(f.session.getSnapshot().messages[1].content).toBe('Replacement'); + await f.session.dispose(); + }); + + it('rechecks read ownership after projection invokes a raw getter', async () => { + const f = fixture(); + const pending = load(f.session); + await f.started(); + const checkpoint = history()[0]; + Object.defineProperty(checkpoint, 'values', { + get() { + void f.session.submit('New owner'); + return { messages: [{ type: 'ai', id: 'stale', content: 'Never' }] }; + }, + }); + f.reads[0].result.resolve([checkpoint]); + await pending; + expect( + f.session.getSnapshot().messages.some((message) => message.id === 'stale') + ).toBe(false); + expect( + f.session + .getSnapshot() + .messages.some((message) => message.content === 'New owner') + ).toBe(true); + await f.session.dispose(); + }); + + it('allows a publication listener to submit after loading without stale owner cleanup cancelling it', async () => { + const f = fixture(); + let run: Promise | undefined; + const off = f.session.subscribe(() => { + if ( + !run && + f.session + .getSnapshot() + .messages.some((message) => message.id === 'saved-answer') + ) + run = f.session.submit('Next'); + }); + const pending = load(f.session); + await f.started(); + f.reads[0].result.resolve(history()); + await pending; + await run; + expect(f.stream).toHaveBeenCalledTimes(1); + expect( + f.session + .getSnapshot() + .messages.some((message) => message.content === 'Next') + ).toBe(true); + off(); + await f.session.dispose(); + }); + + it('binds the captured history method to its transport receiver', async () => { + const stream = vi.fn(async function* () { + yield answer; + }); + const transport = { + stream, + marker: history(), + getHistory() { + return Promise.resolve(this.marker); + }, + }; + const session = createSession({ + assistantId: 'a', + threadId: 't', + transport, + }); + transport.getHistory = () => Promise.reject(new Error('Replaced method')); + await load(session); + expect(session.getSnapshot().messages[1].content).toBe('Persisted'); + await session.dispose(); + }); +}); + +describe('history admission around tool ownership', () => { + it('never runs registered historical pending tools during load or later baseline replay', async () => { + const handler = vi.fn(() => 'Never'); + const store: ToolExecutionStore = { + claim: vi.fn(async () => 'claimed' as const), + record: vi.fn(async () => undefined), + }; + const getHistory = vi.fn(async () => history('', [toolMessage])); + const stream = vi.fn(async function* () { + yield { + type: 'values', + data: { + messages: [ + toolMessage, + { type: 'ai', id: 'fresh', content: 'Fresh' }, + ], + }, + }; + }); + const updateState = vi.fn(async () => undefined); + const session = createSession({ + assistantId: 'a', + threadId: 't', + transport: { stream, getHistory, updateState }, + executionStore: store, + tools: { work: { description: 'Work', handler } }, + }); + await load(session); + expect(session.getSnapshot().toolCalls).toMatchObject([ + { id: 'call', status: 'pending' }, + ]); + expect(stream).not.toHaveBeenCalled(); + expect(handler).not.toHaveBeenCalled(); + expect(store.claim).not.toHaveBeenCalled(); + expect(updateState).not.toHaveBeenCalled(); + await session.submit('New turn'); + expect(handler).not.toHaveBeenCalled(); + expect(store.claim).not.toHaveBeenCalled(); + expect(store.record).not.toHaveBeenCalled(); + expect(updateState).not.toHaveBeenCalled(); + await session.dispose(); + }); + + it('keeps execution dedupe after load while dropping authored-result provenance on later wire replay', async () => { + const handler = vi.fn(() => ({ authored: true })); + const wire = { + type: 'tool', + id: 'wire', + tool_call_id: 'call', + content: 'Wire string cannot be authored object', + }; + const getHistory = vi.fn(async () => history('', [toolMessage, wire])); + let streams = 0; + const stream = vi.fn(async function* () { + yield { + type: 'values', + data: { + messages: + ++streams === 1 + ? [toolMessage] + : [ + toolMessage, + wire, + { type: 'ai', id: 'fresh', content: 'Fresh' }, + ], + }, + }; + }); + const session = createSession({ + assistantId: 'a', + threadId: 't', + transport: { stream, getHistory, updateState: async () => undefined }, + tools: { work: { description: 'Work', followUp: false, handler } }, + }); + await session.submit('First'); + expect(session.getSnapshot().toolCalls).toMatchObject([ + { status: 'complete', result: { authored: true } }, + ]); + await load(session); + expect(session.getSnapshot().toolCalls).toEqual([]); + const seen: unknown[] = []; + const off = session.subscribe(() => + seen.push(...session.getSnapshot().toolCalls) + ); + await session.submit('Next'); + expect(seen).toEqual([]); + expect(session.getSnapshot().toolCalls).toEqual([]); + expect(handler).toHaveBeenCalledTimes(1); + // An empty replacement also cannot authorize an already executed call ID. + getHistory.mockResolvedValue([]); + await load(session); + stream.mockImplementation(async function* () { + yield { type: 'values', data: { messages: [toolMessage] } }; + }); + await session.submit('Replay old id'); + expect(handler).toHaveBeenCalledTimes(1); + off(); + await session.dispose(); + }); + + it('rejects staged results after a failed write before any history request', async () => { + const getHistory = vi.fn(async () => history()); + const session = createSession({ + assistantId: 'a', + threadId: 't', + transport: { + getHistory, + stream: async function* () { + yield { type: 'values', data: { messages: [toolMessage] } }; + }, + updateState: async () => { + throw new Error('Failed persistence'); + }, + }, + tools: { + work: { description: 'Work', followUp: false, handler: () => 'Done' }, + }, + }); + await expect(session.submit('Work')).resolves.toBe('error'); + expect(session.getSnapshot().error?.recovery).toBe('none'); + await expect(load(session)).rejects.toThrow(); + expect(getHistory).not.toHaveBeenCalled(); + await session.dispose(); + }); + + it.each(['claim', 'record', 'write'] as const)( + 'blocks replacement until late %s and persistence settle after stop', + async (phase) => { + const claim = deferred<'claimed'>(); + const recorded = deferred(); + const written = deferred(); + const claimStarted = deferred(); + const recordStarted = deferred(); + const writeStarted = deferred(); + const getHistory = vi.fn(async () => history('After settlement')); + const store: ToolExecutionStore = { + claim: vi.fn(() => { + claimStarted.resolve(); + return phase === 'claim' + ? claim.promise + : Promise.resolve('claimed' as const); + }), + record: vi.fn(() => { + recordStarted.resolve(); + return phase === 'record' ? recorded.promise : Promise.resolve(); + }), + }; + const session = createSession({ + assistantId: 'a', + threadId: 't', + transport: { + getHistory, + stream: async function* () { + yield { type: 'values', data: { messages: [toolMessage] } }; + }, + updateState: () => { + writeStarted.resolve(); + return written.promise; + }, + }, + executionStore: store, + tools: { + work: { description: 'Work', followUp: false, handler: () => 'Done' }, + }, + }); + const submitted = session.submit('Work'); + await (phase === 'claim' + ? claimStarted.promise + : phase === 'record' + ? recordStarted.promise + : writeStarted.promise); + await session.stop(); + await expect(submitted).resolves.toBe('aborted'); + await expect(load(session)).rejects.toThrow(); + expect(getHistory).not.toHaveBeenCalled(); + claim.resolve('claimed'); + recorded.resolve(); + await writeStarted.promise; + await expect(load(session)).rejects.toThrow(); + expect(getHistory).not.toHaveBeenCalled(); + written.resolve(); + await written.promise; + await Promise.resolve(); + await load(session); + expect(session.getSnapshot().messages[1].content).toBe( + 'After settlement' + ); + await session.dispose(); + } + ); +}); diff --git a/libs/langgraph/src/runtime/history.type-test.ts b/libs/langgraph/src/runtime/history.type-test.ts new file mode 100644 index 000000000..879cec47b --- /dev/null +++ b/libs/langgraph/src/runtime/history.type-test.ts @@ -0,0 +1,36 @@ +import type { AgentSession, ToolContract } from '@threadplane/core'; +import { createSession, type LangGraphSession } from './create-session'; + +const session = createSession({ + assistantId: 'a', + threadId: 't', + tools: { + work: { + description: 'Work', + handler: (args: { amount: number }) => ({ doubled: args.amount * 2 }), + }, + }, +}); +const observer: AgentSession = session; +const adapter: LangGraphSession = session; +const broad: AgentSession> = adapter; +const loaded: Promise | undefined = session.load?.({ + signal: new AbortController().signal, +}); +session.load?.(); +// @ts-expect-error load accepts only a standard abort signal +session.load?.({ signal: 'not a signal' }); +// @ts-expect-error load does not introduce submit input +session.load?.('message'); +// @ts-expect-error core observer does not gain adapter-specific load +observer.load?.(); +for (const call of session.getSnapshot().toolCalls) { + if (call.status === 'complete') { + const result: number = call.result.doubled; + const input: number = call.args.amount; + // @ts-expect-error authored result inference survives adapter extension + const incorrect: string = call.result.doubled; + void [result, input, incorrect]; + } +} +void [observer, adapter, broad, loaded]; diff --git a/libs/langgraph/src/runtime/stream-projection.ts b/libs/langgraph/src/runtime/stream-projection.ts index 206e7cf14..5de14d72b 100644 --- a/libs/langgraph/src/runtime/stream-projection.ts +++ b/libs/langgraph/src/runtime/stream-projection.ts @@ -12,6 +12,9 @@ import { } from './message-reducer'; import type { StreamEvent } from './transport.types'; import { ownMessage, ownToolCall } from './ownership'; +import { hasPause, record, roleOf, textContent } from './wire-message'; + +export { hasPause, record } from './wire-message'; type CanonicalMessage = Extract; @@ -28,54 +31,6 @@ export interface StreamProjection { readonly toolCallIds?: readonly string[]; } -export function record(value: unknown): Record | undefined { - return typeof value === 'object' && value !== null && !Array.isArray(value) - ? (value as Record) - : undefined; -} - -function roleOf(raw: Record): Message['role'] | undefined { - switch (raw['type'] ?? raw['role']) { - case 'human': - case 'HumanMessage': - case 'user': - return 'user'; - case 'ai': - case 'AIMessage': - case 'AIMessageChunk': - case 'assistant': - return 'assistant'; - case 'system': - case 'SystemMessage': - return 'system'; - case 'tool': - case 'ToolMessage': - return 'tool'; - default: - return undefined; - } -} - -function textContent(value: unknown): string { - if (typeof value === 'string') return value; - if (!Array.isArray(value)) return ''; - return value - .flatMap((block) => { - const content = record(block); - return content?.['type'] === 'text' && typeof content['text'] === 'string' - ? [content['text']] - : []; - }) - .join(''); -} - -export function hasPause(value: unknown): boolean { - const data = record(value); - return ( - Array.isArray(data?.['__interrupt__']) && data['__interrupt__'].length > 0 - ); -} - /** Pure projection of text and finalized tool data from root stream events. * Values are interim while the stream is open. EOF or a distinct next assistant * confirms a terminal candidate. Same-ID chunks invalidate only that message's diff --git a/libs/langgraph/src/runtime/testing/binding-fixture.ts b/libs/langgraph/src/runtime/testing/binding-fixture.ts index 048b067ed..97f4f9226 100644 --- a/libs/langgraph/src/runtime/testing/binding-fixture.ts +++ b/libs/langgraph/src/runtime/testing/binding-fixture.ts @@ -1,4 +1,5 @@ import type { AgentSession, AgentSnapshot } from '@threadplane/core'; +import type { ThreadState } from '@langchain/langgraph-sdk'; import { createSession } from '../create-session'; import type { AgentTransport, StreamEvent } from '../transport.types'; import { controlledTransport } from './controlled-transport'; @@ -18,6 +19,19 @@ export function bindingFixture() { const toolResult = deferred<{ temperature: number }>(); let handlerCalls = 0; 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', + }], + }; const stream: AgentTransport['stream'] = (_a, _t, _p, signal) => { const controlled = controlledTransport({ signal }); streams.push(controlled); @@ -27,7 +41,7 @@ export function bindingFixture() { const runtime = createSession({ assistantId: 'binding-agent', threadId: 'binding-thread', - transport: { stream }, + transport: { stream, getHistory: async () => { history.reads++; return history.value; } }, tools: { weather: { description: 'Weather', @@ -75,6 +89,11 @@ export function bindingFixture() { this.stopCalls++; return runtime.stop(); } + load(options?: { signal?: AbortSignal }) { + void this.subscriptions; + if (!runtime.load) throw new Error('Fixture requires history loading'); + return runtime.load(options); + } dispose() { this.disposeCalls++; return runtime.dispose(); @@ -84,6 +103,7 @@ export function bindingFixture() { const session = new BorrowedSession(); return { session, + history, streams, started: (index = 0) => starts[index].promise, entered: entered.promise, diff --git a/libs/langgraph/src/runtime/transport.integration.spec.ts b/libs/langgraph/src/runtime/transport.integration.spec.ts index 109873f09..0cc9fe49e 100644 --- a/libs/langgraph/src/runtime/transport.integration.spec.ts +++ b/libs/langgraph/src/runtime/transport.integration.spec.ts @@ -70,6 +70,89 @@ describe('neutral real SDK transport', () => { vi.unstubAllGlobals(); }); + it('loads decoded history through the real SDK endpoint without issuing runs, writes or tools', async () => { + const request = vi.fn( + async () => + new Response( + JSON.stringify([ + { + values: { + messages: [ + { type: 'human', id: 'persisted-user', content: 'Weather?' }, + toolCall, + ], + }, + next: ['tools'], + tasks: [], + checkpoint: { + thread_id: 'thread-1', + checkpoint_id: 'persisted', + checkpoint_ns: '', + checkpoint_map: {}, + }, + metadata: null, + created_at: null, + parent_checkpoint: null, + }, + ]), + { headers: { 'content-type': 'application/json' } } + ) + ); + vi.stubGlobal('fetch', request); + const handler = vi.fn(() => 'Never'); + const store: ToolExecutionStore = { + claim: vi.fn(async () => 'claimed' as const), + record: vi.fn(async () => undefined), + }; + const session = createSession({ + assistantId: 'assistant-1', + threadId: 'thread-1', + apiUrl: 'https://runtime.example/api', + clientOptions: { defaultHeaders: { authorization: 'session-token' } }, + executionStore: store, + tools: { weather: { description: 'Weather', handler } }, + }); + const external = new AbortController(); + const notify = vi.fn(); + const off = session.subscribe(notify); + try { + session.getSnapshot(); + expect(request).not.toHaveBeenCalled(); + expect(session.load).toBeTypeOf('function'); + await session.load?.({ signal: external.signal }); + expect(request).toHaveBeenCalledTimes(1); + const [url, init] = request.mock.calls[0]; + expect(String(url)).toBe( + 'https://runtime.example/api/threads/thread-1/history' + ); + expect(init?.method).toBe('POST'); + expect(JSON.parse(String(init?.body))).toEqual({ limit: 10 }); + expect(new Headers(init?.headers).get('authorization')).toBe( + 'session-token' + ); + expect(init?.signal).toBeInstanceOf(AbortSignal); + expect(init?.signal).not.toBe(external.signal); + expect( + session.getSnapshot().messages.map((message) => message.id) + ).toEqual(['persisted-user', 'assistant-tool']); + expect(session.getSnapshot().messages[1].delivery).toEqual({ + generation: 'assistant-tool', + phase: 'complete', + outcome: 'success', + }); + expect(session.getSnapshot().toolCalls).toMatchObject([ + { id: 'call-weather', status: 'pending', args: { city: 'Paris' } }, + ]); + expect(handler).not.toHaveBeenCalled(); + expect(store.claim).not.toHaveBeenCalled(); + expect(store.record).not.toHaveBeenCalled(); + expect(notify).toHaveBeenCalledTimes(1); + } finally { + off(); + await session.dispose(); + } + }); + it.each([ ['missing options', undefined, 1], ['empty options', {}, 1], diff --git a/libs/langgraph/src/runtime/wire-message.ts b/libs/langgraph/src/runtime/wire-message.ts new file mode 100644 index 000000000..48518ee2f --- /dev/null +++ b/libs/langgraph/src/runtime/wire-message.ts @@ -0,0 +1,51 @@ +import type { Message } from '@threadplane/core'; + +export function record(value: unknown): Record | undefined { + return typeof value === 'object' && value !== null && !Array.isArray(value) + ? (value as Record) + : undefined; +} + +export function roleOf( + raw: Record +): Message['role'] | undefined { + switch (raw['type'] ?? raw['role']) { + case 'human': + case 'HumanMessage': + case 'user': + return 'user'; + case 'ai': + case 'AIMessage': + case 'AIMessageChunk': + case 'assistant': + return 'assistant'; + case 'system': + case 'SystemMessage': + return 'system'; + case 'tool': + case 'ToolMessage': + return 'tool'; + default: + return undefined; + } +} + +export function textContent(value: unknown): string { + if (typeof value === 'string') return value; + if (!Array.isArray(value)) return ''; + return value + .flatMap((block) => { + const content = record(block); + return content?.['type'] === 'text' && typeof content['text'] === 'string' + ? [content['text']] + : []; + }) + .join(''); +} + +export function hasPause(value: unknown): boolean { + const data = record(value); + return ( + Array.isArray(data?.['__interrupt__']) && data['__interrupt__'].length > 0 + ); +} diff --git a/libs/react/src/use-agent.spec.tsx b/libs/react/src/use-agent.spec.tsx index 7805efae2..b2b6cbe64 100644 --- a/libs/react/src/use-agent.spec.tsx +++ b/libs/react/src/use-agent.spec.tsx @@ -28,6 +28,39 @@ afterEach(async () => { }); describe('useAgent borrowed session', () => { + it('observes explicit history loads without owning reads, refreshes, or teardown', async () => { + const f = fixture(); + let renders = 0; + function History() { + const snapshot = useAgent(f.session); + renders++; + return {snapshot.messages.map((message) => message.content).join('\n')}; + } + 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'); + const snapshot = f.session.getSnapshot(); + const beforeRefresh = renders; + await act(async () => { await f.session.load(); }); + expect(f.session.getSnapshot()).toBe(snapshot); + expect(renders).toBe(beforeRefresh); + expect(f.history.reads).toBe(2); + view.unmount(); + const reattached = render(, { reactStrictMode: true }); + expect(reattached.getByRole('status').textContent).toBe('Saved question\nSaved answer'); + expect(f.history.reads).toBe(2); + reattached.unmount(); + f.history.value = []; + await f.session.load(); + expect(f.session.getSnapshot().messages).toEqual([]); + expect(f.history.reads).toBe(3); + expect(f.handlerCalls).toBe(0); + expect(f.streams).toHaveLength(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 () => { const f = fixture(); let run: ReturnType | undefined; diff --git a/scripts/react-parity/baseline.json b/scripts/react-parity/baseline.json index 88fac05b5..29d6b5653 100644 --- a/scripts/react-parity/baseline.json +++ b/scripts/react-parity/baseline.json @@ -1,35 +1,15 @@ { "schemaVersion": 1, - "baselineHead": "bdcc22ed31aa94f420077e046e88e1481088d453", + "baselineHead": "e1da2bd10d0f009924eeb2ea67203da92d71cd0b", "sourceState": { "modified": [ - ".github/workflows/ci.yml", - "libs/langgraph/eslint.config.mjs", - "libs/langgraph/project.json", - "libs/langgraph/src/lib/agent.types.ts", - "libs/langgraph/src/lib/client/create-langgraph-client.ts", - "libs/langgraph/src/lib/runtime-operation-reporter.ts", - "libs/langgraph/src/lib/transport/fetch-stream.transport.ts", - "libs/langgraph/src/lib/transport/transport.interface.ts", - "libs/langgraph/tsconfig.lib.json", - "libs/langgraph/tsconfig.lib.prod.json", - "libs/langgraph/vite.config.mts", - "package-lock.json" - ], - "untracked": [ "libs/langgraph/src/runtime/create-session.ts", - "libs/langgraph/src/runtime/function-tools.ts", - "libs/langgraph/src/runtime/message-reducer.ts", - "libs/langgraph/src/runtime/operation-errors.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", - "libs/langgraph/src/runtime/testing/controlled-transport.ts", - "libs/langgraph/src/runtime/testing/deferred.ts", - "libs/langgraph/src/runtime/transport.types.ts", - "libs/langgraph/tsconfig.runtime-tests.json", - "libs/langgraph/vite.runtime.config.mts" + "libs/langgraph/src/runtime/testing/binding-fixture.ts" + ], + "untracked": [ + "libs/langgraph/src/runtime/history-projection.ts", + "libs/langgraph/src/runtime/wire-message.ts" ] }, "scope": { @@ -12585,7 +12565,7 @@ "id": "source:libs/langgraph/src/runtime/create-session.ts", "kind": "source", "path": "libs/langgraph/src/runtime/create-session.ts", - "sha256": "4867cf1b7c6901174ba965265f6593b0b84113b28dcc75f5de9d68088a4b572e" + "sha256": "80db69d29a2fb196d149d10d925f12c7b988695a5e5bbd0c2b6508b15d42bc6e" }, { "id": "source:libs/langgraph/src/runtime/function-tools.ts", @@ -12593,6 +12573,12 @@ "path": "libs/langgraph/src/runtime/function-tools.ts", "sha256": "873ce2341b883ab6bcc17b7abf2d3eb0b61a3d5a9a7128449b8d4b90f5ae06fd" }, + { + "id": "source:libs/langgraph/src/runtime/history-projection.ts", + "kind": "source", + "path": "libs/langgraph/src/runtime/history-projection.ts", + "sha256": "f7065276cc2d2122d8cec64311592210b521a9d76cc35550ba94d85d8cbd4742" + }, { "id": "source:libs/langgraph/src/runtime/message-reducer.ts", "kind": "source", @@ -12621,13 +12607,13 @@ "id": "source:libs/langgraph/src/runtime/stream-projection.ts", "kind": "source", "path": "libs/langgraph/src/runtime/stream-projection.ts", - "sha256": "da722b46e0d9e72ef79e6e762a2571786bdba5b2fea96cb8dbd6cc30aac60c97" + "sha256": "4e2d299abf3d0c5452366515eb9af4a227aecf8b7681496b5f5c6da83f411caf" }, { "id": "source:libs/langgraph/src/runtime/testing/binding-fixture.ts", "kind": "source", "path": "libs/langgraph/src/runtime/testing/binding-fixture.ts", - "sha256": "e5f140179b43c203cca0222464a77a8447658c36747121893da498d3bb25385e" + "sha256": "41f4df4d94bfedee7f0bfa5b822bd58c07a7725eba2115ed2156665d1ff733e7" }, { "id": "source:libs/langgraph/src/runtime/testing/controlled-transport.ts", @@ -12647,6 +12633,12 @@ "path": "libs/langgraph/src/runtime/transport.types.ts", "sha256": "ca5d20d673bec0ca60d3af59b50b948e0dbd93b0cc26efe59d962f666a6c12cb" }, + { + "id": "source:libs/langgraph/src/runtime/wire-message.ts", + "kind": "source", + "path": "libs/langgraph/src/runtime/wire-message.ts", + "sha256": "0f134e1c7a664cec9df39b16aefae7595b7400c74fcb68c9c0b84b61dc500c13" + }, { "id": "source:libs/langgraph/src/test-setup.ts", "kind": "source", diff --git a/scripts/react-parity/dispositions.json b/scripts/react-parity/dispositions.json index eb74bfbbf..5eba73d1b 100644 --- a/scripts/react-parity/dispositions.json +++ b/scripts/react-parity/dispositions.json @@ -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": "Bounded runtime proof only; broader task capabilities and final public package cutover remain open." + "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." }, { "id": "source:libs/langgraph/src/runtime/function-tools.ts", @@ -11407,6 +11407,16 @@ "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/history-projection.ts", + "taskIds": [ + "T10" + ], + "treatment": "internal", + "status": "in-progress", + "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/message-reducer.ts", "taskIds": [ @@ -11456,7 +11466,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": "Shared wire decoding moved to a private helper for live and history projection. Existing T09 ownership is unchanged; broader event/reducer migration remains open." }, { "id": "source:libs/langgraph/src/runtime/testing/binding-fixture.ts", @@ -11467,7 +11477,7 @@ "treatment": "internal", "status": "in-progress", "reason": "Private controlled runtime/binding test helper; excluded from public exports and production packages.", - "note": "Bounded runtime proof only; broader task capabilities and final public package cutover remain open." + "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." }, { "id": "source:libs/langgraph/src/runtime/testing/controlled-transport.ts", @@ -11501,6 +11511,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/wire-message.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 wire role/text/pause decoding shared by stream and latest-history projection; broader T09/T10 capabilities remain open." + }, { "id": "source:libs/langgraph/src/test-setup.ts", "taskIds": [ diff --git a/scripts/react-parity/runtime-consumer.mjs b/scripts/react-parity/runtime-consumer.mjs index 6ee3413b5..4f1e9f56b 100644 --- a/scripts/react-parity/runtime-consumer.mjs +++ b/scripts/react-parity/runtime-consumer.mjs @@ -25,6 +25,21 @@ const catalog = [{ name: 'weather', description: 'Current weather' }, { name: 'c const toolCall = { type: 'ai', id: 'assistant-tool', content: '', tool_calls: [{ id: 'call-weather', name: 'weather', args: { city: 'Paris' }, type: 'tool_call' }] }; 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: [ + { 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' }, + { id: 'saved-count', name: 'count', args: { values: ['saved'] }, type: 'tool_call' }, + ] }, + { id: 'saved-result', type: 'tool', tool_call_id: 'saved-weather', content: 'Raw historical weather result' }, + { id: 'saved-final', type: 'ai', content: [{ type: 'text', text: 'Saved final answer' }] }, + ] }, + next: [], tasks: [], metadata: {}, + checkpoint: { thread_id: 'fixture-thread', checkpoint_ns: '', checkpoint_id: 'saved-checkpoint', checkpoint_map: {} }, + parent_checkpoint: null, + created_at: '2026-09-21T00:00:00Z', +}]; /** A strict wire fixture: malformed/extra operations fail the browser run. */ export function runtimeResponse(body) { @@ -110,9 +125,10 @@ export async function prepareRuntimeConsumer(root, consumer, kind) { } finally { rmSync(temporary, { recursive: true, force: true }); } } -/** Bounded fixture server: built files and exactly one deterministic run route. */ +/** Bounded fixture server: built files and deterministic history/run routes. */ export async function serveRuntimeConsumer(directory) { const requests = []; + const historyRequests = []; const errors = []; const held = new Set(); let notifyHeld; @@ -124,10 +140,17 @@ export async function serveRuntimeConsumer(directory) { const pathname = new URL(request.url, 'http://fixture').pathname; if (pathname.startsWith('/api/')) { assert.equal(request.method, 'POST'); - assert.equal(pathname, '/api/threads/fixture-thread/runs/stream', 'only expected run endpoint'); + assert.ok(['/api/threads/fixture-thread/history', '/api/threads/fixture-thread/runs/stream'].includes(pathname), 'only expected history/run endpoint'); const chunks = []; for await (const chunk of request) chunks.push(chunk); const body = JSON.parse(Buffer.concat(chunks).toString()); + if (pathname === '/api/threads/fixture-thread/history') { + assert.deepEqual(body, { limit: 10 }, 'exact SDK history body'); + assert.ok(historyRequests.length < 3, 'only three explicit history reads'); + historyRequests.push(body); + response.writeHead(200, { 'content-type': 'application/json' }); + return response.end(JSON.stringify(historyRequests.length < 3 ? savedHistory : [])); + } requests.push(body); const trace = runtimeResponse(body); response.writeHead(200, { 'content-type': 'text/event-stream', 'cache-control': 'no-cache' }); @@ -155,7 +178,7 @@ export async function serveRuntimeConsumer(directory) { server.listen(0, '127.0.0.1'); await once(server, 'listening'); return { - url: `http://127.0.0.1:${server.address().port}`, requests, errors, holdStarted, holdAborted, + url: `http://127.0.0.1:${server.address().port}`, requests, historyRequests, errors, holdStarted, holdAborted, async close() { for (const response of held) response.destroy(); const closed = once(server, 'close'); @@ -192,8 +215,43 @@ export async function runRuntimeScenarios(directory, kind) { await expect(page.getByTestId('handler-calls')).toHaveText('0'); await expect(page.getByTestId('submissions')).toHaveText('0'); 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'); + 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 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'); + await expect(page.getByTestId('delivery')).toHaveText('complete:success'); + await expect(page.getByTestId('status')).toHaveText('idle'); + assert.deepEqual(JSON.parse(await page.getByTestId('tool').innerText()), [{ id: 'saved-count', name: 'count', args: { values: ['saved'] }, status: 'pending' }]); + assert.equal(server.historyRequests.length, 1); + assert.equal(server.requests.length, 0); + await expect(page.getByTestId('handler-calls')).toHaveText('0'); + completed.push('explicit history load'); + + 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 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); + assert.equal(server.requests.length, 0); + completed.push('equal history refresh'); + + 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 expect(page.getByTestId('text')).toHaveText(''); + await expect(page.getByTestId('transcript')).toHaveText(''); + await expect(page.getByTestId('tool')).toHaveText('[]'); + await expect(page.getByTestId('handler-calls')).toHaveText('0'); + assert.equal(server.historyRequests.length, 3); + assert.equal(server.requests.length, 0); + completed.push('empty history replacement'); + await page.getByRole('button', { name: 'Send', exact: true }).click(); await expect(page.getByTestId('text')).toHaveText('Hello 🌍.'); await expect(page.getByTestId('delivery')).toHaveText('complete:success'); @@ -246,11 +304,12 @@ export async function runRuntimeScenarios(directory, kind) { await page.getByRole('button', { name: 'Send after dispose', exact: true }).click(); await expect(page.getByTestId('owner')).toHaveText('aborted'); assert.equal(server.requests.length, 6, 'cleanup/disposal/post-disposal submit creates no extra runs'); + assert.deepEqual(server.historyRequests, [{ limit: 10 }, { limit: 10 }, { limit: 10 }], 'only explicit loads read history'); completed.push('unmount and explicit disposal'); assert.deepEqual(server.errors.map(String), []); assert.deepEqual(pageErrors, []); assert.deepEqual(unexpected, []); - console.log(`${kind}: ${completed.length} browser scenarios passed (${completed.join('; ')}); 6 exact requests, one tool handler, no page errors/unexpected requests.`); + console.log(`${kind}: ${completed.length} browser scenarios passed (${completed.join('; ')}); 3 exact history reads, 6 exact run requests, one tool handler, no page errors/unexpected requests.`); return completed; } finally { try { await context?.close(); } diff --git a/scripts/react-parity/runtime-consumer.spec.mjs b/scripts/react-parity/runtime-consumer.spec.mjs index b11fbea67..eb66c0a96 100644 --- a/scripts/react-parity/runtime-consumer.spec.mjs +++ b/scripts/react-parity/runtime-consumer.spec.mjs @@ -75,3 +75,37 @@ test('unexpected fixture HTTP operations are recorded instead of silently served assert.equal(server.requests.length, 0); } finally { await server.close(); } }); + +test('history uses the exact SDK body and counts reads separately from runs', async () => { + const server = await runtime.serveRuntimeConsumer(tmpdir()); + try { + const read = () => fetch(`${server.url}/api/threads/fixture-thread/history`, { method: 'POST', body: JSON.stringify({ limit: 10 }) }); + const first = await read(); + assert.equal(first.status, 200); + const saved = await first.json(); + 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(), []); + assert.deepEqual(server.historyRequests, [{ limit: 10 }, { limit: 10 }, { limit: 10 }]); + assert.equal(server.requests.length, 0); + assert.deepEqual(server.errors, []); + } finally { await server.close(); } +}); + +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], + ['missing limit', '/api/threads/fixture-thread/history', 'POST', {}], + ['extra fields', '/api/threads/fixture-thread/history', 'POST', { limit: 10, before: 'unexpected' }], +]) { + test(`history rejects ${label} without recording a valid read or run`, async () => { + const server = await runtime.serveRuntimeConsumer(tmpdir()); + try { + const response = await fetch(`${server.url}${route}`, { method, ...(body ? { body: JSON.stringify(body) } : {}) }); + assert.equal(response.status, 500); + assert.deepEqual(server.historyRequests, []); + assert.deepEqual(server.requests, []); + assert.equal(server.errors.length, 1); + } finally { await server.close(); } + }); +}