diff --git a/apps/website/content/docs/ag-ui/getting-started/installation.mdx b/apps/website/content/docs/ag-ui/getting-started/installation.mdx index 10d5b73a4..0661b1586 100644 --- a/apps/website/content/docs/ag-ui/getting-started/installation.mdx +++ b/apps/website/content/docs/ag-ui/getting-started/installation.mdx @@ -9,7 +9,7 @@ ## Install packages ```bash -npm install @threadplane/chat @threadplane/ag-ui @ag-ui/client +npm install @threadplane/chat @threadplane/ag-ui @ag-ui/client @ag-ui/core ``` `@threadplane/chat` provides the chat UI primitives. `@threadplane/ag-ui` provides the adapter that wires an AG-UI backend into the `Agent` contract those primitives consume. @@ -21,9 +21,9 @@ npm install @threadplane/chat @threadplane/ag-ui @ag-ui/client | Package | Version | |---|---| | `@threadplane/chat` | `*` | -| `@threadplane/licensing` | `*` | | `@angular/core` | `^20.0.0 \|\| ^21.0.0` | | `@ag-ui/client` | `*` | +| `@ag-ui/core` | `*` | | `rxjs` | `~7.8.0` | ## Configure the provider diff --git a/apps/website/content/docs/ag-ui/getting-started/quickstart.mdx b/apps/website/content/docs/ag-ui/getting-started/quickstart.mdx index 4187e5882..f0a054511 100644 --- a/apps/website/content/docs/ag-ui/getting-started/quickstart.mdx +++ b/apps/website/content/docs/ag-ui/getting-started/quickstart.mdx @@ -14,7 +14,7 @@ Want to see the finished result before you build? Open the live [AG-UI demo](htt ```bash -npm install @threadplane/chat @threadplane/ag-ui @ag-ui/client +npm install @threadplane/chat @threadplane/ag-ui @ag-ui/client @ag-ui/core ``` @@ -77,7 +77,7 @@ So swapping backends is a one-line change in `app.config.ts`, and the component ```diff - import { provideAgent } from '@threadplane/langgraph'; -- providers: [provideAgent({ apiUrl: '...' })], // LangGraph +- providers: [provideAgent({ apiUrl: '...', assistantId: '...' })], // LangGraph + import { provideAgent } from '@threadplane/ag-ui'; + providers: [provideAgent({ url: '...' })], // AG-UI ``` diff --git a/apps/website/content/docs/ag-ui/guides/interrupts.mdx b/apps/website/content/docs/ag-ui/guides/interrupts.mdx index feace8831..fa852cd2b 100644 --- a/apps/website/content/docs/ag-ui/guides/interrupts.mdx +++ b/apps/website/content/docs/ag-ui/guides/interrupts.mdx @@ -39,7 +39,7 @@ decision = interrupt({ `injectAgent()` exposes a `interrupt()` signal that is populated whenever the adapter receives an `on_interrupt` CUSTOM event. Pair it with `` from `@threadplane/chat` to render an approval dialog without manual event wiring: ```typescript -import { Component } from '@angular/core'; +import { ChangeDetectionStrategy, Component } from '@angular/core'; import { ChatComponent, ChatApprovalCardComponent } from '@threadplane/chat'; import { injectAgent } from '@threadplane/ag-ui'; import type { ChatApprovalAction } from '@threadplane/chat'; diff --git a/apps/website/content/docs/chat/a2ui/surface-component.mdx b/apps/website/content/docs/chat/a2ui/surface-component.mdx index d1b9f204a..bef6cd45b 100644 --- a/apps/website/content/docs/chat/a2ui/surface-component.mdx +++ b/apps/website/content/docs/chat/a2ui/surface-component.mdx @@ -39,7 +39,7 @@ The `events` output forwards all `RenderEvent` emissions from the render engine. **1. Convert surface to Spec (`surfaceToSpec`)** -The `surfaceToSpec()` function walks the flat component map and produces a json-render `Spec`. It requires a component with `id: 'root'` to be present — if no root exists, nothing renders. +The internal surface-to-spec conversion walks the flat component map and produces a json-render `Spec`. It uses a component with `id: 'root'` when present; otherwise it uses the first component in the map as the render root. Empty component maps render nothing. **2. Resolve dynamic values** @@ -52,7 +52,7 @@ Before the spec is emitted, each component prop is evaluated against the surface **3. Map actions to `on` bindings** -`surfaceToSpec()` converts each component's A2UI `action` prop into a render-spec `on` binding on the corresponding element. Event actions map to the `a2ui:event` handler, and function call actions map to the `a2ui:localAction` handler. This bridges the A2UI interaction model to the render-lib event system. +The internal conversion maps each component's A2UI `action` prop into a render-spec `on` binding on the corresponding element. Event actions map to the `a2ui:event` handler, and function call actions map to the `a2ui:localAction` handler. This bridges the A2UI interaction model to the render-lib event system. **4. Expand template children** @@ -94,8 +94,8 @@ export class AgentPanelComponent { } ``` - -The surface must contain a component with `id: 'root'`. If no root component has been received yet (for example, while the agent is still streaming), `A2uiSurfaceComponent` renders nothing until one arrives. + +`A2uiSurfaceComponent` prefers a component with `id: 'root'` when one exists. If the surface has components but no `root`, it renders from the first component ID. Empty component maps render nothing. ### Wiring handlers and actions @@ -149,23 +149,11 @@ export class InteractiveSurfaceComponent { } ``` -## surfaceToSpec() +## Internal conversion -The conversion function is exported for testing or custom rendering pipelines. +`A2uiSurfaceComponent` handles surface-to-spec conversion internally when it receives a legacy `surface` input. That conversion returns no rendered output for an empty component map. Otherwise it produces a complete json-render `Spec`, preferring a `root` component when present and falling back to the first component ID. -**Import:** - -```typescript -import { surfaceToSpec } from '@threadplane/chat'; -``` - -**Signature:** - -```typescript -function surfaceToSpec(surface: A2uiSurface): Spec | null -``` - -Returns `null` when the surface has no `root` component. Otherwise returns a complete json-render `Spec` with all dynamic values resolved against the current `dataModel`. +The conversion helper is not part of the public `@threadplane/chat` API. Use `A2uiSurfaceComponent` or `createA2uiSurfaceStore()` for supported integration points. ## What's Next diff --git a/apps/website/content/docs/chat/api/provide-chat.mdx b/apps/website/content/docs/chat/api/provide-chat.mdx index 672b6b02b..331fa1116 100644 --- a/apps/website/content/docs/chat/api/provide-chat.mdx +++ b/apps/website/content/docs/chat/api/provide-chat.mdx @@ -83,7 +83,7 @@ import { provideChat } from '@threadplane/chat'; export const appConfig: ApplicationConfig = { providers: [ - provideAgent({ apiUrl: 'http://localhost:2024' }), + provideAgent({ apiUrl: 'http://localhost:2024', assistantId: 'chat' }), provideChat({ avatarLabel: 'B', assistantName: 'Bot', diff --git a/apps/website/content/docs/chat/getting-started/installation.mdx b/apps/website/content/docs/chat/getting-started/installation.mdx index 845659b56..56b2b2919 100644 --- a/apps/website/content/docs/chat/getting-started/installation.mdx +++ b/apps/website/content/docs/chat/getting-started/installation.mdx @@ -42,7 +42,7 @@ npm install @threadplane/chat @threadplane/ag-ui marked `marked` is a required peer dependency used to render assistant message markdown (code blocks, tables, headings). The chat components ship with their own design tokens and component-scoped styles — no Tailwind, PostCSS, or global stylesheet import is required. -`@threadplane/chat` declares peers on `@angular/core`, `@angular/common`, `@angular/forms`, `@angular/platform-browser` (all `^20.0.0 || ^21.0.0`), plus `@threadplane/licensing`, `@threadplane/render`, `@threadplane/a2ui`, `@json-render/core` (`^0.16.0`), `@langchain/core` (`^1.1.33`), `rxjs` (`~7.8.0`), and `marked` (`^15 || ^16`). npm 7+ installs all of these automatically. +`@threadplane/chat` declares peers on `@angular/core`, `@angular/common`, `@angular/platform-browser`, `@angular/router` (all `^20.0.0 || ^21.0.0`), plus `@threadplane/licensing`, `@threadplane/render`, `@threadplane/a2ui`, `@json-render/core` (`^0.16.0`), `@langchain/core` (`^1.1.33`), `rxjs` (`~7.8.0`), `marked` (`^15 || ^16`), `zod` (`^3.25.0`), and optional `katex` (`^0.16.0 || ^0.17.0`). npm 7+ installs all required peers automatically. ## 2. Add your license token diff --git a/apps/website/content/docs/chat/getting-started/quickstart.mdx b/apps/website/content/docs/chat/getting-started/quickstart.mdx index 1b5e355d9..026ea9126 100644 --- a/apps/website/content/docs/chat/getting-started/quickstart.mdx +++ b/apps/website/content/docs/chat/getting-started/quickstart.mdx @@ -18,10 +18,10 @@ Evaluating? No license needed — the chat runs without a token (with a one-time ```bash -npm install @threadplane/chat marked +npm install @threadplane/chat @threadplane/langgraph marked ``` -`marked` is the markdown renderer used for assistant messages. +`@threadplane/langgraph` provides the adapter used by the provider snippet below. `marked` is the markdown renderer used for assistant messages. diff --git a/docs/superpowers/plans/2026-07-06-langgraph-delta-merge-identity.md b/docs/superpowers/plans/2026-07-06-langgraph-delta-merge-identity.md new file mode 100644 index 000000000..06a342e50 --- /dev/null +++ b/docs/superpowers/plans/2026-07-06-langgraph-delta-merge-identity.md @@ -0,0 +1,408 @@ +# LangGraph Identity-Based Delta Merge Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Stop the messages-tuple merge from silently dropping streamed delta chunks (proven: bare `|` tokens in tables) by merging according to declared event kind instead of string-prefix heuristics. + +**Architecture:** `mergeMessages`/`accumulateContent` in `stream-manager.bridge.ts` gain a `mode` parameter. Tuple events (`event.messageMetadata`) merge as **deltas**: unconditional append, no prefix guards, plus a `canonicalMessageIds` backstop (ids whose final text is known are immune to late deltas). `messages/partial` and values-sync keep today's **snapshot** semantics unchanged (mutual-prefix reconcile — correct for snapshot-shaped payloads, including lagging mid-run values events). One file changes; TDD through the existing bridge spec harness; a live Chrome MCP gate verifies wire==accumulation before the PR. + +**Tech Stack:** TypeScript, Vitest (`MockAgentTransport` harness), Angular workspace lib `libs/langgraph`; Chrome MCP for the live gate. + +**Spec:** [docs/superpowers/specs/2026-07-06-langgraph-delta-merge-identity-design.md](../specs/2026-07-06-langgraph-delta-merge-identity-design.md) + +**Branch:** create `fix/langgraph-delta-merge-identity` off main. NOTE: `libs/langgraph/src/lib/transport/fetch-stream.transport.ts` currently carries an uncommitted TEMP DEBUG recorder block (added during diagnosis) — keep it in the working tree until Task 4's live gate passes, then revert it in Task 4 before the PR. Do not commit it. + +--- + +## File Structure + +- Modify: `libs/langgraph/src/lib/internals/stream-manager.bridge.ts` — mode threading, delta path, canonical-id set. +- Modify: `libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts` — new `describe` block, 6 tests. +- Revert (Task 4): `libs/langgraph/src/lib/transport/fetch-stream.transport.ts` — remove the TEMP DEBUG recorder. + +## Current-code anchors (verify before editing; line numbers approximate) + +- `processEvent` messages branch, tuple/partial merge call (~503): `if (event.type === 'messages/partial' || event.messageMetadata) { subjects.messages$.next(mergeMessages(subjects.messages$.value, normalized, reasoningTimingMap)); … }` +- values-sync merge call (~572): `subjects.messages$.next(mergeMessages(subjects.messages$.value, remapped, reasoningTimingMap));` +- `runStream` per-run reset block (~388-397): `subjects.custom$.next([]); subjects.toolProgress$.next([]); toolProgressMap.clear();` +- `mergeMessages` (~1032): signature `(existing, incoming, reasoningTimingMap?)`; merge body computes `accumulatedContent = accumulateContent(existing.content, incomingRaw['content'])`. +- `accumulateContent` (~1169): the two `startsWith` guards + `isFinalCanonicalReasoningContent` + append. +- Thread switch: `switchThread` clears per-thread state (find `function switchThread` / the `threadId$` subscription that resets state). + +--- + +### Task 1: Failing tests — delta merge preserves every chunk + +**Files:** +- Modify: `libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts` (append a new describe block) + +- [ ] **Step 1: Create the branch** + +```bash +cd ~/repos/angular-agent-framework +git stash push -m tmp-recorder libs/langgraph/src/lib/transport/fetch-stream.transport.ts +git checkout main && git pull --ff-only +git checkout -b fix/langgraph-delta-merge-identity +git stash pop +``` +(The stash carries the uncommitted TEMP DEBUG recorder onto the new branch's working tree without committing it.) + +- [ ] **Step 2: Write the failing tests** + +Append to `libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts` (reuse the file's existing `makeSubjects()` and `MockAgentTransport` imports; tuple events are emitted post-normalization as `{ type: 'messages', data: [chunk, metadata], messageMetadata: metadata }` — mirror how existing tests in this file construct events, and confirm the exact tuple shape by reading a passing `messageMetadata` test in the file before writing): + +```ts +describe('identity-based delta merge (messages-tuple)', () => { + const META = { langgraph_node: 'chatbot' }; + + function setup() { + const transport = new MockAgentTransport(); + const subjects = makeSubjects(); + const destroy$ = new Subject(); + const bridge = createStreamManagerBridge({ + options: { apiUrl: '', assistantId: 'test', transport }, + subjects, + threadId$: of(null), + destroy$: destroy$.asObservable(), + }); + return { transport, subjects, destroy$, bridge }; + } + + function tupleEvent(id: string, content: unknown) { + return { + type: 'messages', + data: [{ id, type: 'ai', content }, META], + messageMetadata: META, + } as any; + } + + function lastAiContent(subjects: ReturnType): string { + const msgs = subjects.messages$.value as Array<{ content?: unknown }>; + const last = msgs[msgs.length - 1]; + return typeof last?.content === 'string' ? last.content : JSON.stringify(last?.content); + } + + it('preserves every delta chunk, including ones that prefix the message (table pipes)', async () => { + const { transport, subjects, destroy$, bridge } = setup(); + bridge.submit({}); + const deltas = ['|', ' Gem', ' |', ' Color', ' |', '\n', '|', '---', '|', '---', '|', '\n', '|', ' Ruby', ' |', ' red', ' |']; + for (const d of deltas) transport.emit([tupleEvent('ai-1', d)]); + transport.close(); + await new Promise(r => setTimeout(r, 10)); + expect(lastAiContent(subjects)).toBe(deltas.join('')); + destroy$.next(); + }); + + it('appends a multi-char delta that begins with the accumulated text', async () => { + const { transport, subjects, destroy$, bridge } = setup(); + bridge.submit({}); + transport.emit([tupleEvent('ai-1', '|')]); + transport.emit([tupleEvent('ai-1', '| Gem')]); // delta, NOT a superset echo + transport.close(); + await new Promise(r => setTimeout(r, 10)); + expect(lastAiContent(subjects)).toBe('|| Gem'); + destroy$.next(); + }); + + it('final canonical reasoning+text array replaces the accumulation and blocks late deltas', async () => { + const { transport, subjects, destroy$, bridge } = setup(); + bridge.submit({}); + transport.emit([tupleEvent('ai-1', '| a |')]); + transport.emit([tupleEvent('ai-1', ' | b |')]); + // authoritative final shape: reasoning + text blocks in one array + transport.emit([tupleEvent('ai-1', [ + { type: 'reasoning', text: 'thought' }, + { type: 'text', text: '| a | | b | done' }, + ])]); + // straggler token after the canonical content landed — must be ignored + transport.emit([tupleEvent('ai-1', '|')]); + transport.close(); + await new Promise(r => setTimeout(r, 10)); + expect(lastAiContent(subjects)).toBe('| a | | b | done'); + destroy$.next(); + }); + + it('a new run resets canonical marking (same-id deltas accumulate again)', async () => { + const { transport, subjects, destroy$, bridge } = setup(); + bridge.submit({}); + transport.emit([tupleEvent('ai-1', [ + { type: 'reasoning', text: 'r' }, + { type: 'text', text: 'final one' }, + ])]); + transport.close(); + await new Promise(r => setTimeout(r, 10)); + // second run: same transport mock, fresh runStream + bridge.submit({}); + transport.emit([tupleEvent('ai-2', 'fresh')]); + transport.emit([tupleEvent('ai-2', ' text')]); + transport.close(); + await new Promise(r => setTimeout(r, 10)); + expect(lastAiContent(subjects)).toBe('fresh text'); + destroy$.next(); + }); + + it('messages/partial snapshots still reconcile by prefix (regression)', async () => { + const { transport, subjects, destroy$, bridge } = setup(); + bridge.submit({}); + const partial = (content: string) => ({ + type: 'messages/partial', + data: [{ id: 'ai-1', type: 'ai', content }], + } as any); + transport.emit([partial('| Gem')]); + transport.emit([partial('| Gem | Color |')]); // superset → replace + transport.emit([partial('| Gem')]); // stale shorter snapshot → keep longer + transport.close(); + await new Promise(r => setTimeout(r, 10)); + expect(lastAiContent(subjects)).toBe('| Gem | Color |'); + destroy$.next(); + }); + + it('values-sync mid-run keeps snapshot semantics (lagging state does not rewind)', async () => { + const { transport, subjects, destroy$, bridge } = setup(); + bridge.submit({}); + transport.emit([tupleEvent('ai-1', '| a | b |')]); + transport.emit([tupleEvent('ai-1', ' | c |')]); + // lagging values event: state.messages carries a shorter version of ai-1 + transport.emit([{ + type: 'values', + data: { messages: [ + { id: 'h-1', type: 'human', content: 'hi' }, + { id: 'ai-1', type: 'ai', content: '| a | b |' }, + ] }, + } as any]); + transport.close(); + await new Promise(r => setTimeout(r, 10)); + expect(lastAiContent(subjects)).toBe('| a | b | | c |'); + destroy$.next(); + }); +}); +``` + +- [ ] **Step 3: Run to verify the right ones fail** + +Run: `npx vitest run --config libs/langgraph/vite.config.mts stream-manager.bridge 2>&1 | tail -20` (or `npx nx test langgraph -- stream-manager.bridge` — check `libs/langgraph/project.json` for the test target config first). +Expected: tests 1 and 2 FAIL on current code (bare `|` deltas dropped → `'| Gem | Color |\n|---…'` missing pipes; `'| Gem'` replaces instead of appending → `'| Gem'` not `'|| Gem'`). Test 3's straggler assertion may pass accidentally today (the old guard drops it) — fine. Tests 5 and 6 must PASS today (regression baselines). If test 6 fails on current code, STOP and report — the values-sync baseline assumption is wrong and the plan needs revision, not force-fitting. + +- [ ] **Step 4: Commit the red tests** + +```bash +git add libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts +git commit -m "test(langgraph): red tests — delta merge must preserve prefix-coinciding chunks" +``` + +--- + +### Task 2: Implement mode-aware merge + canonical-id backstop + +**Files:** +- Modify: `libs/langgraph/src/lib/internals/stream-manager.bridge.ts` + +- [ ] **Step 1: Add the mode type + canonical set** + +Near the top of `createStreamManagerBridge` (beside the other per-run state like `toolProgressMap` — find `const toolProgressMap` and declare alongside): + +```ts + // Message ids whose content is known-final (installed by a canonical + // replacement). Late streamed deltas for these ids are stale stragglers and + // are ignored — decided by identity, never by comparing text to text. + const canonicalMessageIds = new Set(); +``` + +In `runStream`, in the per-run reset block (immediately after `toolProgressMap.clear();`): + +```ts + canonicalMessageIds.clear(); +``` + +Also clear it wherever thread switching resets per-thread state (the same place other maps are cleared in `switchThread`/the thread-change path — read that function and mirror it). + +- [ ] **Step 2: Thread the mode through the two call sites** + +Define the type near `mergeMessages`: + +```ts +type MergeMode = 'delta' | 'snapshot'; +``` + +At the tuple/partial call site (~503), split by event kind: + +```ts + if (event.type === 'messages/partial' || event.messageMetadata) { + const mode: MergeMode = event.messageMetadata ? 'delta' : 'snapshot'; + subjects.messages$.next(mergeMessages(subjects.messages$.value, normalized, reasoningTimingMap, mode, canonicalMessageIds)); +``` + +At the values-sync call site (~572), keep snapshot semantics explicitly: + +```ts + subjects.messages$.next(mergeMessages(subjects.messages$.value, remapped, reasoningTimingMap, 'snapshot', canonicalMessageIds)); +``` + +- [ ] **Step 3: Extend `mergeMessages`** + +Change the signature: + +```ts +function mergeMessages( + existing: BaseMessage[], + incoming: BaseMessage[], + reasoningTimingMap?: Map, + mode: MergeMode = 'snapshot', + canonicalMessageIds?: Set, +): BaseMessage[] { +``` + +Inside the `if (idx >= 0)` merge branch, BEFORE computing `accumulatedContent`, add the identity backstop: + +```ts + const targetId = (existingId ?? incomingRaw['id']) as string | undefined; + // Identity backstop: once a message's content is known-final, late + // streamed deltas for it are stale stragglers — ignore them outright. + if (mode === 'delta' && targetId && canonicalMessageIds?.has(targetId) + && !isFinalCanonicalReasoningContent(incomingRaw['content'])) { + continue; + } +``` + +Pass the mode into the content merge: + +```ts + const accumulatedContent = accumulateContent( + existing.content as unknown, + incomingRaw['content'], + mode, + ); +``` + +And after computing it, mark canonical when the final shape just landed: + +```ts + if (targetId && isFinalCanonicalReasoningContent(incomingRaw['content'])) { + canonicalMessageIds?.add(targetId); + } +``` + +(NOTE: `existingId` is already declared in this branch — reuse it; do not redeclare. Adjust placement so `targetId` is defined once before both uses.) + +- [ ] **Step 4: Make `accumulateContent` mode-aware** + +```ts +function accumulateContent(existing: unknown, incoming: unknown, mode: MergeMode = 'snapshot'): string { + const existingText = extractText(existing); + const incomingText = extractText(incoming); + + if (existingText.length === 0) return incomingText; + if (incomingText.length === 0) return existingText; + // Final-canonical detection applies in both modes: the authoritative + // "reasoning + text" array replaces whatever was accumulated. + if (isFinalCanonicalReasoningContent(incoming)) return incomingText; + if (mode === 'delta') { + // Tuple chunks are declared deltas. Append unconditionally — any + // text-comparison "dedupe" here can silently drop legitimate tokens + // that coincide with the message prefix (e.g. every bare "|" in a + // markdown table). Staleness is handled by identity in mergeMessages. + return existingText + incomingText; + } + // Snapshot mode (messages/partial, values-sync): payloads carry the + // message-so-far, so mutual prefix comparison picks the longer state and + // ignores stale shorter snapshots. + if (incomingText.startsWith(existingText)) return incomingText; + if (existingText.startsWith(incomingText)) return existingText; + return existingText + incomingText; +} +``` + +(Keep the existing doc comment above the function, updating it to describe the two modes. Note the `isFinalCanonicalReasoningContent` check moved BEFORE the prefix guards — verify no existing test depended on the old ordering; the suite run in Step 5 confirms.) + +- [ ] **Step 5: Run the new suite + full lib** + +Run: `npx nx test langgraph --skip-nx-cache 2>&1 | tail -6` +Expected: all 6 new tests pass; zero regressions in the existing bridge/agent suites. If an existing test fails, read it before touching anything — it encodes a behavior decision; reconcile deliberately (most likely candidates: tests exercising the old guard's replacement ordering). + +- [ ] **Step 6: Lint + chat suite (consumer)** + +Run: `npx nx lint langgraph && npx nx test chat --skip-nx-cache 2>&1 | tail -4` +Expected: lint 0 errors; chat suite green. + +- [ ] **Step 7: Commit** + +```bash +git add libs/langgraph/src/lib/internals/stream-manager.bridge.ts +git commit -m "fix(langgraph): merge streamed chunks by declared event kind, not text prefixes + +Tuple events are deltas: append unconditionally. The old prefix-based dedupe +guard dropped any delta that coincidentally prefixed the accumulated message — +every bare pipe token in a streamed markdown table — silently corrupting +content mid-stream (wire had 48 pipes, accumulation 35; replaying captured +events through the guard reproduced the corruption byte-identically). +Snapshot payloads (messages/partial, values-sync) keep prefix reconciliation. +Stale post-final stragglers are now suppressed by canonical message id, not +by comparing text to text. + +Co-Authored-By: Claude Opus 4.8 (1M context) " +``` + +--- + +### Task 3: Cross-suite gates + +- [ ] **Step 1: Full affected suites** + +Run: `npx nx run-many -t test -p langgraph chat ag-ui --skip-nx-cache 2>&1 | tail -8` +Expected: all green (ag-ui included as a sibling consumer of @threadplane/chat contracts). + +- [ ] **Step 2: Builds** + +Run: `npx nx run-many -t build -p langgraph chat 2>&1 | tail -4` +Expected: both build clean. + +--- + +### Task 4: Live Chrome MCP verification gate (controller-run, REQUIRED) + +Performed by the controller session (needs Chrome MCP + OPENAI key). The TEMP DEBUG recorder in `fetch-stream.transport.ts` is still in the working tree — it powers this gate. + +- [ ] **Step 1: Serve the fix** — kill stale servers on :4200/:2024, then start backend and frontend as separate detached daemons (double-fork pattern): `(cd examples/chat/python && nohup uv run langgraph dev --port 2024 --no-browser > /tmp/lg-backend.log 2>&1 &)` and `(nohup npx nx run examples-chat-angular:serve --port 4200 > /tmp/ng-frontend.log 2>&1 &)`. Clear `.angular/cache/*/examples-chat-angular/vite` first so the dep cache can't serve stale lib code; wait for `Application bundle generation complete` AND port 200s. + +- [ ] **Step 2: Run ≥6 live table prompts** via Chrome MCP at `http://localhost:4200/embed` with the 3-layer capture: `window.__lgEvents` (SDK recorder), content progression via `ng.getComponent().content()` on MutationObserver, DOM table/paragraph counts. + +- [ ] **Step 3: Assert per run** — `sdkPipes === accumPipes` (wire == accumulation), no non-prefix content divergence at stream end, zero frames where the table collapses to raw-pipe paragraphs (excluding the pre-first-closed-cell frame). If ANY run fails: STOP, capture the progression, return to root-cause — do not proceed to the PR. + +- [ ] **Step 4: Revert the temp recorder** + +```bash +git checkout -- libs/langgraph/src/lib/transport/fetch-stream.transport.ts +git status --porcelain # expect: clean except committed work +``` + +- [ ] **Step 5: Shut down the servers** — kill by PID from `lsof -ti :4200 :2024`; verify both ports free (kill children repeatedly if the process tree respawns). + +--- + +### Task 5: PR + merge on green + +- [ ] **Step 1: Push + PR** + +```bash +git push -u origin fix/langgraph-delta-merge-identity +gh pr create --title "fix(langgraph): stop streamed chunks being dropped by prefix-based dedupe" --fill +``` +Body: root cause (prefix guard vs. delta semantics), the byte-identical replay proof, the identity-based design, unit + live-gate evidence (numbers from Task 4). End with the standard Claude Code attribution. No external project references. + +- [ ] **Step 2: Merge on green** + +```bash +gh pr checks --watch # required: Vercel – threadplane; also confirm Library + chat e2e advisory +gh pr merge --squash --delete-branch +``` +Admin fallback (`--admin`) only for the known stale-cache/behind quirk with green head checks. + +- [ ] **Step 3: Verify main** — `git checkout main && git pull --ff-only && gh run list --branch main --limit 3` (ignore the pre-existing non-required PostHog red). + +--- + +## Out of scope (do not implement) + +- `subagent-tracker.ts`'s `mergeMessages` (append/id-based already). +- Stream resumability, SDK retry tuning, server changes. +- Any partial-markdown or chat rendering changes. diff --git a/docs/superpowers/specs/2026-07-06-langgraph-delta-merge-identity-design.md b/docs/superpowers/specs/2026-07-06-langgraph-delta-merge-identity-design.md new file mode 100644 index 000000000..0148f386c --- /dev/null +++ b/docs/superpowers/specs/2026-07-06-langgraph-delta-merge-identity-design.md @@ -0,0 +1,83 @@ +# LangGraph delta merge: identity-based reconciliation — Design + +**Date:** 2026-07-06 +**Status:** Approved, ready for implementation plan +**Repo:** `~/repos/angular-agent-framework` (`libs/langgraph`) +**Relates to:** the streaming-table work (partial-markdown 0.5.2/0.5.3, chat #743/#744). This fixes the remaining intermittent "table collapses to raw text mid-stream" — which turned out to be silent data loss in the message accumulation, not a rendering issue. + +## Problem + +While an assistant message streams over the LangGraph transport, the accumulated content intermittently **loses characters**. Live capture (examples/chat, gpt-5-mini, table prompts): the wire delivered a delimiter row `|--------------|--------|-------------|` (48 pipes across the message) but the accumulated content rendered `-----------------------------------|` (35 pipes). The final message (values-sync) is always correct, so the UI self-heals at stream end — mid-stream, tables collapse to raw pipe text (invalid GFM) until then. + +### Root cause (proven) + +`accumulateContent` in `libs/langgraph/src/lib/internals/stream-manager.bridge.ts` (~line 1183) merges each incoming messages-tuple chunk into the accumulated text with **string-prefix heuristics**: + +```ts +if (incomingText.startsWith(existingText)) return incomingText; // superset → replace +if (existingText.startsWith(incomingText)) return existingText; // "stale duplicate" → DROP +return existingText + incomingText; // delta → append +``` + +The second guard drops any delta that happens to be a **prefix of the accumulated message**. A markdown table message starts with `|`, so every bare-`"|"` token the model streams (row boundaries, delimiter pipes) is silently swallowed. Proof: replaying a corrupted run's captured wire events through this exact algorithm reproduced the observed accumulation **byte-identically** (466 chars / 35 pipes; 13 dropped chunks, every one the single character `|`); the same replay without the guard yields the wire-exact text (48 pipes). The wire itself is intact — SDK-delivered events carry all characters. + +Contributing facts: + +- The failure is content-dependent and intermittent: only runs where the tokenizer emits bare `|` (or other message-prefix-coinciding) tokens trip it; runs with fused tokens (`|---`, `:|`) don't. Any message whose opening characters recur as a lone delta is exposed (e.g. `-` deltas in a message starting with a bullet). +- The first guard is also unsound for deltas: a `"| Gem"` delta arriving when the accumulation is `"|"` *replaces* instead of appending (drops a pipe). It didn't fire in captured runs but is the same class of bug. +- The guards exist for a real reason: after a node completes, values-sync installs canonical full text for the message, and a straggler token chunk arriving afterwards must not append duplicate text. The prefix check is a text-based *guess* at that situation. +- Both unit suites and aimock-replay e2e are blind to this: fixtures deliver content atomically, and the corruption depends on live token boundaries. + +## Design principle + +**Reconcile by identity and event kind — never by comparing text to text.** + +The event system already declares what each event is: messages-tuple events (`event.messageMetadata` present) are **deltas**; `messages/partial` events are **snapshots** (message-so-far); values-sync carries **canonical** state. The merge must use those declarations instead of inferring intent from string prefixes. Measured on live streams, tuple chunks carry a stable message id (150/151 chunks in a sampled run shared one id; the outlier is covered by the existing trailing-AI fallback), so identity-keyed accumulation is reliable. + +## Change (all in `libs/langgraph/src/lib/internals/stream-manager.bridge.ts`) + +1. **Thread a merge mode from the event site.** `processEvent`'s messages branch calls `mergeMessages(existing, normalized, reasoningTimingMap, mode)` with `mode: 'delta'` when `event.messageMetadata` is present (messages-tuple) and `mode: 'snapshot'` for `messages/partial`. `mergeMessages` passes the mode to `accumulateContent`. + +2. **`accumulateContent(existing, incoming, mode)`:** + - `snapshot` → current behavior, unchanged (mutual-prefix reconcile; snapshots are prefixes of one another by nature). + - `delta` → **append unconditionally**, with exactly two exceptions: + - `isFinalCanonicalReasoningContent(incoming)` → replace (unchanged; this is the authoritative final array shape, and it marks the message canonical — see 3). + - empty incoming → keep existing (unchanged). + - Both `startsWith` guards are removed from the delta path. + +3. **Canonical-id backstop (identity-based version of what the dropped guard was for).** The bridge keeps a `canonicalMessageIds: Set`: + - The `isFinalCanonicalReasoningContent` replacement adds the message's id — the one point where finality is *certain*. + - Values-sync does NOT mark ids: mid-run values events carry **lagging** state (the code comments document this), so treating them as final would suppress legitimate later deltas. Values-sync instead merges with **snapshot semantics** (see 2), which is lag-safe: a shorter stale snapshot never rewinds a longer accumulation. + - In `mergeMessages`, a **delta** whose target message id is in the set is ignored (a straggler token after canonical text landed — the scenario the old guard guessed at, now decided by identity). + - The set is cleared at the start of every `runStream` and on thread switch (same lifecycle as the other per-run maps, e.g. `toolProgressMap`). + +4. **No changes** to `findContentMatch`, the trailing-AI fallback, `preserveIds`, reasoning accumulation, or the snapshot path. + +## Error handling / invariants + +- **No text-shape assumptions:** the delta path never inspects content beyond extracting text; pipes, dashes, or any prefix-coinciding token are appended like any other. +- **No duplicates:** post-canonical stragglers are dropped by id membership, not text similarity. Values-sync continues to replace content wholesale for matching ids (existing behavior), so canonical text remains authoritative. +- **Ordering:** SSE delivers events in order on a single connection; a message's deltas precede its values-sync. The canonical set only ever suppresses deltas that arrive *after* canonical text for that id, which is precisely the stale case. +- **Lifecycle:** clearing the set per run prevents a regenerated message (new run, reused thread) from being suppressed by a previous run's canonical marking. + +## Testing + +**Unit (TDD, `libs/langgraph`)** — extend the existing harness in `libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts`, driving events through the bridge (a fake transport yielding scripted tuple/values events), matching that file's established pattern: + +1. **Pipe-preservation (red first):** feed tuple delta events `['|', ' Gem', ' |', ' Color', ' |', '\n', '|', '---', '|', '---', '|', '\n', '|', ' Ruby', ' |', ' red', ' |']` for one message id; assert the accumulated content equals the exact concatenation (fails today: bare `|` deltas vanish). +2. **Prefix-delta append:** accumulation `'|'` + delta `'| Gem'` → `'|| Gem'` (guards the first `startsWith` misfire). +3. **Straggler-after-canonical:** deltas accumulate → the final-canonical reasoning+text array replaces the accumulation (and marks the id) → one more late delta for that id → content unchanged (no duplicate, no loss). Separately, a lagging mid-run values-sync (shorter state for the same id) must not rewind the accumulation (snapshot semantics regression). +4. **Final-canonical replacement:** streamed accumulation + final reasoning+text array → replaced by canonical text (existing behavior preserved). +5. **Snapshot regression:** `messages/partial` snapshots (each a prefix-extension of the last, then a shorter stale one) reconcile exactly as today. +6. **Per-run reset:** after a new `runStream` on the same thread, deltas for a fresh message id accumulate normally even if a prior run marked ids canonical. + +**Suites:** full `libs/langgraph` + `libs/chat` (lint/test/build). + +**Live verification (required gate, Chrome MCP):** with the dev stack serving the fix, run ≥6 table prompts using the 3-layer capture (SDK events recorder + accumulation progression + DOM). Require on **every** run: `sdkPipes === accumPipes`, no non-prefix content divergence at stream end (accumulation now equals final), and zero table-collapse frames. Then remove the temporary SDK-event recorder from `fetch-stream.transport.ts` (added during diagnosis) before the PR. + +## Out of scope + +- The langgraph dev server / SDK (wire is proven intact). +- `subagent-tracker.ts`'s own `mergeMessages` (id/append-based already; no prefix guards). +- Stream resumability (`streamResumable`) — orthogonal transport hardening; not implicated. +- Any rendering-layer changes. diff --git a/libs/ag-ui/README.md b/libs/ag-ui/README.md index e0e42f790..6fc3829b1 100644 --- a/libs/ag-ui/README.md +++ b/libs/ag-ui/README.md @@ -32,10 +32,10 @@ Part of [Threadplane](https://github.com/cacheplane/angular-agent-framework). ## Install ```bash -npm install @threadplane/ag-ui @threadplane/chat @ag-ui/client +npm install @threadplane/ag-ui @threadplane/chat @ag-ui/client @ag-ui/core ``` -**Peer dependencies:** `@threadplane/chat: *`, `@angular/core: ^20.0.0 || ^21.0.0`, `@ag-ui/client: *`, `rxjs: ~7.8.0` +**Peer dependencies:** `@threadplane/chat: *`, `@angular/core: ^20.0.0 || ^21.0.0`, `@ag-ui/client: *`, `@ag-ui/core: *`, `rxjs: ~7.8.0` --- diff --git a/libs/chat/README.md b/libs/chat/README.md index bde78f6c0..d556874c1 100644 --- a/libs/chat/README.md +++ b/libs/chat/README.md @@ -28,7 +28,7 @@ Part of [Threadplane](https://github.com/cacheplane/angular-agent-framework). ## Install ```bash -npm install @threadplane/chat +npm install @threadplane/chat @threadplane/langgraph marked ``` **Peer dependencies:** @@ -37,6 +37,7 @@ npm install @threadplane/chat @angular/core ^20.0.0 || ^21.0.0 @angular/common ^20.0.0 || ^21.0.0 @angular/platform-browser ^20.0.0 || ^21.0.0 +@angular/router ^20.0.0 || ^21.0.0 @threadplane/licensing * @threadplane/render * @threadplane/a2ui * @@ -44,6 +45,8 @@ npm install @threadplane/chat @langchain/core ^1.1.33 rxjs ~7.8.0 marked ^15.0.0 || ^16.0.0 +zod ^3.25.0 +katex ^0.16.0 || ^0.17.0 (optional) ``` --- @@ -54,9 +57,13 @@ marked ^15.0.0 || ^16.0.0 // app.config.ts import { ApplicationConfig } from '@angular/core'; import { provideChat } from '@threadplane/chat'; +import { provideAgent } from '@threadplane/langgraph'; export const appConfig: ApplicationConfig = { - providers: [provideChat({ license: 'eyJ…' })], + providers: [ + provideAgent({ apiUrl: '/api/langgraph', assistantId: 'agent' }), + provideChat({ license: 'eyJ…' }), + ], }; ``` @@ -64,7 +71,7 @@ export const appConfig: ApplicationConfig = { // my.component.ts import { Component } from '@angular/core'; import { ChatComponent } from '@threadplane/chat'; -import { agent } from '@threadplane/langgraph'; +import { injectAgent } from '@threadplane/langgraph'; @Component({ selector: 'app-root', @@ -72,11 +79,11 @@ import { agent } from '@threadplane/langgraph'; template: ``, }) export class AppComponent { - myAgent = agent({ apiUrl: '/api/langgraph', graphId: 'agent' }); + protected readonly myAgent = injectAgent(); } ``` -Get the `agent` signal from `@threadplane/langgraph` (for LangGraph Platform backends) or `@threadplane/ag-ui` (for AG-UI-compatible backends). See those packages for setup details. +Get the agent from `@threadplane/langgraph` (for LangGraph Platform backends) or `@threadplane/ag-ui` (for AG-UI-compatible backends). See those packages for setup details. --- diff --git a/libs/langgraph/README.md b/libs/langgraph/README.md index 83ebadc39..61276b8ca 100644 --- a/libs/langgraph/README.md +++ b/libs/langgraph/README.md @@ -147,7 +147,7 @@ import { provideAgent, LangGraphThreadsAdapter, LANGGRAPH_THREADS_CONFIG } from export const appConfig: ApplicationConfig = { providers: [ - provideAgent({ apiUrl: 'https://your-langgraph-platform.com' }), + provideAgent({ apiUrl: 'https://your-langgraph-platform.com', assistantId: 'my-agent' }), { provide: LANGGRAPH_THREADS_CONFIG, useValue: { apiUrl: 'https://your-langgraph-platform.com' } }, LangGraphThreadsAdapter, ],