From e4862cfaa93e3ae6a8b6c100ec9737e763bc9e23 Mon Sep 17 00:00:00 2001 From: Brian Love Date: Mon, 21 Sep 2026 21:39:30 -0700 Subject: [PATCH] feat(langgraph): observe readonly interrupt batches --- fixtures/react-parity/README.md | 24 +- fixtures/react-parity/runtime/README.md | 42 +- fixtures/react-parity/runtime/angular-app.ts | 2 + fixtures/react-parity/runtime/evidence.json | 373 ++++++--- fixtures/react-parity/runtime/react-app.tsx | 2 + fixtures/react-parity/runtime/scenarios.ts | 9 + libs/angular/src/observe-agent.spec.ts | 36 + libs/angular/src/observe-agent.type-test.ts | 46 +- libs/langgraph/src/runtime/create-session.ts | 83 +- .../src/runtime/history-projection.spec.ts | 38 +- .../src/runtime/history-projection.ts | 14 +- .../src/runtime/interrupt-projection.spec.ts | 386 +++++++++ .../src/runtime/interrupt-projection.ts | 117 +++ .../runtime/interrupt-projection.type-test.ts | 37 + libs/langgraph/src/runtime/interrupts.spec.ts | 777 ++++++++++++++++++ .../src/runtime/interrupts.type-test.ts | 38 + .../src/runtime/langgraph-snapshot.ts | 7 + libs/langgraph/src/runtime/ownership.ts | 19 +- .../langgraph/src/runtime/publication.spec.ts | 40 +- .../src/runtime/session-lifecycle.spec.ts | 8 +- .../src/runtime/stream-projection.spec.ts | 13 + .../src/runtime/stream-projection.ts | 7 +- .../src/runtime/testing/binding-fixture.ts | 42 +- .../src/runtime/transport.integration.spec.ts | 124 +++ libs/langgraph/src/runtime/values.spec.ts | 9 +- libs/langgraph/src/runtime/wire-message.ts | 7 - libs/react/src/use-agent.spec.tsx | 33 + libs/react/src/use-agent.type-test.ts | 46 +- scripts/react-parity/baseline.json | 32 +- scripts/react-parity/dispositions.json | 25 +- scripts/react-parity/runtime-consumer.mjs | 74 +- .../react-parity/runtime-consumer.spec.mjs | 22 +- 32 files changed, 2308 insertions(+), 224 deletions(-) create mode 100644 libs/langgraph/src/runtime/interrupt-projection.spec.ts create mode 100644 libs/langgraph/src/runtime/interrupt-projection.ts create mode 100644 libs/langgraph/src/runtime/interrupt-projection.type-test.ts create mode 100644 libs/langgraph/src/runtime/interrupts.spec.ts create mode 100644 libs/langgraph/src/runtime/interrupts.type-test.ts diff --git a/fixtures/react-parity/README.md b/fixtures/react-parity/README.md index aafbce4c5..0f42c1517 100644 --- a/fixtures/react-parity/README.md +++ b/fixtures/react-parity/README.md @@ -17,15 +17,20 @@ tools never execute. Private snapshots now include broad readonly application `values`, published atomically with messages. Full root state, history and conclusive correlated recovery replace the map; unchanged nested data retains identity, and token-only updates reuse it without traversal. `undefined` is unobserved state and -`{}` an observed empty map. Child/update/custom/live-interrupt envelopes are ignored. +`{}` an observed empty map. Child/update/custom/live-interrupt envelopes do not +replace application values. A separate readonly interrupt batch now joins the same +aggregate: root controls accumulate full payloads, and checkpoints/history replace +them. An empty `__interrupt__` array represents a static breakpoint; explicit empty standalone +batches clear interrupts. Observation offers no resume or target-selection API. Core contracts are unchanged; native structural signatures preserve the concrete snapshot extension and tool inference, with the same receiver and lifetime behavior. Angular and React borrow the app-owned session and observe it through their native -lifecycles. The installed consumers run ten scenarios each: inert mount, explicit +lifecycles. The installed consumers run eleven 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 +roundtrip, protected error, held partial text and Stop, full pause batch retained +after 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 +Six live component submissions plus one tool continuation produce seven 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 T09/T10 coverage: thread switching, pagination, branching, @@ -40,12 +45,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,455 records**: the historical 1,438 plus twelve private +The current inventory has **1,456 records**: the historical 1,438 plus thirteen private runtime production sources, three testing helpers, a runtime Vitest config and its type-test config asset. Public export occurrences remain 550 with 514 distinct local definitions. The original runtime extraction changed fourteen existing export records' declaration/import text; history loading and values observation each add -two private sources and change no legacy public export records. There are no legacy export-name additions +two private sources; interrupt observation adds one. These increments change no +legacy public export records. There are no legacy export-name additions or removals. Existing task assignments are preserved; touched extraction/configuration subsets are in progress, not whole T03–T16 completion. Core and native package contracts remain outside this legacy @@ -169,7 +175,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 ten shared scenarios. Inferred native contract +production browser apps with the eleven 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 @@ -225,7 +231,7 @@ burst streams and repeated agent/thread disposal. The foundation branch was `codex/react-support-baseline`; the runtime branch is `codex/shared-runtime-quality`, based on `bdcc22ed31aa94f420077e046e88e1481088d453`. -The current values increment is `codex/langgraph-state-values`; its verified +The current interrupt increment is `codex/langgraph-interrupt-observation`; 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` @@ -243,7 +249,7 @@ research documents. It is a scope map, not evidence that the tasks are complete. T01/T02 describe the foundation increment. The current G1 proof is deliberately limited to shared LangGraph text streaming and fixed function-tool execution with borrowed native Angular and React bindings: the runtime owns execution while each binding observes -it. Explicit history loading and application-values observation cover further +it. Explicit history loading, application-values and interrupt observation cover further subsets of T09/T10. Renderer reuse and SSR are deferred gates, alongside the broader T01–T39 map. This bounded runtime proof does not establish complete migration parity. diff --git a/fixtures/react-parity/runtime/README.md b/fixtures/react-parity/runtime/README.md index 0fb8974b5..944b4b96a 100644 --- a/fixtures/react-parity/runtime/README.md +++ b/fixtures/react-parity/runtime/README.md @@ -28,7 +28,9 @@ versions come from the root lockfile. Contract probes compile the installed public entries with `strict` and `skipLibCheck:false`, standard DOM signals, and no workspace aliases. Negative probes check names, arguments, results and deep readonly types directly on each binding's inferred snapshot, including broad -backend values without application-schema inference. +backend values and interrupt metadata/payloads without application-schema inference. +The installed fixture shape uses core types only; the compiler-emitted factory +declaration and native inferred result are checked together, including tool types. `runtime-entry.ts` is development-only composition around private `createSession`, the production `FetchStreamTransport`, and the real LangGraph SDK. A focused @@ -72,6 +74,26 @@ retaining typed tool results, method receivers and borrowed lifetime semantics. The factory and snapshot extension remain private; state writes, application schema inference, SSR and package-root cutover remain outside this slice. +`LangGraphSnapshot.interrupts` now observes a readonly batch in that same immutable +aggregate. Each item retains SDK interrupt metadata and an owned plain payload; +neither metadata nor payload is a resume command or an inferred application schema. +Root `values`/`updates` controls must have an own array-valued `__interrupt__` field. +Separate dynamic batches accumulate: the first occurrence of a string ID wins, +while anonymous entries stay distinct. An empty `__interrupt__` array replaces +the batch with the static breakpoint sentinel `{ when: 'breakpoint' }`; the next +dynamic batch replaces that sentinel. An explicit empty standalone `interrupts` +batch clears the observed batch. + +Authoritative checkpoints and latest history replace interrupts. A valid values +control takes precedence; otherwise all top-level task interrupt arrays contribute. +Child namespaces, nested task state and `next` alone do not establish a root pause. +History pause delivery is derived from the same candidate as messages, values and +interrupts. Equal refreshes retain identity, and values-only refreshes share the +unchanged interrupt batch. An accepted new submission clears the prior batch; stop, disposal +and failures retain the last observed batch. Observation adds no implicit I/O, +resume or target-selection API. Existing getter/stale-candidate guards, recovery +correlation, tool handoff and next-user-input behavior remain covered. + 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, @@ -91,8 +113,8 @@ subset of T10, not thread switching, pagination, branching, state writes, interrupt resume, SSR, or a public LangGraph package cutover. Core public contracts are unchanged; the native signatures now retain the concrete snapshot extension. -The native fixtures expose Load, Send, Tool, Error, Hold and Stop buttons plus text, -transcript, values, load completion/error, status, tool result, delivery, submission and +The native fixtures expose Load, Send, Tool, Error, Hold, Pause and Stop buttons plus text, +transcript, values, interrupts, 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. @@ -100,12 +122,13 @@ React uses a Vite production build. Angular uses the existing consumer template' 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 ten browser scenarios in installed Playwright +Both built apps run the same eleven 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 +DOM updates and Stop, the full pause batch retained after Stop, reuse after Stop, +then unmount/dispose/post-disposal submission. +Six submissions through the component controls make exactly seven run requests (including one tool continuation) and call the handler once. The separate post-disposal submit attempt resolves aborted without making a request. Three explicit Load clicks make exactly three history reads with `{ limit: 10 }` @@ -116,7 +139,12 @@ the catalog and actual serialized ToolMessage payload. Values assertions distinguish unobserved from empty state, show loaded application fields, and verify replacement/deletion across root, tool, held and reused runs. Separate native component tests make four history reads to cover a values-only -refresh with unchanged messages; installed browser scenarios still make three. +refresh with unchanged messages and interrupts; installed browser scenarios still +make three. History fixtures contain two separate task payloads and show paused +delivery. The Pause button sends two separate root controls and renders both +payloads; Stop retains them without another request, and the next Send clears them. +Unrelated custom/child noise does not contain a root empty control, because an +actual empty root `__interrupt__` is a static breakpoint rather than noise. A small in-process HTTP fixture serves only built artifacts and the expected LangGraph run/history routes on dynamic port 0. The held response writes an actual SSE diff --git a/fixtures/react-parity/runtime/angular-app.ts b/fixtures/react-parity/runtime/angular-app.ts index 8623fda32..e6a8eb23b 100644 --- a/fixtures/react-parity/runtime/angular-app.ts +++ b/fixtures/react-parity/runtime/angular-app.ts @@ -19,11 +19,13 @@ const submit = (input: string) => { submissions += 1; return session.submit(inpu + {{ snapshot().status }} {{ view().text }} {{ view().transcript }} {{ view().values }} + {{ view().interrupts }} {{ loadsFinished() }} {{ loadError() }} {{ view().error }} diff --git a/fixtures/react-parity/runtime/evidence.json b/fixtures/react-parity/runtime/evidence.json index ee65e38a6..7708b9b70 100644 --- a/fixtures/react-parity/runtime/evidence.json +++ b/fixtures/react-parity/runtime/evidence.json @@ -1,14 +1,14 @@ { "schemaVersion": 1, "status": "verified-local", - "increment": "Readonly application-state values observation (partial T09/T10)", + "increment": "Readonly interrupt batch observation (partial T09/T10)", "observedOn": "2026-09-21", - "recordedAt": "2026-09-22T03:19:34.895Z", + "recordedAt": "2026-09-22T04:36:01.807Z", "source": { - "branch": "codex/langgraph-state-values", - "baseCommit": "192f94e9b57b77e1858e2e4337f0d34bcd6a9c64", - "verificationHead": "192f94e9b57b77e1858e2e4337f0d34bcd6a9c64", - "workingTree": "Verified uncommitted values-projection, native signature, transport normalization, fixture/test and metadata changes on the merged history-loading HEAD. The fingerprint identifies actual selected working-tree bytes; no future commit or future CI result is claimed.", + "branch": "codex/langgraph-interrupt-observation", + "baseCommit": "db867f83be6cb6f9735de41f6aadddf64a0d9937", + "verificationHead": "db867f83be6cb6f9735de41f6aadddf64a0d9937", + "workingTree": "Verified uncommitted interrupt projection/session integration, fixtures, tests and metadata on the merged values-observation HEAD. The fingerprint identifies the actual selected working-tree bytes, not a future commit. I04 changed documentation/inventory/evidence only after the final I03 code checks.", "fingerprint": { "algorithm": "SHA-256 of a UTF-8 manifest: one line per selected file, lowercase SHA-256(file bytes), two ASCII spaces, repo-relative path, LF; unique paths sorted by JavaScript default string ordering.", "pathspecs": [ @@ -38,9 +38,9 @@ "excludedPaths": [ "fixtures/react-parity/runtime/evidence.json" ], - "fileCount": 775, - "sha256": "0ec1ffeca45fa046535ccd6f95c3754735ab5de0f8a355d23afe41f0dbe1d9ab", - "selection": "git ls-files -z --cached --others --exclude-standard -- ; keep existing files and remove excludedPaths. Includes tracked and non-ignored untracked files. Evidence itself is excluded to avoid self-reference; eight local research/planning reports are outside all selected paths.", + "fileCount": 780, + "sha256": "431e03e9ee0e38c14a267b205962f8bb300dbb01767c0892139f1b4d3047800e", + "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; nine local research/planning reports are outside all selected paths.", "reproduce": "node --input-type=module <<'JS'\nimport {createHash} from 'node:crypto';\nimport {execFileSync} from 'node:child_process';\nimport {readFileSync,existsSync} from 'node:fs';\nconst {fingerprint:f}=JSON.parse(readFileSync('fixtures/react-parity/runtime/evidence.json')).source;\nconst sha=value=>createHash('sha256').update(value).digest('hex');\nconst paths=[...new Set(execFileSync('git',['ls-files','-z','--cached','--others','--exclude-standard','--',...f.pathspecs],{encoding:'utf8'}).split('\\0').filter(Boolean))].filter(path=>!f.excludedPaths.includes(path)&&existsSync(path)).sort();\nconst actual=sha(paths.map(path=>sha(readFileSync(path))+' '+path+'\\n').join(''));\nif(paths.length!==f.fileCount||actual!==f.sha256) throw new Error('Source fingerprint mismatch');\nconsole.log(paths.length+' files: '+actual);\nJS" }, "sourceState": { @@ -48,39 +48,36 @@ "fixtures/react-parity/README.md", "fixtures/react-parity/runtime/README.md", "fixtures/react-parity/runtime/angular-app.ts", - "fixtures/react-parity/runtime/installed-types.ts", "fixtures/react-parity/runtime/react-app.tsx", - "fixtures/react-parity/runtime/runtime-entry.ts", "fixtures/react-parity/runtime/scenarios.ts", - "libs/angular/README.md", "libs/angular/src/observe-agent.spec.ts", - "libs/angular/src/observe-agent.ts", "libs/angular/src/observe-agent.type-test.ts", - "libs/langgraph/src/lib/transport/fetch-stream.transport.spec.ts", - "libs/langgraph/src/lib/transport/fetch-stream.transport.ts", "libs/langgraph/src/runtime/create-session.ts", + "libs/langgraph/src/runtime/history-projection.spec.ts", + "libs/langgraph/src/runtime/history-projection.ts", + "libs/langgraph/src/runtime/langgraph-snapshot.ts", "libs/langgraph/src/runtime/ownership.ts", "libs/langgraph/src/runtime/publication.spec.ts", - "libs/langgraph/src/runtime/publication.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/transport.integration.spec.ts", - "libs/react/README.md", + "libs/langgraph/src/runtime/values.spec.ts", + "libs/langgraph/src/runtime/wire-message.ts", "libs/react/src/use-agent.spec.tsx", - "libs/react/src/use-agent.ts", "libs/react/src/use-agent.type-test.ts", "scripts/react-parity/baseline.json", "scripts/react-parity/dispositions.json", "scripts/react-parity/runtime-consumer.mjs", - "scripts/react-parity/runtime-consumer.spec.mjs", - "scripts/react-parity/verify-angular-package.mjs" + "scripts/react-parity/runtime-consumer.spec.mjs" ], "untracked": [ - "libs/langgraph/src/runtime/langgraph-snapshot.ts", - "libs/langgraph/src/runtime/values-projection.spec.ts", - "libs/langgraph/src/runtime/values-projection.ts", - "libs/langgraph/src/runtime/values.spec.ts", - "libs/langgraph/src/runtime/values.type-test.ts" + "libs/langgraph/src/runtime/interrupt-projection.spec.ts", + "libs/langgraph/src/runtime/interrupt-projection.ts", + "libs/langgraph/src/runtime/interrupt-projection.type-test.ts", + "libs/langgraph/src/runtime/interrupts.spec.ts", + "libs/langgraph/src/runtime/interrupts.type-test.ts" ] } }, @@ -115,14 +112,28 @@ { "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": 413, + "verificationPhase": "I04 fresh", + "testsPassed": 414, "testsFailed": 0, - "log": "/tmp/v04-final-focused.log", - "logSha256": "6415b5511886a8f22b7681fd61e0d0a125372f0a12ffaae04282c968b9bf7bed" + "log": "/tmp/i04-final-focused.log", + "logSha256": "11fadd56f031e36e2944058eb2d629e9ac773cf588d79424da892093abdc3ed5" }, { - "command": "NX_DAEMON=false npx nx run-many -t lint test type-tests build --projects=core,content,angular,react --parallel=2 --skip-nx-cache", + "command": "NX_DAEMON=false npx nx run-many -t runtime-quality,runtime-type-tests,lint -p langgraph --skip-nx-cache", "exitCode": 0, + "verificationPhase": "I03 final unchanged-code result reused", + "testFiles": 15, + "testsPassed": 359, + "typeTestsPassed": true, + "lintErrors": 0, + "existingLintWarnings": 68, + "log": "/tmp/i03-runtime-green.log", + "logSha256": "cc20283ffd737eeba07fa9ceb9bccb9eb601503ba9592d343dd7108b93ae6312" + }, + { + "command": "NX_DAEMON=false npx nx run-many -t test,type-tests,lint,build -p core,content,angular,react --skip-nx-cache", + "exitCode": 0, + "verificationPhase": "I03 final unchanged-code result reused", "projects": 4, "targetConfigurations": 16, "behaviorTests": { @@ -131,109 +142,86 @@ "react": 7 }, "content": "Empty scaffold with passWithNoTests; no content behavior count claimed.", - "log": "/tmp/v04-final-foundations.log", - "logSha256": "01c0cf761887e34b3411d98372c328d0f1fda388ce6f51b0708a8ffb58ac4230" - }, - { - "command": "NX_DAEMON=false npx nx run langgraph:runtime-quality --skip-nx-cache", - "exitCode": 0, - "testFiles": 13, - "testsPassed": 270, - "log": "/tmp/v04-final-runtime.log", - "logSha256": "c447e885bbe74aa3a2c8da9bd46e7db99b6e8953cb9cbcc40d16a9e1d92857db" - }, - { - "command": "NX_DAEMON=false npx nx run langgraph:runtime-type-tests --skip-nx-cache", - "exitCode": 0, - "log": "/tmp/v04-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/v04-final-langgraph-lint.log", - "logSha256": "3d2546b889adf06c0dedeb1966a4e9614458ce84bf40678192f46a7df21d4af4" + "typeProbeProvenance": "This final foundation run includes the strengthened native inferred-payload probes; it supersedes the earlier native-only run.", + "log": "/tmp/i03-foundations.log", + "logSha256": "5a8e68f36a6544272ea242b556cec01b47834eed8e7e86dd7026e2ef3beb0dd5" }, { - "command": "NX_DAEMON=false npx nx test langgraph --testFile=fetch-stream.transport.spec.ts --testFile=fetch-stream.transport.integration.spec.ts --testFile=create-langgraph-client.spec.ts --testFile=client-options.spec.ts --skip-nx-cache", - "exitCode": 0, - "reason": "Fresh selected legacy regression run after the transport source changed. Nx success is recorded; no undisplayed test count is inferred.", - "log": "/tmp/v04-final-legacy.log", - "logSha256": "d13fb1fe2f58b24cacb8d09fde4a4b08fcd8f5735cff62dbe1ee840b72414e4f" - }, - { - "command": "NX_DAEMON=false npx nx run-many -t build --projects=chat,langgraph,ag-ui,render,a2ui,telemetry --configuration=production --parallel=2 --skip-nx-cache", + "command": "NX_DAEMON=false npx nx run-many -t build --projects=chat,langgraph,ag-ui,render,a2ui,telemetry --configuration=production --skip-nx-cache", "exitCode": 0, + "verificationPhase": "I03 final unchanged-code result reused", "productionProjects": 6, - "log": "/tmp/v04-final-production-builds.log", - "logSha256": "6ab27d4499cf297148dde8b8699232b230ca86d085ca4d26b9730fe874617d7b" + "log": "/tmp/i03-legacy-builds.log", + "logSha256": "384a43124ce92b6f105142a4c2c0a4edaa293872f43783453ec37d8ef7cf6271" }, { "command": "node scripts/react-parity/verify-boundaries.mjs", "exitCode": 0, - "log": "/tmp/v04-final-source-boundaries.log", + "verificationPhase": "I03 final unchanged-code result reused", + "log": "/tmp/i03-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/v04-final-built-boundaries.log", + "verificationPhase": "I03 final unchanged-code result reused", + "order": "After successful foundation and six legacy production builds.", + "log": "/tmp/i03-built-boundaries.log", "logSha256": "c983a97d17ff7ced7aa9113afa031f464103f566f1bc94eec8160558ec2553f7" }, { "command": "node scripts/react-parity/verify-packages.mjs", "exitCode": 0, + "verificationPhase": "I03 final unchanged-code result reused", "privatePlainTarballs": 3, "esmTypeExports": 9, "isolatedCoreExports": 3, - "browserScenariosPassed": 10, - "log": "/tmp/v04-final-packages.log", - "logSha256": "1e6e80a4a9c4568bfe6639d61d168ee7ffa9b08e7a435f57c4615c0e231c5075" + "browserScenariosPassed": 11, + "log": "/tmp/i03-react-browser-green.log", + "logSha256": "575f95d4bba603551c85a5910e07ce9a9e51dc41cb85225718cf60030e7db18e" }, { "command": "node scripts/react-parity/verify-angular-package.mjs", "exitCode": 0, + "verificationPhase": "I03 final unchanged-code result reused", "angularAPFExports": 1, - "browserScenariosPassed": 10, - "order": "Successful retry after production builds completed; earlier artifact-availability failure is recorded separately.", - "log": "/tmp/v04-final-angular-package.log", - "logSha256": "ec02d28e5233ca0ba1276047d5dd83ed72048a6e0c1e3d2e3eebf995607ba07f" + "browserScenariosPassed": 11, + "log": "/tmp/i03-angular-browser-green.log", + "logSha256": "0c123f648eb1b3fb2a59447021945f01c0eecaeec20a99b9bb784b6fb88669a8" }, { "command": "node scripts/react-parity/inventory.mjs --write-baseline; node scripts/react-parity/inventory.mjs --check", "exitCode": 0, - "inventoryRows": 1455, - "reviewedChange": "Two private production sources added; six existing source hashes changed. No legacy public export drift or prior disposition reassignment.", - "log": "/tmp/v04-final-inventory.log", - "logSha256": "757a321eb090a7dc7ef343f29202a3ad540d3bd70f386161e11271f06abefb0f" + "verificationPhase": "I04 fresh", + "inventoryRows": 1456, + "reviewedChange": "One private production projector added; seven existing private source hashes changed. No legacy public export drift or prior disposition reassignment.", + "log": "/tmp/i04-final-inventory.log", + "logSha256": "1f2afedb4b61388b4e9880a65639cb508bf28c9af2491e96c10d0b9719f4da3b" }, { "command": "git diff --check", "exitCode": 0, - "log": "/tmp/v04-final-diff-check.log", + "verificationPhase": "I04 fresh", + "log": "/tmp/i04-final-diff-check.log", "logSha256": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" } ], "scope": { - "privateSnapshot": "LangGraphSnapshot extends the core snapshot with readonly values: Readonly> | undefined. Undefined means no current application-values map; an empty object is an observed empty root.", - "projection": "Root full values/checkpoints, explicit latest history and conclusive correlated recovery replace the map, including deletions. Only messages and __interrupt__ are excluded from application values. Missing authoritative history clears values to undefined.", - "excludedEvents": "Child namespaces, updates, custom events and live interrupt envelopes do not replace root application values.", - "publication": "Messages and values are owned and published as one immutable aggregate. Equal state retains root identity; changed state shares unchanged nested branches. Token-only updates reuse owned values without traversing them.", - "normalization": "Legacy SDK normalization protects protocol type/namespace from raw fields. messageMetadata selects delta text semantics only for actual message events; a same-named application field remains observable.", - "nativeInference": "Structural getSnapshot/subscribe parameters retain concrete backend snapshot extensions, typed tool results, method receivers and borrowed lifetimes. Omit replaces the getter instead of intersecting incompatible overloads.", - "history": "Explicit fixed-thread load remains optional; pending/failing/stale reads, active work admission, historical tool non-execution and execution deduplication retain the previously verified behavior.", - "boundary": "No core public API changes, application-schema inference, state writes, new I/O, SSR or public factory/root cutover. Private runtime and fixture extensions remain unpublished." + "privateSnapshot": "LangGraphSnapshot adds readonly interrupts: readonly LangGraphInterrupt[], where LangGraphInterrupt is DeepReadonly with owned plain payload data. Metadata and payloads are observations, not schema inference or resume/targeting instructions.", + "liveProjection": "Root values/updates with own array-valued __interrupt__ controls accumulate dynamic batches; first string ID wins and anonymous entries remain distinct. Empty __interrupt__ replaces the batch with { when: breakpoint }; the next dynamic batch replaces that sentinel. Explicit empty standalone interrupts clears the batch.", + "authoritativeProjection": "Root checkpoints/latest history replace interrupts. A valid values control takes precedence; otherwise all top-level task interrupt arrays contribute. Child namespaces, nested task state and next-only metadata do not infer a root pause.", + "publication": "Interrupts, values and messages are one owned immutable aggregate. History pause delivery uses the same candidate batch; equal reads retain identity and values-only changes share unchanged interrupts. Getter/stale-candidate guards prevent partial publication.", + "lifecycle": "Accepted new submissions clear prior interrupts. Pre-aborted/disposed submissions are inert; stop, disposal and failures retain the last observed batch. Recovery correlation, tool handoff and retained next-user input remain covered.", + "nativeAndInstalledTypes": "Observer implementations are unchanged. Native concrete inference and actual compiler-emitted installed factory declarations retain readonly metadata, namespace arrays, broad plain payloads and tool inference; fixture snapshot structure uses installed core only.", + "unchanged": "Existing values observation, transport normalization and borrowed native lifetime behavior are unchanged. Observation adds no implicit I/O, resume, target selection, state writes, application schema, public factory/root cutover or SSR." }, "acceptanceMatrix": { - "status": "Both production-built installed browser verifiers passed, including values assertions within the existing ten scenarios.", + "status": "Both final I03 production-built installed browser verifiers passed against unchanged code; evidence reused during metadata-only I04.", "frameworks": [ "Angular installed APF production app", "React installed Vite production app" ], - "scenariosPerFramework": 10, + "scenariosPerFramework": 11, "scenarios": [ "inert mount", "explicit history load", @@ -243,7 +231,8 @@ "weather tool roundtrip", "protected visible error", "held partial DOM update and Stop/native response close", - "reuse after Stop", + "full interrupt batch and retained pause after Stop", + "reuse after Stop with next accepted submission clearing interrupts", "unmount/app-owned disposal/post-disposal submit" ], "observedPerFramework": { @@ -261,9 +250,9 @@ ], "historyRunRequests": 0, "historyHandlerCalls": 0, - "componentSubmissions": 5, + "componentSubmissions": 6, "toolContinuations": 1, - "runRequests": 6, + "runRequests": 7, "toolHandlerCalls": 1, "postDisposalSubmissions": 1, "postDisposalRequests": 0, @@ -319,7 +308,135 @@ }, "reuseAfterStop": { "stage": "complete" + }, + "pause": { + "stage": "approval" } + }, + "loadedDelivery": "complete:paused", + "interrupts": { + "inertMount": [], + "historyLoad": [ + { + "id": "saved-approval", + "value": { + "question": "Approve saved request?", + "choices": [ + "yes", + "no" + ] + }, + "namespace": [ + "review", + "task-1" + ], + "when": "during", + "resumable": true, + "ns": [ + "legacy-review" + ] + }, + { + "id": "saved-confirmation", + "value": 0, + "namespace": [], + "when": "during", + "resumable": false, + "ns": [] + } + ], + "equalRefresh": [ + { + "id": "saved-approval", + "value": { + "question": "Approve saved request?", + "choices": [ + "yes", + "no" + ] + }, + "namespace": [ + "review", + "task-1" + ], + "when": "during", + "resumable": true, + "ns": [ + "legacy-review" + ] + }, + { + "id": "saved-confirmation", + "value": 0, + "namespace": [], + "when": "during", + "resumable": false, + "ns": [] + } + ], + "emptyHistoryReplacement": [], + "pause": [ + { + "id": "live-approval", + "value": { + "question": "Approve action?", + "choices": [ + "yes", + "no" + ] + }, + "namespace": [ + "review", + "live" + ], + "when": "during", + "resumable": true, + "ns": [ + "legacy-live" + ] + }, + { + "id": "live-confirmation", + "value": false, + "namespace": [], + "when": "during", + "resumable": false, + "ns": [] + } + ], + "afterStopAtCompletedPause": [ + { + "id": "live-approval", + "value": { + "question": "Approve action?", + "choices": [ + "yes", + "no" + ] + }, + "namespace": [ + "review", + "live" + ], + "when": "during", + "resumable": true, + "ns": [ + "legacy-live" + ] + }, + { + "id": "live-confirmation", + "value": false, + "namespace": [], + "when": "during", + "resumable": false, + "ns": [] + } + ], + "nextAcceptedSend": [], + "pauseDelivery": "complete:paused", + "pauseRunRequestCount": 6, + "stopAfterPauseAdditionalRequests": 0 } }, "notificationAssertions": { @@ -333,7 +450,7 @@ }, "nativeHistoryObservation": { "readsPerFrameworkTest": 4, - "coverage": "Initial load, equal refresh, values-only replacement with unchanged messages, then empty history. Installed browser scenarios still use three reads.", + "coverage": "Two task payloads with paused history delivery; equal refresh preserves snapshot/batch identity; values-only refresh shares the batch; empty history clears it.", "lifetime": "No mount reads; observer teardown does not own session execution/disposal." }, "valuesUnitEvidence": { @@ -343,39 +460,50 @@ "libs/langgraph/src/runtime/publication.spec.ts" ], "coverage": "Root/checkpoint/history/recovery replacement and deletion; nested sharing/no-op identity; atomic publication; stale candidate guards; immutable caller isolation; unobserved versus empty; ignored child/control envelopes; no token-time values traversal. These are behavioral assertions, not performance measurements." + }, + "interruptUnitEvidence": { + "sources": [ + "libs/langgraph/src/runtime/interrupt-projection.spec.ts", + "libs/langgraph/src/runtime/interrupts.spec.ts", + "libs/langgraph/src/runtime/history-projection.spec.ts", + "libs/langgraph/src/runtime/transport.integration.spec.ts" + ], + "coverage": "Root control/standalone event projection, batch ID rules, static breakpoints, checkpoint/task precedence, plain-data ownership, atomic getter guards, retention/clearing, correlated recovery, tool handoff and next-user input. No assertion count is a performance measurement." } }, "cleanup": { - "assertions": "Unmount removes component controls; app-owned disposal resolves; post-disposal submit resolves aborted without extra I/O. Only three explicit browser loads read history. Held SSE waits for native response close.", + "assertions": "Unmount removes component controls; app disposal resolves; post-disposal submit resolves aborted without extra I/O. Three explicit browser loads read history. Held SSE awaits native response close; Stop after a completed pause retains both payloads and adds no request.", "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.", + "processes": "Both final I03 verifier processes exited 0 after awaited cleanup; I04 independently reconfirmed their temporary directories are absent.", "temporaryDirectories": [ { - "path": "/var/folders/_b/0t5_pyt94n7dlqkv1gmt29300000gn/T/threadplane-consumer-x4TyEp", + "path": "/var/folders/_b/0t5_pyt94n7dlqkv1gmt29300000gn/T/threadplane-consumer-acRfSz", "existsAfterExit": false }, { - "path": "/var/folders/_b/0t5_pyt94n7dlqkv1gmt29300000gn/T/threadplane-angular-consumer-zl68Zs", + "path": "/var/folders/_b/0t5_pyt94n7dlqkv1gmt29300000gn/T/threadplane-angular-consumer-owMIZi", "existsAfterExit": false } ], - "runtimeBundleDirectoriesRemaining": [] + "runtimeBundleDirectoriesRemaining": [], + "capturedI03Cleanup": { + "log": "/tmp/i03-cleanup.log", + "logSha256": "02f4832739d9d112a2f5c35e29a0f83d895f24a852e308d19ce1e35b6b61d195", + "newArtifacts": [] + } }, "inventory": { "historicalFoundationRows": 1438, - "previousHistoryRows": 1453, - "currentRows": 1455, - "newValuesSourceFiles": [ - "libs/langgraph/src/runtime/langgraph-snapshot.ts", - "libs/langgraph/src/runtime/values-projection.ts" - ], - "sourceFiles": 477, + "previousValuesRows": 1455, + "currentRows": 1456, + "newInterruptSourceFile": "libs/langgraph/src/runtime/interrupt-projection.ts", + "sourceFiles": 478, "legacyExports": 550, "uniqueLocalDefinitions": 514, "legacyExportRecordChangesThisIncrement": 0, - "existingAssignmentsPreserved": 1453, - "dispositions": "All prior IDs, task assignments, treatments, reasons and statuses preserved. Six scoped notes updated; two internal in-progress rows added under T09/T10. No whole-task completion claim.", - "scope": "Existing 16-library inventory only; core/native contracts are separately checked. Historical baseline-evidence.json untouched; prior runtime/history evidence retained in Git history." + "existingAssignmentsPreserved": 1455, + "dispositions": "All prior IDs, task assignments, treatments, reasons and statuses preserved. Seven scoped notes updated; one internal in-progress row added under T09/T10. No whole-task completion claim.", + "scope": "Existing 16-library inventory only; core/native contracts separately checked. Historical baseline-evidence.json untouched; prior runtime evidence retained in Git history." }, "runtimePolicy": { "ownedSDKDefaultMaxRetries": 0, @@ -394,8 +522,8 @@ "fileBytes": 51240541, "lockLocationsIncludingOptionalPlatforms": 74, "productionAppModulesTransformed": 34, - "productionAppJavaScriptRawReported": "414.48 kB", - "productionAppJavaScriptGzipReported": "124.92 kB", + "productionAppJavaScriptRawReported": "416.16 kB", + "productionAppJavaScriptGzipReported": "125.45 kB", "developmentRootImportProbe": { "inputs": 5, "bytes": 47778, @@ -408,24 +536,24 @@ "lockLocationsIncludingOptionalPlatforms": 516, "bundleInputs": 257, "contentParserInputs": 0, - "productionAppRawReported": "315.92 kB", - "productionAppEstimatedTransferReported": "83.75 kB", + "productionAppRawReported": "317.59 kB", + "productionAppEstimatedTransferReported": "84.10 kB", "includes": "Angular CLI/compiler/build tooling" }, "limits": "Installed files, optional lock locations, development probes and production apps are separate diagnostics. App bundles include the staged SDK. These are not performance benchmarks or comparable framework overhead measurements." }, "ci": { "definition": "Existing library lane covers runtime, source/built boundaries and packed browser checks; it installs Chromium with --with-deps before browser verification.", - "result": "No new values PR CI result is claimed by this local record. Prior merged history/runtime CI does not verify these uncommitted changes." + "result": "No new interrupt-observation PR CI result is claimed by this local record. Prior merged values/history/runtime CI does not verify these uncommitted changes." }, "limits": [ - "Read-only broad application-data observation only; no application-schema inference, state writes, thread switching, pagination, branching or interrupt resume.", + "Read-only interrupt observation only: no resume, target selection, implicit I/O, state writes, application-schema inference, thread switching, pagination or branching.", "Private staged runtime/factory and snapshot extension remain fixture-only; no neutral LangGraph tarball or public root cutover.", - "The narrow fixture declaration is compiler-generated against installed core; Omit replaces its getter while preserving the concrete snapshot.", - "Native signatures retain backend fields and tool inference; lifecycle ownership remains with the app.", + "Compiler-emitted installed factory declarations are checked against installed core; metadata and payload types remain readonly and broad, not inferred application shapes.", + "No core public API or observer implementation changes in this increment; session lifetime remains app-owned.", "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 parity, renderer reuse, SSR, calibrated latency, retained-heap or performance claim.", + "No full T09/T10/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." @@ -435,16 +563,9 @@ "Nx reports NO_COLOR/FORCE_COLOR overlap; legacy builds report stale Browserslist data, ng-packagr export-condition overrides and keepLifecycleScripts notices.", "Vite ignores use client directives in the client-only production consumer bundle; no SSR claim is made." ], - "verificationRetry": { - "command": "node scripts/react-parity/verify-angular-package.mjs", - "initialExitCode": 1, - "log": "/tmp/v04-angular-before-build-complete.log", - "logSha256": "62a271c8ae1aba31afa623380080b550015b46f4ea18c421f2ac343441efe5f3", - "cause": "Verifier was initially scheduled concurrently with legacy production builds, which rebuild core; it saw ENOENT for dist/libs/core/package.json. This was local verification ordering, not a source failure.", - "resolution": "Waited for successful build completion, confirmed the core manifest existed, then reran the unchanged verifier successfully. No production/test changes were made." - }, "documentation": { "generatorsRun": [], - "reason": "The private runtime and native package READMEs are outside the registered legacy API/narrative/agent-context generation inputs. Legacy export records are unchanged; only directly affected README guidance was updated." - } + "reason": "Registered API roots are legacy public entries, narrative generation uses website content, and agent context uses its templates. Private runtime and fixture README guidance are outside those inputs; no generator or native README update was needed." + }, + "verificationProvenance": "I03 implementer confirmed exact commands and successful exits. Logs were inspected and hashed here. Runtime code was unchanged after its combined check; the later foundation run checked final strengthened native type probes, and final packed browsers compiled the strengthened installed probes. No relevant code changed after those final builds/scans/browser runs. I04 changed only fixture README guidance, inventory metadata and this evidence; it reran the full focused suite, inventory and diff checks. No older V04 results or retry metadata are used." } diff --git a/fixtures/react-parity/runtime/react-app.tsx b/fixtures/react-parity/runtime/react-app.tsx index 5099c128e..1f24b5e74 100644 --- a/fixtures/react-parity/runtime/react-app.tsx +++ b/fixtures/react-parity/runtime/react-app.tsx @@ -28,11 +28,13 @@ function App() { + {snapshot.status} {view.text} {view.transcript} {view.values} + {view.interrupts} {loadsFinished} {loadError} {view.error} diff --git a/fixtures/react-parity/runtime/scenarios.ts b/fixtures/react-parity/runtime/scenarios.ts index 6ec22c8fe..27621aaad 100644 --- a/fixtures/react-parity/runtime/scenarios.ts +++ b/fixtures/react-parity/runtime/scenarios.ts @@ -8,6 +8,14 @@ export interface FixtureTools { /** Fixture-local backend extension, expressed entirely through installed core. */ export type FixtureSnapshot = AgentSnapshot & { readonly values: Readonly> | undefined; + readonly interrupts: readonly { + readonly id?: string; + readonly value?: PlainValue; + readonly namespace?: readonly string[]; + readonly when?: string; + readonly resumable?: boolean; + readonly ns?: readonly string[]; + }[]; }; export function display(snapshot: FixtureSnapshot) { @@ -17,6 +25,7 @@ export function display(snapshot: FixtureSnapshot) { text: assistant.map((message) => message.content).join('\n'), transcript: snapshot.messages.map((message) => message.content).join('\n'), values: JSON.stringify(snapshot.values) ?? 'unobserved', + interrupts: JSON.stringify(snapshot.interrupts), 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 d08858d61..d73169311 100644 --- a/libs/angular/src/observe-agent.spec.ts +++ b/libs/angular/src/observe-agent.spec.ts @@ -19,6 +19,24 @@ import { } from '../../langgraph/src/runtime/testing/binding-fixture'; const fixtures: ReturnType[] = []; +const savedInterrupts = [ + { + id: 'saved-approval', + value: { question: 'Approve saved request?', choices: ['yes', 'no'] }, + namespace: ['review', 'task-1'], + when: 'during', + resumable: true, + ns: ['legacy-review'], + }, + { + id: 'saved-confirmation', + value: 0, + namespace: [], + when: 'during', + resumable: false, + ns: [], + }, +]; const injectors: EnvironmentInjector[] = []; const SESSION = new InjectionToken< ReturnType['session'] @@ -32,6 +50,7 @@ const SESSION = new InjectionToken< {{ snapshot().status }}
{{ messages }}
{{ values }} + {{ interrupts }}
{{ delivery }}
{{ tools }}
{{ snapshot().error?.message }}
@@ -61,6 +80,9 @@ class Chat { get tools() { return JSON.stringify(this.snapshot().toolCalls); } + get interrupts() { + return JSON.stringify(this.snapshot().interrupts); + } } function fixture() { const value = bindingFixture(); @@ -97,6 +119,9 @@ describe('observeAgent borrowed session', () => { view.detectChanges(); expect(f.session.load).toBeTypeOf('function'); expect(f.history.reads).toBe(0); + expect( + view.nativeElement.querySelector('[data-testid="interrupts"]').textContent + ).toBe('[]'); expect( view.nativeElement.querySelector('[data-testid="values"]').textContent ).toBe('unobserved'); @@ -115,6 +140,15 @@ describe('observeAgent borrowed session', () => { ) ).toEqual({ counter: 1, stable: { items: ['saved'] } }); const snapshot = view.componentInstance.snapshot(); + expect( + JSON.parse( + view.nativeElement.querySelector('[data-testid="interrupts"]') + .textContent + ) + ).toEqual(savedInterrupts); + expect( + view.nativeElement.querySelector('[data-testid="delivery"]').textContent + ).toContain('paused'); await f.session.load(); expect(view.componentInstance.snapshot()).toBe(snapshot); expect(notifications).toBe(1); @@ -124,6 +158,7 @@ describe('observeAgent borrowed session', () => { await f.session.load(); view.detectChanges(); const refreshed = view.componentInstance.snapshot(); + expect(refreshed.interrupts).toBe(snapshot.interrupts); expect(refreshed.values?.['stable']).toBe(snapshot.values?.['stable']); expect( JSON.parse( @@ -141,6 +176,7 @@ describe('observeAgent borrowed session', () => { await f.session.load(); expect(f.session.getSnapshot().messages).toEqual([]); expect(f.session.getSnapshot().values).toBeUndefined(); + expect(f.session.getSnapshot().interrupts).toEqual([]); expect(f.history.reads).toBe(4); expect(f.handlerCalls).toBe(0); expect(f.streams).toHaveLength(0); diff --git a/libs/angular/src/observe-agent.type-test.ts b/libs/angular/src/observe-agent.type-test.ts index 159a4e5c5..dce1bf47e 100644 --- a/libs/angular/src/observe-agent.type-test.ts +++ b/libs/angular/src/observe-agent.type-test.ts @@ -1,6 +1,50 @@ import type { Signal } from '@angular/core'; -import type { AgentSession, AgentSnapshot } from '@threadplane/core'; +import type { + AgentSession, + AgentSnapshot, + PlainValue, +} from '@threadplane/core'; import { observeAgent } from './public-api.js'; +// eslint-disable-next-line @nx/enforce-module-boundaries -- This type-only integration probe observes the actual private factory through the borrowed fixture. +import type { bindingFixture } from '../../langgraph/src/runtime/testing/binding-fixture'; + +export function observeInferredRuntime( + session: ReturnType['session'] +) { + const snapshot = observeAgent(session)(); + const payload = snapshot.interrupts[0]?.value; + const plainPayload: PlainValue = payload; + const namespace: readonly string[] | undefined = + snapshot.interrupts[0]?.namespace; + // @ts-expect-error The backend interrupt field remains readonly. + snapshot.interrupts = []; + // @ts-expect-error The observed batch remains readonly. + snapshot.interrupts.push({ value: false }); + // @ts-expect-error Namespace metadata remains readonly. + snapshot.interrupts[0].namespace?.push('changed'); + // @ts-expect-error The broad backend payload has no inferred application schema. + const assumed: { approved: boolean } = payload; + if (payload && typeof payload === 'object' && !Array.isArray(payload)) { + // @ts-expect-error Nested payload fields remain readonly. + payload['changed'] = true; + } + for (const call of snapshot.toolCalls) + if (call.status === 'complete') { + if (call.name === 'weather') { + const temperature: number = call.result.temperature; + // @ts-expect-error The weather result retains its authored object shape. + const wrong: number = call.result; + void [temperature, wrong]; + } else { + const count: number = call.result; + // @ts-expect-error The count result retains its authored primitive shape. + void call.result.temperature; + void count; + } + } + void [plainPayload, namespace, assumed]; + return snapshot; +} interface Tools { weather: { args: { city: string }; result: { temperature: number } }; diff --git a/libs/langgraph/src/runtime/create-session.ts b/libs/langgraph/src/runtime/create-session.ts index 6f1829f09..3ba8160df 100644 --- a/libs/langgraph/src/runtime/create-session.ts +++ b/libs/langgraph/src/runtime/create-session.ts @@ -17,14 +17,21 @@ import { FetchStreamTransport } from '../lib/transport/fetch-stream.transport'; import { initialMessageState, reduceMessages } from './message-reducer'; import { createPublication } from './publication'; import { projectHistory } from './history-projection'; -import type { LangGraphSnapshot, LangGraphValues } from './langgraph-snapshot'; +import type { + LangGraphInterrupt, + LangGraphSnapshot, + LangGraphValues, +} from './langgraph-snapshot'; import { projectHistoryValues, projectValues } from './values-projection'; +import { + projectHistoryInterrupts, + projectInterrupts, +} from './interrupt-projection'; import { ownMessage } from './ownership'; import { createSafeRequestError } from './operation-errors'; import { failureProjection, finalizeProjection, - hasPause, interruptionError, projectStream, record, @@ -138,9 +145,12 @@ export function createSession( messages: [], toolCalls: [], values: undefined, + interrupts: [], }); let state = initialMessageState(); let values: LangGraphValues | undefined; + let interrupts: readonly LangGraphInterrupt[] = + publication.getSnapshot().interrupts; let owner: Attempt | undefined; let recoveryAttempt: Attempt | undefined; let disposed = false; @@ -155,6 +165,7 @@ export function createSession( publication.publish({ status, values, + interrupts, messages: state.messages, toolCalls: typedTools ? state.toolCalls.filter( @@ -339,6 +350,8 @@ export function createSession( function reconcile(attempt: Attempt, history: ThreadState[]) { const previousState = state; const previousValues = values; + const previousInterrupts = interrupts; + const previousProjection = attempt.projection; const latest = history[0]; if (!latest) return undefined; const checkpointValues = record(latest.values); @@ -349,7 +362,7 @@ export function createSession( // user ID can correlate this checkpoint to this request, including when the // persisted assistant retains the ID of its earlier streamed partial. const anchor = messages.findIndex( - (value) => record(value)?.['id'] === attempt.projection.userId + (value) => record(value)?.['id'] === previousProjection.userId ); if (anchor < 0) return undefined; const after = messages.slice(anchor + 1); @@ -367,10 +380,11 @@ export function createSession( const evidence = handoffs.length ? turn.slice(Math.max(...handoffs) + 1) : turn; - const paused = - nextUser < 0 && - (hasPause(checkpointValues) || - latest.tasks?.some((task) => (task.interrupts?.length ?? 0) > 0)); + const projectedInterrupts = projectHistoryInterrupts( + previousInterrupts, + history + ); + const paused = nextUser < 0 && projectedInterrupts.length > 0; const committed = (nextUser >= 0 || latest.next.length === 0) && evidence.some((value) => { @@ -384,15 +398,16 @@ export function createSession( ); }); if (!paused && !committed) return undefined; - const projected = projectStream(previousState, attempt.projection, { + const projected = projectStream(previousState, previousProjection, { type: 'values', - data: { ...checkpointValues, messages: [messages[anchor], ...turn] }, + data: { messages: [messages[anchor], ...turn] }, }); const projectedValues = projectHistoryValues(previousValues, history); return { state: finalizeProjection(projected.state, projected.projection), - projection: projected.projection, + projection: { ...projected.projection, paused }, values: projectedValues, + interrupts: projectedInterrupts, outcome: paused ? ('paused' as const) : ('success' as const), }; } @@ -424,6 +439,10 @@ export function createSession( if (!owns(attempt)) return; if (next.done) break; const event = next.value; + const previousState = state; + const previousValues = values; + const previousInterrupts = interrupts; + const previousProjection = attempt.projection; if (event.type === 'error' && !event.namespace?.length) { const error = failureProjection( event['data'] ?? event, @@ -434,14 +453,26 @@ export function createSession( settle(attempt, 'error', error); return; } - const projected = projectStream(state, attempt.projection, event); - const projectedValues = projectValues(values, event); - // Both projections may invoke transport-owned getters. Commit neither + const projected = projectStream( + previousState, + previousProjection, + event + ); + const projectedValues = projectValues(previousValues, event); + const projectedInterrupts = projectInterrupts( + previousInterrupts, + event + ); + // All three projections may invoke transport-owned getters. Commit no // candidate if projection failed or a getter changed the owner. if (!owns(attempt)) return; state = projected.state; values = projectedValues; - attempt.projection = projected.projection; + interrupts = projectedInterrupts; + attempt.projection = { + ...projected.projection, + paused: projectedInterrupts.length > 0, + }; publish('running'); // publish drains observer commands before returning. Never dispatch or // read on behalf of an attempt a listener just stopped/superseded. @@ -465,6 +496,7 @@ export function createSession( if (recovered) { state = recovered.state; values = recovered.values; + interrupts = recovered.interrupts; attempt.projection = recovered.projection; outcome = recovered.outcome; } @@ -594,6 +626,7 @@ export function createSession( delivery: completeDelivery(generation, 'success'), }, }); + interrupts = projectHistoryInterrupts(interrupts, []); publish('running'); checking?.abort(); if (previous) close(previous, true); @@ -614,19 +647,26 @@ export function createSession( 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 + const previousState = state; + const previousValues = values; + const previousInterrupts = interrupts; + const projectedInterrupts = projectHistoryInterrupts( + previousInterrupts, + history ); - const projectedValues = projectHistoryValues(values, history); + const projected = projectHistory(previousState, history, { + interrupts: projectedInterrupts, + ...(typedTools + ? { registeredTools: new Set(definitions.keys()) } + : {}), + }); + const projectedValues = projectHistoryValues(previousValues, history); // Even a plain projection can invoke getters supplied by a transport. // Such a getter can submit/stop/dispose; never commit its stale result. if (!ownsLoad(read)) return; state = projected; values = projectedValues; + interrupts = projectedInterrupts; authoredTools.clear(); loading = undefined; read.resolve(); @@ -744,6 +784,7 @@ export function createSession( checkController = undefined; state = recoveredState; values = recovered.values; + interrupts = recovered.interrupts; captured.attempt.projection = recovered.projection; publish('idle'); }); diff --git a/libs/langgraph/src/runtime/history-projection.spec.ts b/libs/langgraph/src/runtime/history-projection.spec.ts index 5e2cfd77f..ef9190dde 100644 --- a/libs/langgraph/src/runtime/history-projection.spec.ts +++ b/libs/langgraph/src/runtime/history-projection.spec.ts @@ -49,10 +49,42 @@ function checkpoint( const initial = () => initialMessageState(); describe('pure authoritative history projection', () => { + it('uses an explicit empty values control as breakpoint evidence for the latest assistant', () => { + const messages = [human('old-u'), ai('old-a'), human('new-u'), ai('new-a')]; + const state = projectHistory(initial(), [ + checkpoint(messages, { values: { messages, __interrupt__: [] } }), + ]); + expect(state.messages[1].delivery).toEqual(staticDelivery('old-a')); + expect(state.messages[3].delivery).toEqual( + completeDelivery('new-a', 'paused') + ); + }); + + it('uses the aggregate caller interrupt candidate without revisiting raw interrupt getters', () => { + const messages = [human('u'), ai('a')]; + const saved = checkpoint(messages, { + values: { + messages, + get __interrupt__() { + throw new Error('already projected'); + }, + }, + }); + const state = projectHistory(initial(), [saved], { interrupts: [] }); + expect(state.messages[1].delivery).toEqual(staticDelivery('a')); + const paused = projectHistory(state, [saved], { + interrupts: [{ when: 'breakpoint' }], + }); + expect(paused.messages[1].delivery).toEqual( + completeDelivery('a', 'paused') + ); + }); 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'] } }), + checkpoint([ai('old')], { + values: { __interrupt__: [{ value: 'old' }] }, + }), ]); expect(state.messages.map((message) => message.id)).toEqual(['u', 'a']); expect(state.messages[1].content).toBe('Latest'); @@ -302,7 +334,7 @@ describe('pure authoritative history projection', () => { checkpoint( messages, source === 'values' - ? { values: { messages, __interrupt__: ['Continue?'] } } + ? { values: { messages, __interrupt__: [{ value: 'Continue?' }] } } : { tasks: [task] } ), ]; @@ -322,7 +354,7 @@ describe('pure authoritative history projection', () => { const messages = [human('old-u'), ai('old-a'), human('new-u')]; const state = projectHistory(initial(), [ checkpoint(messages, { - values: { messages, __interrupt__: ['Waiting'] }, + values: { messages, __interrupt__: [{ value: 'Waiting' }] }, }), ]); expect(state.messages.map((message) => message.delivery)).toEqual( diff --git a/libs/langgraph/src/runtime/history-projection.ts b/libs/langgraph/src/runtime/history-projection.ts index 003d2b692..230138e09 100644 --- a/libs/langgraph/src/runtime/history-projection.ts +++ b/libs/langgraph/src/runtime/history-projection.ts @@ -13,12 +13,17 @@ import { sameMessage, sameToolCall, } from './ownership'; -import { hasPause, record, roleOf, textContent } from './wire-message'; +import { record, roleOf, textContent } from './wire-message'; +import { projectHistoryInterrupts } from './interrupt-projection'; +import type { LangGraphInterrupt } from './langgraph-snapshot'; 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; + /** An aggregate caller supplies its already-owned candidate to avoid reading + * transport-owned interrupt getters a second time. */ + readonly interrupts?: readonly LangGraphInterrupt[]; } function sameEntries(left: readonly unknown[], right: readonly unknown[]) { @@ -39,6 +44,8 @@ export function projectHistory( history: readonly ThreadState[], options: HistoryProjectionOptions = {} ): MessageState { + const interrupts = + options.interrupts ?? projectHistoryInterrupts([], history); const latest = history[0]; const values = record(latest?.values); const rawMessages: unknown[] = Array.isArray(values?.['messages']) @@ -113,10 +120,7 @@ export function projectHistory( } }); - if ( - hasPause(values) || - latest?.tasks?.some((task) => (task.interrupts?.length ?? 0) > 0) - ) { + if (interrupts.length > 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) { diff --git a/libs/langgraph/src/runtime/interrupt-projection.spec.ts b/libs/langgraph/src/runtime/interrupt-projection.spec.ts new file mode 100644 index 000000000..c6c2115f5 --- /dev/null +++ b/libs/langgraph/src/runtime/interrupt-projection.spec.ts @@ -0,0 +1,386 @@ +import type { ThreadState } from '@langchain/langgraph-sdk'; +import { describe, expect, it } from 'vitest'; +import { + projectHistoryInterrupts, + projectInterrupts, +} from './interrupt-projection'; +import type { StreamEvent } from './transport.types'; + +const control = ( + interrupts: unknown, + type: 'values' | 'updates' = 'values' +): StreamEvent => ({ + type, + data: { __interrupt__: interrupts }, +}); +const history = (...checkpoints: unknown[]) => checkpoints as ThreadState[]; +const pending = () => + projectInterrupts([], control([{ id: 'first', value: 'pending' }])); + +describe('interrupt projection', () => { + for (const type of ['values', 'updates'] as const) { + it(`reads root ${type} controls and preserves all SDK metadata and falsy payloads`, () => { + const batch = [ + { + id: 'a', + value: { action_requests: [{ name: 'keep_wire_names' }] }, + namespace: ['child', 'task'], + when: 'during', + resumable: false, + ns: ['legacy'], + }, + { value: false }, + { value: 0 }, + { value: '' }, + { value: null }, + { value: undefined }, + {}, + ]; + expect(projectInterrupts([], control(batch, type))).toEqual(batch); + }); + } + + for (const type of ['interrupt', 'interrupts'] as const) { + for (const location of ['data', 'outer'] as const) { + it(`reads the ${location} standalone ${type} wrapper`, () => { + const item = { id: 'one', value: false }; + const payload = { [type]: type === 'interrupt' ? item : [item] }; + const event: StreamEvent = + location === 'data' ? { type, data: payload } : { type, ...payload }; + expect(projectInterrupts([], event)).toEqual([item]); + }); + } + } + + it('merges batches in wire order with first string IDs retained and anonymous entries distinct', () => { + const first = projectInterrupts( + [], + control([{ id: 'a', value: 1 }, { value: false }]) + ); + const next = projectInterrupts( + first, + control([ + { id: 'a', value: 'duplicate' }, + { id: 'b', value: 2 }, + { id: 'b', value: 'duplicate in batch' }, + { value: false }, + { id: '', value: 3 }, + { id: '', value: 4 }, + ]) + ); + expect(next).toEqual([ + { id: 'a', value: 1 }, + { value: false }, + { id: 'b', value: 2 }, + { value: false }, + { id: '', value: 3 }, + ]); + expect(next[0]).toBe(first[0]); + expect(next[1]).toBe(first[1]); + expect(next[3]).not.toBe(next[1]); + expect( + projectInterrupts(next, control([{ id: 'a', value: 'later' }])) + ).toBe(next); + }); + + it('replaces dynamic batches with a breakpoint and replaces that sentinel with later dynamic controls', () => { + const first = pending(); + const breakpoint = projectInterrupts(first, control([])); + expect(breakpoint).toEqual([{ when: 'breakpoint' }]); + expect(Object.isFrozen(breakpoint)).toBe(true); + expect(Object.isFrozen(breakpoint[0])).toBe(true); + expect(projectInterrupts(breakpoint, control([], 'updates'))).toBe( + breakpoint + ); + expect( + projectInterrupts(breakpoint, control([{ id: 'next', value: 2 }])) + ).toEqual([{ id: 'next', value: 2 }]); + }); + + it('clears on explicit empty standalone interrupts rather than creating a breakpoint', () => { + const empty = projectInterrupts(pending(), { + type: 'interrupts', + data: { interrupts: [] }, + }); + expect(empty).toEqual([]); + expect(Object.isFrozen(empty)).toBe(true); + expect( + projectInterrupts(empty, { type: 'interrupts', interrupts: [] }) + ).toBe(empty); + }); + + for (const data of [ + undefined, + null, + false, + 0, + '', + [], + {}, + { interrupts: null }, + ]) { + it(`does not fall back to an outer standalone wrapper when data is explicitly ${JSON.stringify( + data + )}`, () => { + const previous = pending(); + expect( + projectInterrupts(previous, { + type: 'interrupts', + data, + interrupts: [{ id: 'wrong' }], + }) + ).toBe(previous); + }); + } + + it('requires the supported record/list wrapper shapes and own control fields', () => { + const previous = pending(); + const malformed: StreamEvent[] = [ + { type: 'interrupt', data: { id: 'unwrapped' } }, + { type: 'interrupt', interrupt: [] }, + { type: 'interrupt', interrupt: 'wrong' }, + { type: 'interrupts', interrupts: {} }, + { type: 'interrupts', interrupts: [null] }, + control({ value: 'not an array' }), + control([false]), + control(undefined), + { + type: 'updates', + data: { node: { __interrupt__: [{ id: 'nested' }] } }, + }, + { type: 'values', data: { messages: [], stage: 'complete' } }, + ]; + for (const event of malformed) + expect(projectInterrupts(previous, event)).toBe(previous); + const inherited = Object.create({ __interrupt__: [{ id: 'inherited' }] }); + // The inherited field is not a control; prototype-only records are unsupported if selected. + expect( + projectInterrupts(previous, { type: 'values', data: inherited }) + ).toBe(previous); + }); + + for (const event of [ + { type: 'values', namespace: ['child'] }, + { type: 'updates|child' }, + { type: 'checkpoints|child' }, + { type: 'interrupt', namespace: ['child'] }, + { type: 'custom' }, + { type: 'messages' }, + { type: 'messages/complete' }, + { type: 'tasks' }, + ] satisfies StreamEvent[]) { + it(`ignores ${event.type} without reading interrupt payload getters`, () => { + const previous = pending(); + expect( + projectInterrupts(previous, { + ...event, + get data() { + throw new Error('unrelated data'); + }, + get interrupts() { + throw new Error('unrelated outer control'); + }, + }) + ).toBe(previous); + }); + } + + it('replaces from root checkpoints and uses values controls before all task controls', () => { + const checkpoint = { + values: { __interrupt__: [{ id: 'values' }] }, + tasks: [{ interrupts: [{ id: 'task' }] }], + }; + expect( + projectInterrupts(pending(), { type: 'checkpoints', data: checkpoint }) + ).toEqual([{ id: 'values' }]); + expect(projectHistoryInterrupts(pending(), history(checkpoint))).toEqual([ + { id: 'values' }, + ]); + checkpoint.values.__interrupt__ = []; + expect(projectHistoryInterrupts(pending(), history(checkpoint))).toEqual([ + { when: 'breakpoint' }, + ]); + }); + + it('restores every latest task interrupt in order without older or nested child task state', () => { + const checkpoint = { + values: { __interrupt__: 'malformed' }, + tasks: [ + { interrupts: [{ id: 'one', value: 1 }, { value: 'anonymous' }] }, + { + interrupts: [ + { id: 'two', value: 2 }, + { id: 'one', value: 'duplicate' }, + ], + }, + { state: { values: { __interrupt__: [{ id: 'child' }] } } }, + { interrupts: [] }, + ], + }; + const expected = [ + { id: 'one', value: 1 }, + { value: 'anonymous' }, + { id: 'two', value: 2 }, + ]; + expect( + projectHistoryInterrupts( + pending(), + history(checkpoint, { values: { __interrupt__: [{ id: 'old' }] } }) + ) + ).toEqual(expected); + expect( + projectInterrupts(pending(), { type: 'checkpoints', data: checkpoint }) + ).toEqual(expected); + }); + + it('clears authoritative missing history and checkpoints without inferring a pause from next', () => { + const previous = pending(); + expect(projectHistoryInterrupts(previous, [])).toEqual([]); + for (const checkpoint of [ + {}, + { next: ['pending'] }, + { values: {}, tasks: [] }, + { tasks: [{}] }, + ]) { + expect(projectHistoryInterrupts(previous, history(checkpoint))).toEqual( + [] + ); + expect( + projectInterrupts(previous, { type: 'checkpoints', data: checkpoint }) + ).toEqual([]); + } + }); + + for (const input of [undefined, null, false, 0, '', []]) { + it(`ignores nonrecord live containers and treats nonrecord history as empty (${JSON.stringify( + input + )})`, () => { + const previous = pending(); + expect(projectInterrupts(previous, { type: 'values', data: input })).toBe( + previous + ); + expect( + projectInterrupts(previous, { type: 'checkpoints', data: input }) + ).toBe(previous); + expect(projectHistoryInterrupts(previous, history(input))).toEqual([]); + }); + } + + it('owns nested plain data, never freezes the source, and isolates later mutations', () => { + const item = { + id: 'a', + value: { items: [{ count: 1 }] }, + namespace: ['root'], + ns: ['legacy'], + }; + const batch = [item]; + const next = projectInterrupts([], control(batch)); + expect(next).not.toBe(batch); + expect(next[0]).not.toBe(item); + expect(next[0].value).not.toBe(item.value); + expect(Object.isFrozen(item)).toBe(false); + expect(Object.isFrozen(item.value.items)).toBe(false); + item.value.items[0].count = 2; + item.namespace.push('later'); + expect(next[0]).toEqual({ + id: 'a', + value: { items: [{ count: 1 }] }, + namespace: ['root'], + ns: ['legacy'], + }); + expect(Object.isFrozen(next[0].value)).toBe(true); + expect(Object.isFrozen(next[0].namespace)).toBe(true); + }); + + it('preserves equal history array identity and unchanged nested branches in corrected records', () => { + const checkpoint = { + values: { + __interrupt__: [ + { id: 'a', value: { stable: { count: 1 }, changed: 1 } }, + { id: 'b', value: false }, + ], + }, + }; + const previous = projectHistoryInterrupts([], history(checkpoint)); + expect( + projectHistoryInterrupts(previous, history(structuredClone(checkpoint))) + ).toBe(previous); + checkpoint.values.__interrupt__[0].value = { + stable: { count: 1 }, + changed: 2, + }; + const next = projectHistoryInterrupts(previous, history(checkpoint)); + expect(next).not.toBe(previous); + expect(next[1]).toBe(previous[1]); + expect((next[0].value as { stable: unknown }).stable).toBe( + (previous[0].value as { stable: unknown }).stable + ); + }); + + it('does not trust shallow-frozen external prior arrays or records', () => { + const value = { count: 1 }; + const source = Object.freeze([Object.freeze({ value })]); + const next = projectHistoryInterrupts( + source, + history({ values: { __interrupt__: source } }) + ); + expect(next).not.toBe(source); + value.count = 2; + expect(next).toEqual([{ value: { count: 1 } }]); + }); + + for (const unsupported of [new Date(0), new Map(), () => undefined, 1n]) { + it(`rejects unsupported nested payloads (${typeof unsupported})`, () => { + expect(() => + projectInterrupts([], control([{ value: unsupported }])) + ).toThrow(TypeError); + expect(() => + projectHistoryInterrupts( + [], + history({ tasks: [{ interrupts: [{ value: unsupported }] }] }) + ) + ).toThrow(TypeError); + }); + } + + it('rejects unsupported root interrupt records before selection can disguise their prototype', () => { + class SdkInterrupt { + id = 'sdk'; + value = 'unsupported'; + } + expect(() => projectInterrupts([], control([new SdkInterrupt()]))).toThrow( + TypeError + ); + expect(() => + projectInterrupts([], { + type: 'interrupt', + interrupt: new SdkInterrupt(), + }) + ).toThrow(TypeError); + class SdkControl { + __interrupt__ = [{ id: 'sdk' }]; + } + expect(() => + projectInterrupts([], { type: 'values', data: new SdkControl() }) + ).toThrow(TypeError); + expect(() => + projectHistoryInterrupts([], history({ values: new SdkControl() })) + ).toThrow(TypeError); + }); + + it('rejects cyclic selected payloads without freezing caller data', () => { + const value: Record = {}; + value['self'] = value; + expect(() => projectInterrupts([], control([{ value }]))).toThrow( + TypeError + ); + expect(() => + projectHistoryInterrupts( + [], + history({ values: { __interrupt__: [{ value }] } }) + ) + ).toThrow(TypeError); + expect(value['self']).toBe(value); + expect(Object.isFrozen(value)).toBe(false); + }); +}); diff --git a/libs/langgraph/src/runtime/interrupt-projection.ts b/libs/langgraph/src/runtime/interrupt-projection.ts new file mode 100644 index 000000000..8d7bbb24d --- /dev/null +++ b/libs/langgraph/src/runtime/interrupt-projection.ts @@ -0,0 +1,117 @@ +import type { ThreadState } from '@langchain/langgraph-sdk'; +import type { PlainValue } from '@threadplane/core'; +import type { LangGraphInterrupt } from './langgraph-snapshot'; +import { ownValue, ownValueWithSharing } from './ownership'; +import type { StreamEvent } from './transport.types'; +import { record } from './wire-message'; + +type Interrupts = readonly LangGraphInterrupt[]; + +const breakpoint: Interrupts = ownValue([{ when: 'breakpoint' }]) as Interrupts; + +function assertPlainRecord(value: Record): void { + const prototype = Object.getPrototypeOf(value); + if (prototype !== Object.prototype && prototype !== null) { + // Check before extracting fields could disguise an SDK/class instance. + ownValue(value as PlainValue); + } +} + +function batch(value: unknown): Interrupts | undefined { + if (!Array.isArray(value) || !value.every((item) => record(item))) + return undefined; + return ownValue(value) as Interrupts; +} + +function valuesControl(value: unknown): Interrupts | undefined { + const values = record(value); + if (!values || !Object.hasOwn(values, '__interrupt__')) return undefined; + assertPlainRecord(values); + const interrupts = batch(values['__interrupt__']); + return interrupts?.length === 0 ? breakpoint : interrupts; +} + +function isBreakpoint(interrupts: Interrupts): boolean { + return ( + interrupts.length === 1 && + interrupts[0].when === 'breakpoint' && + Object.keys(interrupts[0]).length === 1 + ); +} + +function replace(previous: Interrupts, interrupts: Interrupts): Interrupts { + const ids = new Set(); + const unique = interrupts.filter((interrupt) => { + if (typeof interrupt.id !== 'string') return true; + if (ids.has(interrupt.id)) return false; + ids.add(interrupt.id); + return true; + }); + return ownValueWithSharing(unique, previous) as Interrupts; +} + +function merge(previous: Interrupts, interrupts: Interrupts): Interrupts { + if (interrupts.length === 0 || isBreakpoint(interrupts)) + return replace(previous, interrupts); + return replace( + previous, + isBreakpoint(previous) ? interrupts : [...previous, ...interrupts] + ); +} + +function checkpointInterrupts(checkpoint: Record): Interrupts { + assertPlainRecord(checkpoint); + const control = valuesControl(checkpoint['values']); + if (control !== undefined) return control; + const tasks = checkpoint['tasks']; + if (!Array.isArray(tasks)) return []; + return tasks.flatMap((input) => { + const task = record(input); + if (!task || !Object.hasOwn(task, 'interrupts')) return []; + assertPlainRecord(task); + return batch(task['interrupts']) ?? []; + }); +} + +/** Root controls accumulate during an attempt; authoritative checkpoints replace + * the batch. Unrelated events never traverse interrupt payloads. */ +export function projectInterrupts( + previous: Interrupts, + event: StreamEvent +): Interrupts { + if ((event.namespace?.length ?? 0) > 0 || event.type.includes('|')) + return previous; + if (event.type === 'values' || event.type === 'updates') { + const control = valuesControl(event['data']); + return control === undefined ? previous : merge(previous, control); + } + if (event.type === 'checkpoints') { + const checkpoint = record(event['data']); + return checkpoint + ? replace(previous, checkpointInterrupts(checkpoint)) + : previous; + } + if (event.type !== 'interrupt' && event.type !== 'interrupts') + return previous; + // Explicit malformed data is not permission to consume a different outer + // payload. Custom transports without data use the existing wrapper fields. + const payload = Object.hasOwn(event, 'data') ? record(event['data']) : event; + if (!payload || !Object.hasOwn(payload, event.type)) return previous; + assertPlainRecord(payload); + const input = payload[event.type]; + const interrupts = + event.type === 'interrupt' + ? batch(record(input) ? [input] : undefined) + : batch(input); + return interrupts === undefined ? previous : merge(previous, interrupts); +} + +/** Only the latest checkpoint restores interrupts; next and nested child task + * state are not evidence of a root pause. */ +export function projectHistoryInterrupts( + previous: Interrupts, + history: readonly ThreadState[] +): Interrupts { + const latest = record(history[0]); + return replace(previous, latest ? checkpointInterrupts(latest) : []); +} diff --git a/libs/langgraph/src/runtime/interrupt-projection.type-test.ts b/libs/langgraph/src/runtime/interrupt-projection.type-test.ts new file mode 100644 index 000000000..47b0134c8 --- /dev/null +++ b/libs/langgraph/src/runtime/interrupt-projection.type-test.ts @@ -0,0 +1,37 @@ +import type { Interrupt } from '@langchain/langgraph-sdk'; +import type { DeepReadonly, PlainValue } from '@threadplane/core'; +import type { LangGraphInterrupt } from './langgraph-snapshot'; +import { projectInterrupts } from './interrupt-projection'; + +const wire: Interrupt = { + id: 'approval', + value: { requested: [{ name: 'action' }] }, + namespace: ['child'], + ns: ['legacy'], + when: 'during', + resumable: false, +}; +const batch = projectInterrupts([], { type: 'interrupt', interrupt: wire }); +const interrupt: LangGraphInterrupt = batch[0]; +const sdkShape: DeepReadonly = interrupt; +const payload: PlainValue = interrupt.value; +const metadata: readonly string[] | undefined = interrupt.namespace; +// @ts-expect-error The projected batch is readonly. +batch.push(wire); +// @ts-expect-error SDK metadata is readonly. +interrupt.id = 'changed'; +// @ts-expect-error Protocol namespace arrays are readonly. +interrupt.namespace?.push('changed'); +// @ts-expect-error Legacy namespace arrays are readonly too. +interrupt.ns?.push('changed'); +// @ts-expect-error Plain payloads do not infer an application schema. +const request: { approved: boolean } = interrupt.value; +if ( + interrupt.value && + typeof interrupt.value === 'object' && + !Array.isArray(interrupt.value) +) { + // @ts-expect-error Nested payload fields remain readonly. + interrupt.value['changed'] = true; +} +void [sdkShape, payload, metadata, request]; diff --git a/libs/langgraph/src/runtime/interrupts.spec.ts b/libs/langgraph/src/runtime/interrupts.spec.ts new file mode 100644 index 000000000..393f6c229 --- /dev/null +++ b/libs/langgraph/src/runtime/interrupts.spec.ts @@ -0,0 +1,777 @@ +import type { ThreadState } from '@langchain/langgraph-sdk'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { createSession } from './create-session'; +import type { LangGraphSnapshot } from './langgraph-snapshot'; +import { controlledTransport } from './testing/controlled-transport'; +import { deferred } from './testing/deferred'; +import type { AgentTransport, StreamEvent } from './transport.types'; + +const ai = (content: string, id = 'answer') => ({ id, type: 'ai', content }); +const full = (data: unknown): StreamEvent => ({ type: 'values', data }); +const checkpoint = (values: unknown, next: string[] = []): ThreadState => ({ + values: values as ThreadState['values'], + next, + tasks: [], + metadata: null, + checkpoint: { + thread_id: 't', + checkpoint_id: 'c', + checkpoint_ns: '', + checkpoint_map: {}, + }, + parent_checkpoint: null, + created_at: null, +}); +const fixtures: { + session: { dispose(): Promise }; + streams: ReturnType>[]; +}[] = []; +function fixture() { + const streams: ReturnType>[] = []; + const acknowledgements: ReturnType>[] = []; + const starts = Array.from({ length: 8 }, () => deferred()); + const stream = vi.fn((_a, _t, _p, signal) => { + const index = streams.length; + const controlled = controlledTransport({ + signal, + ignoreAbort: true, + }); + streams.push(controlled); + starts[index].resolve(); + const iterator: AsyncIterableIterator = { + [Symbol.asyncIterator]() { + return iterator; + }, + next() { + acknowledgements[index]?.resolve(); + return controlled.stream.next(); + }, + return() { + acknowledgements[index]?.resolve(); + return Promise.resolve({ done: true, value: undefined }); + }, + }; + return iterator; + }); + const history = vi.fn>( + async () => [] + ); + const write = vi.fn>( + async () => undefined + ); + const handler = vi.fn((args: { input: string }) => args.input); + const claim = vi.fn(async () => 'claimed' as const); + const record = vi.fn(async () => undefined); + const session = createSession({ + assistantId: 'a', + threadId: 't', + transport: { stream, getHistory: history, updateState: write }, + executionStore: { claim, record }, + tools: { work: { description: 'Work', handler } }, + }); + const f = { + session, + streams, + stream, + history, + write, + handler, + claim, + record, + snapshot: () => session.getSnapshot() as LangGraphSnapshot, + started: (index = 0) => starts[index].promise, + async emit(event: StreamEvent, index = 0) { + acknowledgements[index] = deferred(); + streams[index].release(event); + await acknowledgements[index].promise; + }, + turn(values: Record = {}, index = 0) { + const user = (stream.mock.calls[index][2] as { messages: unknown[] }) + .messages[0]; + const data = Object.create( + Object.getPrototypeOf(values), + Object.getOwnPropertyDescriptors(values) + ); + Object.defineProperty(data, 'messages', { + value: [user, ai('Recovered')], + enumerable: true, + configurable: true, + }); + return checkpoint(data); + }, + }; + fixtures.push(f); + return f; +} +async function seed(f: ReturnType) { + f.history.mockResolvedValueOnce([ + checkpoint({ + messages: [ai('Saved', 'saved')], + count: 1, + __interrupt__: [{ id: 'saved', value: { prompt: ['Confirm'] } }], + }), + ]); + await f.session.load?.(); +} +const control = (interrupts: unknown): StreamEvent => + full({ __interrupt__: interrupts }); +afterEach(async () => { + await Promise.all( + fixtures.splice(0).map(async (f) => { + await f.session.dispose(); + f.streams.forEach((stream) => stream.finish()); + }) + ); +}); + +describe('session interrupt observation', () => { + for (const mode of ['close', 'check'] as const) { + it(`recovers the earlier turn as success while observing a later turn's pending batch (${mode})`, async () => { + const f = fixture(); + const run = f.session.submit('Earlier request'); + await f.started(); + if (mode === 'check') { + f.streams[0].finish(); + expect(await run).toBe('interrupted'); + } + const user = (f.stream.mock.calls[0][2] as { messages: unknown[] }) + .messages[0]; + f.history.mockResolvedValueOnce([ + checkpoint( + { + count: 2, + messages: [ + user, + ai('Earlier answer', 'earlier'), + { type: 'human', id: 'later-user', content: 'Later request' }, + { + ...ai('Later pause', 'later'), + tool_calls: [ + { id: 'later-tool', name: 'work', args: { input: 'later' } }, + ], + }, + ], + __interrupt__: [{ id: 'later-pause', value: false }], + }, + ['approval'] + ), + ]); + if (mode === 'close') { + f.streams[0].finish(); + expect(await run).toBe('success'); + } else await f.session.checkStatus?.(); + expect(f.snapshot().interrupts).toEqual([ + { id: 'later-pause', value: false }, + ]); + expect(f.snapshot().values).toEqual({ count: 2 }); + expect(f.snapshot().messages.at(-1)).toMatchObject({ + id: 'earlier', + delivery: { outcome: 'success' }, + }); + expect(f.snapshot().toolCalls).toEqual([]); + expect(f.handler).not.toHaveBeenCalled(); + expect(f.claim).not.toHaveBeenCalled(); + expect(f.write).not.toHaveBeenCalled(); + }); + } + it('starts with an owned empty batch and observation performs no I/O', () => { + const f = fixture(); + const first = f.snapshot(); + expect(first.interrupts).toEqual([]); + expect(Object.isFrozen(first.interrupts)).toBe(true); + f.session.subscribe(() => undefined)(); + expect(f.snapshot()).toBe(first); + expect(f.history).not.toHaveBeenCalled(); + expect(f.stream).not.toHaveBeenCalled(); + expect(f.handler).not.toHaveBeenCalled(); + expect(f.write).not.toHaveBeenCalled(); + }); + + it('publishes separate controls in wire order, retains them through full state, and pauses at EOF', async () => { + const f = fixture(); + const run = f.session.submit('Go'); + await f.started(); + const seen: LangGraphSnapshot[] = []; + f.session.subscribe(() => seen.push(f.snapshot())); + await f.emit(control([{ id: 'a', value: false }])); + await f.emit({ + type: 'updates', + data: { + __interrupt__: [ + { id: 'a', value: 'duplicate' }, + { id: 'b', value: 0 }, + ], + }, + }); + const batch = f.snapshot().interrupts; + expect(batch).toEqual([ + { id: 'a', value: false }, + { id: 'b', value: 0 }, + ]); + await f.emit(full({ count: 2, messages: [ai('Awaiting approval')] })); + expect(f.snapshot().interrupts).toBe(batch); + expect(seen.at(-1)).toMatchObject({ + values: { count: 2 }, + interrupts: batch, + messages: [{ role: 'user' }, { content: 'Awaiting approval' }], + }); + f.streams[0].finish(); + expect(await run).toBe('paused'); + expect(f.snapshot().messages.at(-1)?.delivery).toMatchObject({ + phase: 'complete', + outcome: 'paused', + }); + expect(f.history).not.toHaveBeenCalled(); + expect(f.handler).not.toHaveBeenCalled(); + }); + + it('treats a messageless empty control as a static breakpoint, without reads or tool execution', async () => { + const f = fixture(); + const run = f.session.submit('Go'); + await f.started(); + await f.emit(control([])); + f.streams[0].finish(); + expect(await run).toBe('paused'); + expect(f.snapshot().interrupts).toEqual([{ when: 'breakpoint' }]); + expect(f.history).not.toHaveBeenCalled(); + expect(f.claim).not.toHaveBeenCalled(); + expect(f.write).not.toHaveBeenCalled(); + }); + + it('clears pause with an explicit standalone empty list and ignores custom and child lookalikes', async () => { + const f = fixture(); + const run = f.session.submit('Go'); + await f.started(); + await f.emit(control([])); + await f.emit({ type: 'interrupts', data: { interrupts: [] } }); + await f.emit({ + type: 'custom', + data: { __interrupt__: [{ value: 'not a control' }] }, + }); + await f.emit({ type: 'values|child', data: { __interrupt__: [] } }); + await f.emit(full({ messages: [ai('Done')] })); + f.streams[0].finish(); + expect(await run).toBe('success'); + expect(f.snapshot().interrupts).toEqual([]); + }); + + it('restores all latest task interrupts atomically, deduplicates equal reads, and clears empty history', async () => { + const f = fixture(); + const saved = checkpoint({ count: 2, messages: [ai('Saved')] }); + saved.tasks = [ + { + id: 't1', + name: 'node', + error: null, + checkpoint: null, + state: null, + interrupts: [{ id: 'a', value: false }], + }, + { + id: 't2', + name: 'node', + error: null, + checkpoint: null, + state: null, + interrupts: [{ id: 'b', value: { nested: [1] } }], + }, + ]; + f.history.mockResolvedValueOnce([saved]); + const seen: LangGraphSnapshot[] = []; + f.session.subscribe(() => seen.push(f.snapshot())); + await f.session.load?.(); + expect(seen).toHaveLength(1); + expect(seen[0]).toMatchObject({ + values: { count: 2 }, + interrupts: [ + { id: 'a', value: false }, + { id: 'b', value: { nested: [1] } }, + ], + }); + expect(seen[0].messages[0].delivery).toMatchObject({ outcome: 'paused' }); + const before = f.snapshot(); + f.history.mockResolvedValueOnce([structuredClone(saved)]); + await f.session.load?.(); + expect(f.snapshot()).toBe(before); + expect(seen).toHaveLength(1); + await f.session.load?.(); + expect(f.snapshot().interrupts).toEqual([]); + expect(f.snapshot().values).toBeUndefined(); + expect(f.snapshot().messages).toEqual([]); + expect(f.stream).not.toHaveBeenCalled(); + expect(f.handler).not.toHaveBeenCalled(); + }); + + it('clears once on accepted submit while pre-aborted submits retain exact history', async () => { + const f = fixture(); + await seed(f); + const before = f.snapshot(); + expect( + await f.session.submit('Cancelled', { signal: AbortSignal.abort() }) + ).toBe('aborted'); + expect(f.snapshot()).toBe(before); + const seen: LangGraphSnapshot[] = []; + f.session.subscribe(() => seen.push(f.snapshot())); + const run = f.session.submit('Go'); + await f.started(); + expect(seen).toHaveLength(1); + expect(seen[0].interrupts).toEqual([]); + expect(seen[0].values).toBe(before.values); + expect(seen[0].status).toBe('running'); + await f.session.stop(); + expect(await run).toBe('aborted'); + }); + + for (const action of ['stop', 'dispose', 'failure'] as const) { + it(`preserves observed interrupts through ${action} and ignores late events`, async () => { + const f = fixture(); + const run = f.session.submit('Go'); + await f.started(); + await f.emit(control([{ id: 'a', value: 'pending' }])); + const before = f.snapshot().interrupts; + if (action === 'failure') + await f.emit({ type: 'error', data: { message: 'failed' } }); + else await f.session[action](); + expect(await run).toBe(action === 'failure' ? 'error' : 'aborted'); + const settled = f.snapshot(); + expect(settled.interrupts).toBe(before); + f.streams[0].release(control([{ id: 'late' }])); + f.streams[0].finish(); + await f.streams[0].closed; + expect(f.snapshot()).toBe(settled); + expect(f.write).not.toHaveBeenCalled(); + if (action === 'dispose') { + expect(await f.session.submit('No')).toBe('aborted'); + expect(f.snapshot()).toBe(settled); + } + }); + } + + it('projects messages, values, and interrupts as one publication for a checkpoint', async () => { + const f = fixture(); + const run = f.session.submit('Go'); + await f.started(); + const seen: LangGraphSnapshot[] = []; + f.session.subscribe(() => seen.push(f.snapshot())); + await f.emit({ + type: 'checkpoints', + data: checkpoint({ + count: 3, + messages: [ai('Checkpoint')], + __interrupt__: [{ id: 'approval', value: 0 }], + }), + }); + expect(seen).toHaveLength(1); + expect(seen[0]).toMatchObject({ + values: { count: 3 }, + interrupts: [{ id: 'approval', value: 0 }], + }); + expect(seen[0].messages.at(-1)?.content).toBe('Checkpoint'); + f.streams[0].finish(); + expect(await run).toBe('paused'); + }); + + it('keeps prior interrupts and values when message ownership rejects the same checkpoint', async () => { + const f = fixture(); + const run = f.session.submit('Go'); + await f.started(); + await f.emit(control([{ id: 'prior', value: false }])); + const before = f.snapshot(); + await f.emit({ + type: 'checkpoints', + data: checkpoint({ + count: 9, + __interrupt__: [{ id: 'rejected' }], + messages: [ + { + ...ai('Invalid'), + tool_calls: [{ id: 'invalid', name: 'work', args: new Date() }], + }, + ], + }), + }); + expect(await run).toBe('interrupted'); + expect(f.snapshot().interrupts).toBe(before.interrupts); + expect(f.snapshot().values).toBe(before.values); + expect(f.snapshot().messages.map((message) => message.content)).toEqual( + before.messages.map((message) => message.content) + ); + expect(f.claim).not.toHaveBeenCalled(); + expect(f.write).not.toHaveBeenCalled(); + }); + + for (const invalid of ['instance', 'cycle', 'getter'] as const) { + it(`commits no message/value/tool eligibility when interrupt ownership fails (${invalid})`, async () => { + const f = fixture(); + const run = f.session.submit('Go'); + await f.started(); + const before = f.snapshot(); + const cycle: Record = {}; + cycle['self'] = cycle; + const item = + invalid === 'getter' + ? { + get value() { + throw new Error('private'); + }, + } + : { value: invalid === 'cycle' ? cycle : new Date() }; + await f.emit({ + type: 'checkpoints', + data: checkpoint({ + count: 9, + __interrupt__: [item], + messages: [ + { + ...ai('Invalid'), + tool_calls: [ + { id: 'bad-call', name: 'work', args: { input: 'bad' } }, + ], + }, + ], + }), + }); + f.streams[0].finish(); + expect(await run).toBe('interrupted'); + expect(f.snapshot().status).toBe('error'); + expect(f.snapshot().values).toBe(before.values); + expect(f.snapshot().interrupts).toBe(before.interrupts); + expect(f.snapshot().messages.map((message) => message.content)).toEqual( + before.messages.map((message) => message.content) + ); + expect(f.snapshot().toolCalls).toEqual([]); + expect(f.claim).not.toHaveBeenCalled(); + expect(f.handler).not.toHaveBeenCalled(); + expect(f.write).not.toHaveBeenCalled(); + }); + } + + for (const phase of ['live', 'load', 'recovery'] as const) { + for (const action of ['stop', 'dispose', 'submit'] as const) { + it(`a ${phase} interrupt getter cannot overwrite a nested ${action}`, async () => { + const f = fixture(); + await seed(f); + let run: Promise | undefined; + if (phase !== 'load') { + run = f.session.submit('Go'); + await f.started(); + } + const before = f.snapshot(); + let nested: Promise | undefined; + let fired = false; + const item = { + get value() { + if (!fired) { + fired = true; + nested = + action === 'submit' + ? f.session.submit('Replacement') + : f.session[action](); + } + return { secret: 'stale' }; + }, + }; + const data = { + count: 9, + __interrupt__: [item], + messages: [ai('Stale')], + }; + if (phase === 'load') { + f.history.mockResolvedValueOnce([checkpoint(data)]); + await f.session.load?.(); + } else if (phase === 'live') { + await f.emit({ type: 'checkpoints', data: checkpoint(data) }); + f.streams[0].finish(); + } else { + f.history.mockImplementationOnce(async () => [f.turn(data)]); + f.streams[0].finish(); + } + if (run) + expect(await run).toBe( + action === 'submit' ? 'interrupted' : 'aborted' + ); + expect(fired).toBe(true); + expect(f.snapshot().values).toBe(before.values); + expect(f.snapshot().interrupts).toEqual( + action === 'submit' ? [] : before.interrupts + ); + expect( + f + .snapshot() + .messages.some((message) => + ['Stale', 'Recovered'].includes(message.content) + ) + ).toBe(false); + if (action === 'submit') { + await f.started(phase === 'load' ? 0 : 1); + await f.session.stop(); + } + await nested; + }); + } + } + + for (const mode of ['close', 'check'] as const) { + it(`recovers correlated ${mode} history with an atomic paused checkpoint`, async () => { + const f = fixture(); + const run = f.session.submit('Go'); + await f.started(); + if (mode === 'check') { + f.streams[0].finish(); + expect(await run).toBe('interrupted'); + } + f.history.mockImplementationOnce(async () => [ + f.turn({ count: 3, __interrupt__: [] }), + ]); + const seen: LangGraphSnapshot[] = []; + f.session.subscribe(() => seen.push(f.snapshot())); + if (mode === 'close') { + f.streams[0].finish(); + expect(await run).toBe('paused'); + } else await f.session.checkStatus?.(); + expect(seen).toHaveLength(1); + expect(seen[0]).toMatchObject({ + status: 'idle', + values: { count: 3 }, + interrupts: [{ when: 'breakpoint' }], + }); + expect(seen[0].messages.at(-1)?.delivery).toMatchObject({ + outcome: 'paused', + }); + }); + } + + it('retains exact interrupted state for unrelated and inconclusive checkpoints', async () => { + const f = fixture(); + const run = f.session.submit('Go'); + await f.started(); + f.streams[0].finish(); + expect(await run).toBe('interrupted'); + const before = f.snapshot(); + f.history.mockResolvedValueOnce([ + checkpoint({ + __interrupt__: [ + { + get value() { + throw new Error('must not project unrelated interrupts'); + }, + }, + ], + }), + ]); + await f.session.checkStatus?.(); + expect(f.snapshot()).toBe(before); + const user = (f.stream.mock.calls[0][2] as { messages: unknown[] }) + .messages[0]; + f.history.mockResolvedValueOnce([ + checkpoint({ messages: [user], count: 9 }, ['work']), + ]); + await f.session.checkStatus?.(); + expect(f.snapshot()).toBe(before); + }); + + for (const mode of ['load', 'close', 'check'] as const) { + it(`retains prior candidates after a throwing ${mode} interrupt getter`, async () => { + const f = fixture(); + await seed(f); + let run: Promise | undefined; + if (mode !== 'load') { + run = f.session.submit('Go'); + await f.started(); + if (mode === 'check') { + f.streams[0].finish(); + expect(await run).toBe('interrupted'); + } + } + const before = f.snapshot(); + const data = { + count: 9, + __interrupt__: [ + { + get value() { + throw new Error('private payload'); + }, + }, + ], + }; + f.history.mockImplementationOnce(async () => [ + mode === 'load' ? checkpoint(data) : f.turn(data), + ]); + if (mode === 'close') { + f.streams[0].finish(); + expect(await run).toBe('interrupted'); + } else if (mode === 'load') + await expect(f.session.load?.()).rejects.not.toThrow('private payload'); + else await expect(f.session.checkStatus?.()).rejects.toThrow(); + expect(f.snapshot().values).toBe(before.values); + expect(f.snapshot().interrupts).toBe(before.interrupts); + expect(f.snapshot().messages.map((message) => message.content)).toEqual( + before.messages.map((message) => message.content) + ); + if (mode !== 'close') expect(f.snapshot()).toBe(before); + }); + } + + for (const action of ['submit', 'checkStatus'] as const) { + it(`a status interrupt getter cannot clear a replacement ${action} owner`, async () => { + const f = fixture(); + const run = f.session.submit('Go'); + await f.started(); + f.streams[0].finish(); + expect(await run).toBe('interrupted'); + const before = f.snapshot(); + let fired = false; + let nested: Promise | undefined; + const replacement = deferred(); + f.history.mockImplementationOnce(async () => [ + f.turn({ + __interrupt__: [ + { + get value() { + if (!fired) { + fired = true; + nested = + action === 'submit' + ? f.session.submit('Replacement') + : f.session.checkStatus?.(); + } + return 'stale'; + }, + }, + ], + }), + ]); + f.history.mockReturnValueOnce(replacement.promise); + await f.session.checkStatus?.(); + expect(fired).toBe(true); + expect(f.snapshot().interrupts).toBe(before.interrupts); + expect(f.snapshot().values).toBe(before.values); + if (action === 'submit') { + await f.started(1); + await f.session.stop(); + } else + replacement.resolve([f.turn({ __interrupt__: [{ id: 'fresh' }] })]); + await nested; + if (action === 'checkStatus') + expect(f.snapshot().interrupts).toEqual([{ id: 'fresh' }]); + }); + } + + it('does not acknowledge a staged handoff after invalid interrupt projection', async () => { + const f = fixture(); + const run = f.session.submit('Work'); + await f.started(); + await f.emit( + full({ + messages: [ + { + ...ai('Tool'), + tool_calls: [ + { id: 'call', name: 'work', args: { input: 'result' } }, + ], + }, + ], + }) + ); + f.streams[0].finish(); + await f.started(1); + const handoff = (f.stream.mock.calls[1][2] as { messages: unknown[] }) + .messages[0]; + await f.emit( + { + type: 'checkpoints', + data: checkpoint({ + messages: [ai('Invalid')], + __interrupt__: [{ value: new Date() }], + }), + }, + 1 + ); + f.streams[1].finish(); + expect(await run).toBe('interrupted'); + expect(f.snapshot().status).toBe('error'); + const retry = f.session.submit('Again'); + await f.started(2); + expect( + (f.stream.mock.calls[2][2] as { messages: unknown[] }).messages[0] + ).toEqual(handoff); + expect(f.handler).toHaveBeenCalledTimes(1); + expect(f.write).not.toHaveBeenCalled(); + await f.session.stop(); + await retry; + }); + + it('keeps two session batches independent', async () => { + const first = fixture(); + const second = fixture(); + await seed(first); + expect(first.snapshot().interrupts).toHaveLength(1); + expect(second.snapshot().interrupts).toEqual([]); + await second.session.load?.(); + expect(first.snapshot().interrupts).toHaveLength(1); + }); + + for (const mode of ['load', 'close', 'check'] as const) { + it(`projects ${mode} interrupt getters once and shares the candidate with message delivery`, async () => { + const f = fixture(); + let run: Promise | undefined; + if (mode !== 'load') { + run = f.session.submit('Go'); + await f.started(); + if (mode === 'check') { + f.streams[0].finish(); + expect(await run).toBe('interrupted'); + } + } + let reads = 0; + const data = { + messages: [ai('Loaded')], + get __interrupt__() { + reads += 1; + return [{ id: 'once', value: false }]; + }, + }; + f.history.mockImplementationOnce(async () => [ + mode === 'load' ? checkpoint(data) : f.turn(data), + ]); + if (mode === 'close') { + f.streams[0].finish(); + expect(await run).toBe('paused'); + } else if (mode === 'load') await f.session.load?.(); + else await f.session.checkStatus?.(); + expect(reads).toBe(1); + expect(f.snapshot().interrupts).toEqual([{ id: 'once', value: false }]); + expect(f.snapshot().messages.at(-1)?.delivery).toMatchObject({ + outcome: 'paused', + }); + }); + } + + it('a pause with finalized pending tools never claims, executes, persists or continues them', async () => { + const f = fixture(); + const run = f.session.submit('Work'); + await f.started(); + await f.emit( + full({ + __interrupt__: [], + messages: [ + { + ...ai('Waiting'), + tool_calls: [ + { id: 'pending', name: 'work', args: { input: 'later' } }, + ], + }, + ], + }) + ); + f.streams[0].finish(); + expect(await run).toBe('paused'); + expect(f.snapshot().toolCalls).toMatchObject([ + { id: 'pending', status: 'pending' }, + ]); + expect(f.claim).not.toHaveBeenCalled(); + expect(f.handler).not.toHaveBeenCalled(); + expect(f.write).not.toHaveBeenCalled(); + expect(f.stream).toHaveBeenCalledTimes(1); + }); +}); diff --git a/libs/langgraph/src/runtime/interrupts.type-test.ts b/libs/langgraph/src/runtime/interrupts.type-test.ts new file mode 100644 index 000000000..aec0fb80b --- /dev/null +++ b/libs/langgraph/src/runtime/interrupts.type-test.ts @@ -0,0 +1,38 @@ +import type { AgentSession, PlainValue } from '@threadplane/core'; +import { createSession } from './create-session'; +import type { LangGraphInterrupt } from './langgraph-snapshot'; + +const session = createSession({ + assistantId: 'a', + threadId: 't', + tools: { + weather: { + description: 'Weather', + handler: (args: { city: string }) => ({ temperature: args.city.length }), + }, + }, +}); +const observer: AgentSession = session; +const snapshot = session.getSnapshot(); +const interrupts: readonly LangGraphInterrupt[] = snapshot.interrupts; +const value: PlainValue = interrupts[0]?.value; +// @ts-expect-error Snapshot fields are readonly. +snapshot.interrupts = []; +// @ts-expect-error The batch is readonly. +snapshot.interrupts.push({ value: false }); +// @ts-expect-error Namespace metadata is deeply readonly. +snapshot.interrupts[0].namespace?.push('changed'); +// @ts-expect-error Broad backend data does not infer an application schema. +const assumed: { approved: boolean } = value; +if (value && typeof value === 'object' && !Array.isArray(value)) { + // @ts-expect-error Payload fields are readonly. + value['approved'] = true; +} +for (const call of snapshot.toolCalls) + if (call.status === 'complete') { + const temperature: number = call.result.temperature; + // @ts-expect-error Adding interrupts preserves authored tool result inference. + const bad: string = call.result.temperature; + void [temperature, bad]; + } +void [observer, interrupts, value, assumed]; diff --git a/libs/langgraph/src/runtime/langgraph-snapshot.ts b/libs/langgraph/src/runtime/langgraph-snapshot.ts index 7cd792813..fdb56d824 100644 --- a/libs/langgraph/src/runtime/langgraph-snapshot.ts +++ b/libs/langgraph/src/runtime/langgraph-snapshot.ts @@ -1,8 +1,14 @@ import type { AgentSnapshot, + DeepReadonly, PlainValue, ToolContract, } from '@threadplane/core'; +import type { Interrupt } from '@langchain/langgraph-sdk'; + +/** Backend wire metadata and a plain payload. DeepReadonly maps the SDK's + * unknown payload to PlainValue without recursively re-mapping PlainValue. */ +export type LangGraphInterrupt = DeepReadonly; /** Observed backend application data, not a validated application schema. */ export type LangGraphValues = Readonly>; @@ -15,4 +21,5 @@ export type LangGraphSnapshot< > > = AgentSnapshot & { readonly values: LangGraphValues | undefined; + readonly interrupts: readonly LangGraphInterrupt[]; }; diff --git a/libs/langgraph/src/runtime/ownership.ts b/libs/langgraph/src/runtime/ownership.ts index 4da6b5d4f..42e16f75c 100644 --- a/libs/langgraph/src/runtime/ownership.ts +++ b/libs/langgraph/src/runtime/ownership.ts @@ -8,7 +8,11 @@ import { type PlainValue, type ToolCall, } from '@threadplane/core'; -import type { LangGraphSnapshot, LangGraphValues } from './langgraph-snapshot'; +import type { + LangGraphInterrupt, + LangGraphSnapshot, + LangGraphValues, +} from './langgraph-snapshot'; // Only objects projected here are trusted. Object.isFrozen on external input is // insufficient: its children may still be mutable. The weak set retains no data. @@ -282,6 +286,15 @@ export function ownLangGraphSnapshot( const values = ownValueWithSharing(input.values, previous?.values) as | LangGraphValues | undefined; - if (core === previous && values === previous?.values) return previous; - return freeze({ ...core, values }); + const interrupts = ownValueWithSharing( + input.interrupts, + previous?.interrupts + ) as readonly LangGraphInterrupt[]; + if ( + core === previous && + values === previous?.values && + interrupts === previous?.interrupts + ) + return previous; + return freeze({ ...core, values, interrupts }); } diff --git a/libs/langgraph/src/runtime/publication.spec.ts b/libs/langgraph/src/runtime/publication.spec.ts index 27a0cd7b8..437a0a386 100644 --- a/libs/langgraph/src/runtime/publication.spec.ts +++ b/libs/langgraph/src/runtime/publication.spec.ts @@ -8,6 +8,7 @@ function snapshot(content = 'one'): LangGraphSnapshot { return { status: 'idle', values: undefined, + interrupts: [], messages: [ { id: 'm', @@ -30,6 +31,36 @@ function snapshot(content = 'one'): LangGraphSnapshot { } describe('private snapshot publication', () => { + it('owns interrupt payloads and suppresses equal aggregate publications', () => { + const payload = { nested: [1] }; + const publication = createPublication({ + ...snapshot(), + interrupts: [{ id: 'a', value: payload, namespace: ['root'] }], + }); + const before = publication.getSnapshot(); + expect(before.interrupts).toEqual([ + { id: 'a', value: { nested: [1] }, namespace: ['root'] }, + ]); + payload.nested.push(2); + expect(before.interrupts[0].value).toEqual({ nested: [1] }); + expect(Object.isFrozen(before.interrupts[0].namespace)).toBe(true); + let calls = 0; + publication.subscribe(() => calls++); + publication.publish({ + ...snapshot(), + interrupts: [{ id: 'a', value: { nested: [1] }, namespace: ['root'] }], + }); + expect(publication.getSnapshot()).toBe(before); + expect(calls).toBe(0); + expect(() => + publication.publish({ + ...snapshot('Invalid'), + values: { count: 9 }, + interrupts: [{ value: new Date() }], + } as unknown as LangGraphSnapshot) + ).toThrow(TypeError); + expect(publication.getSnapshot()).toBe(before); + }); it('leaves the aggregate unchanged if application values ownership throws', () => { const publication = createPublication(snapshot()); const before = publication.getSnapshot(); @@ -55,8 +86,12 @@ describe('private snapshot publication', () => { publication.subscribe(() => { if (publication.getSnapshot().status === 'running') { const values = { stable: { x: 1 }, count: 2 }; - publication.publish({ ...snapshot('two'), values }); + const interrupts = [ + { id: 'queued', value: { choices: ['continue'] } }, + ]; + publication.publish({ ...snapshot('two'), values, interrupts }); values.count = 99; + interrupts[0].value.choices.push('late mutation'); } }); publication.subscribe(() => @@ -66,6 +101,9 @@ describe('private snapshot publication', () => { expect(seen).toHaveLength(2); expect(seen[1].values).toEqual({ stable: { x: 1 }, count: 2 }); expect(seen[1].values?.['stable']).toBe(first.values?.['stable']); + expect(seen[1].interrupts).toEqual([ + { id: 'queued', value: { choices: ['continue'] } }, + ]); expect(seen[1].messages[0].content).toBe('two'); expect(Object.isFrozen(seen[1].values)).toBe(true); }); diff --git a/libs/langgraph/src/runtime/session-lifecycle.spec.ts b/libs/langgraph/src/runtime/session-lifecycle.spec.ts index 91f154f89..42dc42521 100644 --- a/libs/langgraph/src/runtime/session-lifecycle.spec.ts +++ b/libs/langgraph/src/runtime/session-lifecycle.spec.ts @@ -70,7 +70,13 @@ describe('neutral session ownership', () => { const unsubscribe = a.session.subscribe(first); a.session.subscribe(second); expect(a.session.getSnapshot()).toBe(initial); - expect(initial).toEqual({ status: 'idle', messages: [], toolCalls: [] }); + expect(initial).toEqual({ + status: 'idle', + messages: [], + toolCalls: [], + values: undefined, + interrupts: [], + }); expect(first).not.toHaveBeenCalled(); expect(a.stream).not.toHaveBeenCalled(); const run = a.session.submit('Hello'); diff --git a/libs/langgraph/src/runtime/stream-projection.spec.ts b/libs/langgraph/src/runtime/stream-projection.spec.ts index 9a91d2039..01ea0c5f7 100644 --- a/libs/langgraph/src/runtime/stream-projection.spec.ts +++ b/libs/langgraph/src/runtime/stream-projection.spec.ts @@ -32,6 +32,19 @@ function start(messages = [assistant('step', ['c1'])]) { } describe('authoritative tool ownership', () => { + it('leaves pause classification to the aggregate interrupt projector without reading control getters', () => { + const before = start(); + const next = projectStream(before.state, before.projection, { + type: 'custom', + data: { + get __interrupt__() { + throw new Error('not a control event'); + }, + }, + }); + expect(next.state).toBe(before.state); + expect(next.projection).toBe(before.projection); + }); it.each([{ ids: [] }, { ids: ['c2'] }])( 'reconciles one assistant call list to $ids', ({ ids }) => { diff --git a/libs/langgraph/src/runtime/stream-projection.ts b/libs/langgraph/src/runtime/stream-projection.ts index 78f0ad475..e03ea6139 100644 --- a/libs/langgraph/src/runtime/stream-projection.ts +++ b/libs/langgraph/src/runtime/stream-projection.ts @@ -12,9 +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'; +import { record, roleOf, textContent } from './wire-message'; -export { hasPause, record } from './wire-message'; +export { record } from './wire-message'; type CanonicalMessage = Extract; @@ -49,9 +49,6 @@ export function projectStream( const messageEvent = type === 'messages' || type.startsWith('messages/'); const terminal = type === 'values' || type === 'messages/complete' || type === 'checkpoints'; - if (type === 'interrupt' || type === 'interrupts' || hasPause(values)) { - projection = { ...projection, paused: true }; - } if (!terminal && !messageEvent) return { state, projection }; const mode = messageEvent && event.messageMetadata ? 'delta' : 'snapshot'; const incoming = Array.isArray(messages) diff --git a/libs/langgraph/src/runtime/testing/binding-fixture.ts b/libs/langgraph/src/runtime/testing/binding-fixture.ts index 4c58a44ca..026ab28ab 100644 --- a/libs/langgraph/src/runtime/testing/binding-fixture.ts +++ b/libs/langgraph/src/runtime/testing/binding-fixture.ts @@ -31,8 +31,46 @@ export function bindingFixture() { { id: 'saved-answer', type: 'ai', content: 'Saved answer' }, ], }, - next: [], - tasks: [], + next: ['review', 'confirmation'], + tasks: [ + { + id: 'review-task', + name: 'review', + error: null, + checkpoint: null, + state: null, + interrupts: [ + { + id: 'saved-approval', + value: { + question: 'Approve saved request?', + choices: ['yes', 'no'], + }, + namespace: ['review', 'task-1'], + when: 'during', + resumable: true, + ns: ['legacy-review'], + }, + ], + }, + { + id: 'confirmation-task', + name: 'confirmation', + error: null, + checkpoint: null, + state: null, + interrupts: [ + { + id: 'saved-confirmation', + value: 0, + namespace: [], + when: 'during', + resumable: false, + ns: [], + }, + ], + }, + ], metadata: {}, checkpoint: { thread_id: 'binding-thread', diff --git a/libs/langgraph/src/runtime/transport.integration.spec.ts b/libs/langgraph/src/runtime/transport.integration.spec.ts index d477ad73f..5f36491dc 100644 --- a/libs/langgraph/src/runtime/transport.integration.spec.ts +++ b/libs/langgraph/src/runtime/transport.integration.spec.ts @@ -70,6 +70,130 @@ describe('neutral real SDK transport', () => { vi.unstubAllGlobals(); }); + it('recognizes an actual SDK empty interrupt control as a messageless breakpoint', async () => { + const request = vi.fn(async (url) => { + expect(String(url)).toBe('https://runtime.example/threads/t/runs/stream'); + return fragmentedResponse( + 'event: values\ndata: {"__interrupt__":[]}\n\n' + ); + }); + vi.stubGlobal('fetch', request); + const session = createSession({ + assistantId: 'a', + threadId: 't', + apiUrl: 'https://runtime.example', + }); + try { + expect(await session.submit('Pause')).toBe('paused'); + expect(session.getSnapshot().interrupts).toEqual([ + { when: 'breakpoint' }, + ]); + expect(session.getSnapshot().status).toBe('idle'); + expect(session.getSnapshot().messages).toHaveLength(1); + expect(request).toHaveBeenCalledTimes(1); + } finally { + await session.dispose(); + } + }); + + it('observes real SDK values and updates interrupt batches and history tasks without extra I/O', async () => { + const item = { + id: 'a', + value: { prompt: ['Approve'] }, + namespace: ['root'], + when: 'during', + resumable: false, + ns: ['legacy'], + }; + const history = [ + { + values: { + count: 2, + messages: [{ type: 'ai', id: 'saved', content: 'Saved' }], + }, + next: ['approval'], + tasks: [ + { + id: 'task-a', + name: 'a', + interrupts: [{ id: 'history-a', value: false }], + }, + { + id: 'task-b', + name: 'b', + interrupts: [{ id: 'history-b', value: 0 }], + }, + ], + metadata: null, + created_at: null, + checkpoint: { + thread_id: 't', + checkpoint_id: 'c', + checkpoint_ns: '', + checkpoint_map: {}, + }, + parent_checkpoint: null, + }, + ]; + const trace = [ + [ + 'values', + { type: 'domain', namespace: ['application'], __interrupt__: [item] }, + ], + [ + 'updates', + { + __interrupt__: [ + { id: 'a', value: 'duplicate' }, + { id: 'b', value: 0 }, + ], + }, + ], + ['values|child', { namespace: [], __interrupt__: [{ id: 'ignored' }] }], + ] + .map( + ([event, data]) => `event: ${event}\ndata: ${JSON.stringify(data)}\n\n` + ) + .join(''); + const request = vi.fn(async (url, init) => { + expect(init?.method).toBe('POST'); + if (String(url).endsWith('/history')) { + expect(JSON.parse(String(init?.body))).toEqual({ limit: 10 }); + return new Response(JSON.stringify(history), { + headers: { 'content-type': 'application/json' }, + }); + } + expect(String(url)).toBe('https://runtime.example/threads/t/runs/stream'); + return fragmentedResponse(trace); + }); + vi.stubGlobal('fetch', request); + const session = createSession({ + assistantId: 'a', + threadId: 't', + apiUrl: 'https://runtime.example', + }); + try { + expect(await session.submit('Go')).toBe('paused'); + expect(session.getSnapshot().interrupts).toEqual([ + item, + { id: 'b', value: 0 }, + ]); + expect(request).toHaveBeenCalledTimes(1); + await session.load?.(); + expect(session.getSnapshot().interrupts).toEqual([ + { id: 'history-a', value: false }, + { id: 'history-b', value: 0 }, + ]); + expect(session.getSnapshot().values).toEqual({ count: 2 }); + expect(session.getSnapshot().messages[0].delivery).toMatchObject({ + outcome: 'paused', + }); + expect(request).toHaveBeenCalledTimes(2); + } finally { + await session.dispose(); + } + }); + it('keeps SDK routing authoritative while exposing colliding application keys', async () => { const root = { type: 'domain', diff --git a/libs/langgraph/src/runtime/values.spec.ts b/libs/langgraph/src/runtime/values.spec.ts index c7cec6031..848c82650 100644 --- a/libs/langgraph/src/runtime/values.spec.ts +++ b/libs/langgraph/src/runtime/values.spec.ts @@ -202,7 +202,8 @@ describe('session application values', () => { }); expect(f.snapshot().values).toEqual({ count: 2 }); f.streams[0].finish(); - expect(await run).toBe('success'); + expect(await run).toBe('paused'); + expect(f.snapshot().interrupts).toEqual([{ when: 'breakpoint' }]); }); for (const [label, invalid] of [ @@ -412,9 +413,13 @@ describe('session application values', () => { f.session.subscribe(() => seen.push(f.snapshot())); if (mode === 'close') { f.streams[0].finish(); - expect(await run).toBe('success'); + expect(await run).toBe('paused'); } else await f.session.checkStatus?.(); expect(f.snapshot().values).toEqual({ count: 3 }); + expect(f.snapshot().interrupts).toEqual([{ when: 'breakpoint' }]); + expect(f.snapshot().messages.at(-1)?.delivery).toMatchObject({ + outcome: 'paused', + }); expect(f.snapshot().messages.at(-1)?.content).toBe('Recovered'); expect( seen.every( diff --git a/libs/langgraph/src/runtime/wire-message.ts b/libs/langgraph/src/runtime/wire-message.ts index 48518ee2f..24c8a2c5a 100644 --- a/libs/langgraph/src/runtime/wire-message.ts +++ b/libs/langgraph/src/runtime/wire-message.ts @@ -42,10 +42,3 @@ export function textContent(value: unknown): string { }) .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 8503f1bf5..0b06fb1da 100644 --- a/libs/react/src/use-agent.spec.tsx +++ b/libs/react/src/use-agent.spec.tsx @@ -17,6 +17,24 @@ import { } from '../../langgraph/src/runtime/testing/binding-fixture'; const fixtures: ReturnType[] = []; +const savedInterrupts = [ + { + id: 'saved-approval', + value: { question: 'Approve saved request?', choices: ['yes', 'no'] }, + namespace: ['review', 'task-1'], + when: 'during', + resumable: true, + ns: ['legacy-review'], + }, + { + id: 'saved-confirmation', + value: 0, + namespace: [], + when: 'during', + resumable: false, + ns: [], + }, +]; function fixture() { const value = bindingFixture(); fixtures.push(value); @@ -42,6 +60,12 @@ describe('useAgent borrowed session', () => { {JSON.stringify(snapshot.values) ?? 'unobserved'} + + {JSON.stringify(snapshot.interrupts)} + + + {JSON.stringify(snapshot.messages.at(-1)?.delivery)} + ); } @@ -49,6 +73,7 @@ describe('useAgent borrowed session', () => { expect(f.session.load).toBeTypeOf('function'); expect(f.history.reads).toBe(0); expect(view.getByTestId('values').textContent).toBe('unobserved'); + expect(view.getByTestId('interrupts').textContent).toBe('[]'); await act(async () => { await f.session.load(); }); @@ -60,6 +85,12 @@ describe('useAgent borrowed session', () => { stable: { items: ['saved'] }, }); const snapshot = f.session.getSnapshot(); + expect( + JSON.parse(view.getByTestId('interrupts').textContent ?? '') + ).toEqual(savedInterrupts); + expect(view.getByTestId('history-delivery').textContent).toContain( + 'paused' + ); const beforeRefresh = renders; await act(async () => { await f.session.load(); @@ -73,6 +104,7 @@ describe('useAgent borrowed session', () => { await f.session.load(); }); const refreshed = f.session.getSnapshot(); + expect(refreshed.interrupts).toBe(snapshot.interrupts); expect(refreshed.values?.['stable']).toBe(snapshot.values?.['stable']); expect(JSON.parse(view.getByTestId('values').textContent ?? '')).toEqual({ counter: 2, @@ -90,6 +122,7 @@ describe('useAgent borrowed session', () => { await f.session.load(); expect(f.session.getSnapshot().messages).toEqual([]); expect(f.session.getSnapshot().values).toBeUndefined(); + expect(f.session.getSnapshot().interrupts).toEqual([]); expect(f.history.reads).toBe(4); expect(f.handlerCalls).toBe(0); expect(f.streams).toHaveLength(0); diff --git a/libs/react/src/use-agent.type-test.ts b/libs/react/src/use-agent.type-test.ts index 4d35b3156..619fc2000 100644 --- a/libs/react/src/use-agent.type-test.ts +++ b/libs/react/src/use-agent.type-test.ts @@ -1,5 +1,49 @@ -import type { AgentSession, AgentSnapshot } from '@threadplane/core'; +import type { + AgentSession, + AgentSnapshot, + PlainValue, +} from '@threadplane/core'; import { useAgent } from './index.js'; +// eslint-disable-next-line @nx/enforce-module-boundaries -- This type-only integration probe observes the actual private factory through the borrowed fixture. +import type { bindingFixture } from '../../langgraph/src/runtime/testing/binding-fixture'; + +export function useInferredRuntime( + session: ReturnType['session'] +) { + const snapshot = useAgent(session); + const payload = snapshot.interrupts[0]?.value; + const plainPayload: PlainValue = payload; + const namespace: readonly string[] | undefined = + snapshot.interrupts[0]?.namespace; + // @ts-expect-error The backend interrupt field remains readonly. + snapshot.interrupts = []; + // @ts-expect-error The observed batch remains readonly. + snapshot.interrupts.push({ value: false }); + // @ts-expect-error Namespace metadata remains readonly. + snapshot.interrupts[0].namespace?.push('changed'); + // @ts-expect-error The broad backend payload has no inferred application schema. + const assumed: { approved: boolean } = payload; + if (payload && typeof payload === 'object' && !Array.isArray(payload)) { + // @ts-expect-error Nested payload fields remain readonly. + payload['changed'] = true; + } + for (const call of snapshot.toolCalls) + if (call.status === 'complete') { + if (call.name === 'weather') { + const temperature: number = call.result.temperature; + // @ts-expect-error The weather result retains its authored object shape. + const wrong: number = call.result; + void [temperature, wrong]; + } else { + const count: number = call.result; + // @ts-expect-error The count result retains its authored primitive shape. + void call.result.temperature; + void count; + } + } + void [plainPayload, namespace, assumed]; + return snapshot; +} interface Tools { weather: { args: { city: string }; result: { temperature: number } }; diff --git a/scripts/react-parity/baseline.json b/scripts/react-parity/baseline.json index cca17dc56..8988cbedb 100644 --- a/scripts/react-parity/baseline.json +++ b/scripts/react-parity/baseline.json @@ -1,18 +1,18 @@ { "schemaVersion": 1, - "baselineHead": "192f94e9b57b77e1858e2e4337f0d34bcd6a9c64", + "baselineHead": "db867f83be6cb6f9735de41f6aadddf64a0d9937", "sourceState": { "modified": [ - "libs/langgraph/src/lib/transport/fetch-stream.transport.ts", "libs/langgraph/src/runtime/create-session.ts", + "libs/langgraph/src/runtime/history-projection.ts", + "libs/langgraph/src/runtime/langgraph-snapshot.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/binding-fixture.ts", + "libs/langgraph/src/runtime/wire-message.ts" ], "untracked": [ - "libs/langgraph/src/runtime/langgraph-snapshot.ts", - "libs/langgraph/src/runtime/values-projection.ts" + "libs/langgraph/src/runtime/interrupt-projection.ts" ] }, "scope": { @@ -12568,7 +12568,7 @@ "id": "source:libs/langgraph/src/runtime/create-session.ts", "kind": "source", "path": "libs/langgraph/src/runtime/create-session.ts", - "sha256": "5341e188ac9663a3b5d058602e91dd1ba5c0df2983d264d92de0049a792b22ce" + "sha256": "74f833ad7a0e2fb29b1c6cef5cc1ede6002db61306288b191b1753fdb1e91e5c" }, { "id": "source:libs/langgraph/src/runtime/function-tools.ts", @@ -12580,13 +12580,19 @@ "id": "source:libs/langgraph/src/runtime/history-projection.ts", "kind": "source", "path": "libs/langgraph/src/runtime/history-projection.ts", - "sha256": "f7065276cc2d2122d8cec64311592210b521a9d76cc35550ba94d85d8cbd4742" + "sha256": "4242b06efe381b4c697a05eacde29f98b44964024f2608f697a413ac4762a4ed" + }, + { + "id": "source:libs/langgraph/src/runtime/interrupt-projection.ts", + "kind": "source", + "path": "libs/langgraph/src/runtime/interrupt-projection.ts", + "sha256": "a4dd19e151581237acdbd0d312957b96bec153ed37e85428ab959c802316388e" }, { "id": "source:libs/langgraph/src/runtime/langgraph-snapshot.ts", "kind": "source", "path": "libs/langgraph/src/runtime/langgraph-snapshot.ts", - "sha256": "2c5f72c9c9b5259426bb94a7a1e44e1e81c401e4c9ec7d6724adfd2fb1b7670f" + "sha256": "7ec83cc0d4387fdb8443e543521e5bf955ff31a2817e420956db378534c68c5e" }, { "id": "source:libs/langgraph/src/runtime/message-reducer.ts", @@ -12604,7 +12610,7 @@ "id": "source:libs/langgraph/src/runtime/ownership.ts", "kind": "source", "path": "libs/langgraph/src/runtime/ownership.ts", - "sha256": "eb7692b96d5113ba690c275624280142df905c78c69b6e5ebdf4f788b6fa7811" + "sha256": "f9a2ad835478aaf8867a02cff49a366f590c9a37439b5fbb03e40af2164785bd" }, { "id": "source:libs/langgraph/src/runtime/publication.ts", @@ -12616,13 +12622,13 @@ "id": "source:libs/langgraph/src/runtime/stream-projection.ts", "kind": "source", "path": "libs/langgraph/src/runtime/stream-projection.ts", - "sha256": "ef917d98131c0311846247550ca653d8340f2dcaa97858a5d9a4acaa6bffc1e4" + "sha256": "0ae84a3605e587f8bbcf34d9c2dc5a93071324f5ea2e850afb0a8852b8e9c440" }, { "id": "source:libs/langgraph/src/runtime/testing/binding-fixture.ts", "kind": "source", "path": "libs/langgraph/src/runtime/testing/binding-fixture.ts", - "sha256": "9299cc57319e881aebe684f6ad5daffd8549aef7e62e83c104f94e7f3a6f5c9c" + "sha256": "3f54126919f9d1fd8b5c966ffb6ee3196ef3ea6b605ab09b560857b5a1d7b056" }, { "id": "source:libs/langgraph/src/runtime/testing/controlled-transport.ts", @@ -12652,7 +12658,7 @@ "id": "source:libs/langgraph/src/runtime/wire-message.ts", "kind": "source", "path": "libs/langgraph/src/runtime/wire-message.ts", - "sha256": "0f134e1c7a664cec9df39b16aefae7595b7400c74fcb68c9c0b84b61dc500c13" + "sha256": "df124134df4288905ba9a46d084efc64653a6b092592e43b178297c22bd241f0" }, { "id": "source:libs/langgraph/src/test-setup.ts", diff --git a/scripts/react-parity/dispositions.json b/scripts/react-parity/dispositions.json index 3386f476e..74e2332f8 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": "Explicit history and root live values now publish application data atomically with messages, including conclusive correlated recovery. Read-only fixed-thread T09/T10 subsets only; broader migration and public cutover remain open." + "note": "Adds readonly interrupt batches to the messages/values aggregate, with submit clearing, retained stop/failure/disposal state, candidate ownership and correlated recovery. Observation only; no resume/targeting or whole-task completion." }, { "id": "source:libs/langgraph/src/runtime/function-tools.ts", @@ -11415,7 +11415,18 @@ "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." + "note": "Latest-checkpoint pause delivery now uses the same candidate interrupt batch as the transcript. Task/control precedence is observed without nested-task or next-only inference; broader T10 and resume remain open." + }, + { + "id": "source:libs/langgraph/src/runtime/interrupt-projection.ts", + "taskIds": [ + "T09", + "T10" + ], + "treatment": "internal", + "status": "in-progress", + "reason": "Private staged LangGraph runtime subset; fixture-only and not a neutral public LangGraph root or tarball.", + "note": "Readonly root interrupt accumulation and authoritative checkpoint/history replacement only, including static breakpoint evidence and ID rules. No resume, targeting or full T09/T10 migration claim." }, { "id": "source:libs/langgraph/src/runtime/langgraph-snapshot.ts", @@ -11426,7 +11437,7 @@ "treatment": "internal", "status": "in-progress", "reason": "Private staged LangGraph runtime subset; fixture-only and not a neutral public LangGraph root or tarball.", - "note": "Private broad readonly application-values snapshot extension; undefined distinguishes no current observation from an observed empty record. No application-schema inference or public root export." + "note": "Private readonly values and interrupt snapshot extensions retain SDK interrupt metadata with plain payloads. No application-schema inference, targeting, resume or public root export." }, { "id": "source:libs/langgraph/src/runtime/message-reducer.ts", @@ -11457,7 +11468,7 @@ "treatment": "internal", "status": "in-progress", "reason": "Private staged LangGraph runtime subset; fixture-only and not a neutral public LangGraph root or tarball.", - "note": "Owns immutable plain application values and shares unchanged nested branches; token-only publication reuses owned graphs. Broader execution/publication migration remains open." + "note": "Owns immutable interrupt batches alongside values/messages, preserving unchanged nested identity. Broader execution/publication migration remains open." }, { "id": "source:libs/langgraph/src/runtime/publication.ts", @@ -11477,7 +11488,7 @@ "treatment": "internal", "status": "in-progress", "reason": "Private staged LangGraph runtime subset; fixture-only and not a neutral public LangGraph root or tarball.", - "note": "Uses messageMetadata for delta text semantics only on actual message events while preserving a same-named application value. Broader T09 migration remains open." + "note": "Live pause delivery uses root interrupt evidence shared with the observed batch; values projection and messageMetadata routing remain unchanged. Broader T09 migration remains open." }, { "id": "source:libs/langgraph/src/runtime/testing/binding-fixture.ts", @@ -11488,7 +11499,7 @@ "treatment": "internal", "status": "in-progress", "reason": "Private controlled runtime/binding test helper; excluded from public exports and production packages.", - "note": "Controlled native fixtures cover four history reads, including values-only replacement with unchanged messages, alongside equal/empty refresh. Broader T05/T06 coverage remains open." + "note": "Controlled native history fixtures include two task interrupt payloads, paused delivery, equal identity, values-only sharing and empty replacement. Broader T05/T06 coverage remains open." }, { "id": "source:libs/langgraph/src/runtime/testing/controlled-transport.ts", @@ -11542,7 +11553,7 @@ "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." + "note": "Private wire role/text decoding remains shared; pause extraction now belongs to the dedicated interrupt projector. Broader T09/T10 capabilities remain open." }, { "id": "source:libs/langgraph/src/test-setup.ts", diff --git a/scripts/react-parity/runtime-consumer.mjs b/scripts/react-parity/runtime-consumer.mjs index cfdadb3d1..753424d90 100644 --- a/scripts/react-parity/runtime-consumer.mjs +++ b/scripts/react-parity/runtime-consumer.mjs @@ -25,6 +25,14 @@ 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 savedInterrupts = [ + { id: 'saved-approval', value: { question: 'Approve saved request?', choices: ['yes', 'no'] }, namespace: ['review', 'task-1'], when: 'during', resumable: true, ns: ['legacy-review'] }, + { id: 'saved-confirmation', value: 0, namespace: [], when: 'during', resumable: false, ns: [] }, +]; +const liveInterrupts = [ + { id: 'live-approval', value: { question: 'Approve action?', choices: ['yes', 'no'] }, namespace: ['review', 'live'], when: 'during', resumable: true, ns: ['legacy-live'] }, + { id: 'live-confirmation', value: false, namespace: [], when: 'during', resumable: false, ns: [] }, +]; const savedHistory = [{ values: { stage: 'saved', profile: { name: 'Saved user' }, messages: [ { id: 'saved-human', type: 'human', content: 'Saved question' }, @@ -35,7 +43,9 @@ const savedHistory = [{ { 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: {}, + next: ['review', 'confirmation'], + tasks: savedInterrupts.map((interrupt, index) => ({ id: `saved-task-${index}`, name: index === 0 ? 'review' : 'confirmation', error: null, checkpoint: null, state: null, interrupts: [interrupt] })), + metadata: {}, checkpoint: { thread_id: 'fixture-thread', checkpoint_ns: '', checkpoint_id: 'saved-checkpoint', checkpoint_map: {} }, parent_checkpoint: null, created_at: '2026-09-21T00:00:00Z', @@ -60,8 +70,11 @@ export function runtimeResponse(body) { if (message.content === 'Send') return textTrace.replaceAll('message-parity', `answer-${message.id}`) + sse('values|child', { type: 'values', namespace: [], stage: 'child' }) + sse('updates', { writer: { stage: 'node-update' } }) - + sse('custom', { stage: 'custom' }) - + sse('values', { __interrupt__: [], stage: 'control-envelope' }); + + sse('custom', { __interrupt__: [], stage: 'custom' }) + + sse('values|child', { __interrupt__: [], stage: 'child-control' }); + if (message.content === 'Pause') return sse('values', { stage: 'approval', messages: [message, { type: 'ai', id: `pause-${message.id}`, content: 'Waiting for approvals' }] }) + + sse('values', { __interrupt__: [liveInterrupts[0]], stage: 'control-envelope' }) + + sse('updates', { __interrupt__: [liveInterrupts[1]] }); if (message.content === 'Tool') return sse('values', { messages: [message, toolCall] }); if (message.content === 'Error') return sse('error', { error: 'FixtureFailure', message: 'PRIVATE backend diagnostic' }); if (message.content === 'Hold') return null; @@ -90,7 +103,28 @@ export function installedTypeSource(template, kind) { nested['name'] = 'mutable'; } } - void [directValues, values, assumedCounter];`) + const directInterrupts: readonly { readonly value?: PlainValue }[] = direct.interrupts; + const interrupts = snapshot.interrupts; + const payload = interrupts[0]?.value; + const plainPayload: PlainValue = payload; + const id: string | undefined = interrupts[0]?.id; + const namespace: readonly string[] | undefined = interrupts[0]?.namespace; + const legacyNamespace: readonly string[] | undefined = interrupts[0]?.ns; + const when: string | undefined = interrupts[0]?.when; + const resumable: boolean | undefined = interrupts[0]?.resumable; + // @ts-expect-error The concrete interrupt field remains readonly. + snapshot.interrupts = []; + // @ts-expect-error The batch remains readonly. + interrupts.push({ value: false }); + // @ts-expect-error SDK namespace metadata remains readonly. + interrupts[0].namespace?.push('changed'); + // @ts-expect-error Broad interrupt data does not infer an application schema. + const assumedApproval: { approved: boolean } = payload; + if (payload && typeof payload === 'object' && !Array.isArray(payload)) { + // @ts-expect-error Nested interrupt payloads remain readonly. + payload['approved'] = true; + } + void [directValues, values, assumedCounter, directInterrupts, plainPayload, id, namespace, legacyNamespace, when, resumable, assumedApproval];`) .replace(' assertSnapshot(snapshot);', ' void snapshot;'); } @@ -228,6 +262,7 @@ export async function runRuntimeScenarios(directory, kind) { context = await browser.newContext(); const page = await context.newPage(); const expectValues = (value) => expect(page.getByTestId('values')).toHaveText(value === undefined ? 'unobserved' : JSON.stringify(value)); + const expectInterrupts = (interrupts) => expect(page.getByTestId('interrupts')).toHaveText(JSON.stringify(interrupts)); page.on('pageerror', (error) => pageErrors.push(error.message)); page.on('request', (request) => { if (!request.url().startsWith(`${server.url}/`)) unexpected.push(request.url()); @@ -238,6 +273,7 @@ export async function runRuntimeScenarios(directory, kind) { await expect(page.getByTestId('handler-calls')).toHaveText('0'); await expect(page.getByTestId('submissions')).toHaveText('0'); await expectValues(undefined); + await expectInterrupts([]); 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'); @@ -246,10 +282,11 @@ export async function runRuntimeScenarios(directory, kind) { await expect(page.getByTestId('loads-finished')).toHaveText('1'); await expect(page.getByTestId('load-error')).toHaveText(''); await expectValues({ stage: 'saved', profile: { name: 'Saved user' } }); + await expectInterrupts(savedInterrupts); 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('delivery')).toHaveText('complete:paused'); 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); @@ -261,6 +298,8 @@ export async function runRuntimeScenarios(directory, kind) { await expect(page.getByTestId('loads-finished')).toHaveText('2'); await expect(page.getByTestId('load-error')).toHaveText(''); await expectValues({ stage: 'saved', profile: { name: 'Saved user' } }); + await expectInterrupts(savedInterrupts); + await expect(page.getByTestId('delivery')).toHaveText('complete:paused'); 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); @@ -271,6 +310,7 @@ export async function runRuntimeScenarios(directory, kind) { await expect(page.getByTestId('loads-finished')).toHaveText('3'); await expect(page.getByTestId('load-error')).toHaveText(''); await expectValues(undefined); + await expectInterrupts([]); await expect(page.getByTestId('text')).toHaveText(''); await expect(page.getByTestId('transcript')).toHaveText(''); await expect(page.getByTestId('tool')).toHaveText('[]'); @@ -285,6 +325,7 @@ export async function runRuntimeScenarios(directory, kind) { await expect(page.getByTestId('status')).toHaveText('idle'); assert.equal(server.requests.length, 1); await expectValues({ stage: 'complete' }); + await expectInterrupts([]); completed.push('text success'); const beforeTool = server.requests.length; @@ -319,14 +360,29 @@ export async function runRuntimeScenarios(directory, kind) { await expectValues({ stage: 'held', transient: true }); completed.push('incremental DOM update and stop abort'); + await page.getByRole('button', { name: 'Pause', exact: true }).click(); + await expect(page.getByTestId('delivery')).toHaveText('complete:paused'); + await expect(page.getByTestId('status')).toHaveText('idle'); + await expect(page.getByTestId('text')).toContainText('Waiting for approvals'); + await expectValues({ stage: 'approval' }); + await expectInterrupts(liveInterrupts); + await expect(page.getByTestId('handler-calls')).toHaveText('1'); + assert.equal(server.requests.length, 6, 'pause creates exactly one run with no tool continuation'); + await page.getByRole('button', { name: 'Stop', exact: true }).click(); + await expectInterrupts(liveInterrupts); + await expect(page.getByTestId('delivery')).toHaveText('complete:paused'); + assert.equal(server.requests.length, 6, 'stop after completed pause does not create backend I/O'); + completed.push('full interrupt batch and retained pause after stop'); + await page.getByRole('button', { name: 'Send', exact: true }).click(); await expect(page.getByTestId('delivery')).toHaveText('complete:success'); await expect(page.getByTestId('status')).toHaveText('idle'); await expect(page.getByTestId('text')).toContainText('Hello 🌍.'); await expect(page.getByTestId('handler-calls')).toHaveText('1'); - assert.equal(server.requests.length, 6); + assert.equal(server.requests.length, 7); await expectValues({ stage: 'complete' }); - await expect(page.getByTestId('submissions')).toHaveText('5'); + await expectInterrupts([]); + await expect(page.getByTestId('submissions')).toHaveText('6'); completed.push('reuse after stop'); await page.getByRole('button', { name: 'Unmount', exact: true }).click(); @@ -336,13 +392,13 @@ export async function runRuntimeScenarios(directory, kind) { await expect(page.getByTestId('owner')).toHaveText('disposed'); 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.equal(server.requests.length, 7, '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('; ')}); 3 exact history reads, 6 exact run requests, one tool handler, no page errors/unexpected requests.`); + console.log(`${kind}: ${completed.length} browser scenarios passed (${completed.join('; ')}); 3 exact history reads, 7 exact run requests, one tool handler, 6 component submissions, 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 d3a2ac2d4..6a8126f8c 100644 --- a/scripts/react-parity/runtime-consumer.spec.mjs +++ b/scripts/react-parity/runtime-consumer.spec.mjs @@ -85,6 +85,10 @@ test('history uses the exact SDK body and counts reads separately from runs', as const saved = await first.json(); assert.equal(saved[0].values.stage, 'saved'); assert.deepEqual(saved[0].values.profile, { name: 'Saved user' }); + assert.deepEqual(saved[0].tasks.flatMap((task) => task.interrupts), [ + { id: 'saved-approval', value: { question: 'Approve saved request?', choices: ['yes', 'no'] }, namespace: ['review', 'task-1'], when: 'during', resumable: true, ns: ['legacy-review'] }, + { id: 'saved-confirmation', value: 0, namespace: [], when: 'during', resumable: false, ns: [] }, + ]); 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(), []); @@ -94,7 +98,7 @@ test('history uses the exact SDK body and counts reads separately from runs', as } finally { await server.close(); } }); -test('text fixture retains root application state while exercising ignored child and control data', () => { +test('text fixture retains root application state while exercising only ignored interrupt lookalikes', () => { const body = { ...heldBody, input: { ...heldBody.input, messages: [{ id: 'user', type: 'human', content: 'Send' }] } }; const trace = runtime.runtimeResponse(body); assert.match(trace, /"stage":"complete"/); @@ -102,6 +106,22 @@ test('text fixture retains root application state while exercising ignored child assert.match(trace, /event: updates/); assert.match(trace, /event: custom/); assert.match(trace, /"__interrupt__":\[\]/); + const events = trace.trim().split('\n\n').map((event) => ({ type: event.match(/^event: (.*)/m)?.[1], data: JSON.parse(event.match(/^data: (.*)/m)?.[1] ?? '{}') })); + assert.ok(events.every((event) => event.type !== 'values' || !Object.hasOwn(event.data, '__interrupt__'))); +}); + +test('Pause contains separate values and updates controls with full SDK payloads and retained application values', () => { + const body = { ...heldBody, input: { ...heldBody.input, messages: [{ id: 'pause-user', type: 'human', content: 'Pause' }] } }; + const trace = runtime.runtimeResponse(body); + const events = trace.trim().split('\n\n').map((event) => ({ type: event.match(/^event: (.*)/m)?.[1], data: JSON.parse(event.match(/^data: (.*)/m)?.[1] ?? '{}') })); + assert.deepEqual(events.map((event) => event.type), ['values', 'values', 'updates']); + assert.equal(events[0].data.stage, 'approval'); + assert.equal(events[0].data.messages.at(-1).content, 'Waiting for approvals'); + assert.deepEqual(events.slice(1).flatMap((event) => event.data.__interrupt__), [ + { id: 'live-approval', value: { question: 'Approve action?', choices: ['yes', 'no'] }, namespace: ['review', 'live'], when: 'during', resumable: true, ns: ['legacy-live'] }, + { id: 'live-confirmation', value: false, namespace: [], when: 'during', resumable: false, ns: [] }, + ]); + assert.equal(events[1].data.stage, 'control-envelope'); }); for (const [label, route, method, body] of [