diff --git a/apps/website/content/docs/middleware/api/api-docs.json b/apps/website/content/docs/middleware/api/api-docs.json index 63fc67c38..4bcf673da 100644 --- a/apps/website/content/docs/middleware/api/api-docs.json +++ b/apps/website/content/docs/middleware/api/api-docs.json @@ -201,6 +201,106 @@ ], "examples": [] }, + { + "name": "ClientToolExecutionKey", + "kind": "interface", + "description": "Durable identity for one client-tool execution on one thread.", + "properties": [ + { + "name": "threadId", + "type": "string", + "description": "", + "optional": false + }, + { + "name": "toolCallId", + "type": "string", + "description": "", + "optional": false + } + ], + "examples": [] + }, + { + "name": "ClientToolExecutionStore", + "kind": "interface", + "description": "Backend-authoritative store for client-tool execution deduplication.", + "properties": [], + "methods": [ + { + "name": "claim", + "signature": "claim(key: ClientToolExecutionKey): Promise<\"claimed\" | ClientToolExecutionRecord>", + "description": "Atomically claim a tool-call execution. Returns prior state when present.", + "params": [ + { + "name": "key", + "type": "ClientToolExecutionKey", + "description": "", + "optional": false + } + ] + }, + { + "name": "lookup", + "signature": "lookup(threadId: string, toolCallIds: readonly string[]): Promise>", + "description": "Reload reconciliation: statuses/results for selected tool calls on a thread.", + "params": [ + { + "name": "threadId", + "type": "string", + "description": "", + "optional": false + }, + { + "name": "toolCallIds", + "type": "readonly string[]", + "description": "", + "optional": false + } + ] + }, + { + "name": "record", + "signature": "record(key: ClientToolExecutionKey, result: ClientToolResult): Promise", + "description": "Record the final client-tool result for a claimed or first-seen execution.", + "params": [ + { + "name": "key", + "type": "ClientToolExecutionKey", + "description": "", + "optional": false + }, + { + "name": "result", + "type": "ClientToolResult", + "description": "", + "optional": false + } + ] + } + ], + "examples": [] + }, + { + "name": "ClientToolResultMessage", + "kind": "interface", + "description": "", + "properties": [ + { + "name": "result", + "type": "ClientToolResult", + "description": "", + "optional": false + }, + { + "name": "toolCallId", + "type": "string", + "description": "", + "optional": false + } + ], + "examples": [] + }, { "name": "ClientToolSpec", "kind": "interface", @@ -273,6 +373,108 @@ ], "examples": [] }, + { + "name": "PostgresClientToolExecutionStoreOptions", + "kind": "interface", + "description": "", + "properties": [ + { + "name": "tenantId", + "type": "string | null", + "description": "", + "optional": true + } + ], + "examples": [] + }, + { + "name": "RecordClientToolResultsInput", + "kind": "interface", + "description": "", + "properties": [ + { + "name": "messages", + "type": "readonly BaseMessage, MessageType>[]", + "description": "", + "optional": false + }, + { + "name": "store", + "type": "ClientToolExecutionStore", + "description": "", + "optional": false + }, + { + "name": "threadId", + "type": "string", + "description": "", + "optional": false + } + ], + "examples": [] + }, + { + "name": "RecordClientToolResultsResult", + "kind": "interface", + "description": "", + "properties": [ + { + "name": "duplicateToolCallIds", + "type": "string[]", + "description": "", + "optional": false + }, + { + "name": "recordedToolCallIds", + "type": "string[]", + "description": "", + "optional": false + } + ], + "examples": [] + }, + { + "name": "ClientToolExecutionRecord", + "kind": "type", + "description": "", + "signature": "object | object | object", + "examples": [] + }, + { + "name": "ClientToolExecutionStatus", + "kind": "type", + "description": "", + "signature": "\"executing\" | \"done\" | \"failed\"", + "examples": [] + }, + { + "name": "ClientToolResult", + "kind": "type", + "description": "Serialized result for a browser-executed client tool.", + "signature": "object | object", + "examples": [] + }, + { + "name": "PostgresRow", + "kind": "type", + "description": "", + "signature": "Record", + "examples": [] + }, + { + "name": "PostgresTaggedSql", + "kind": "type", + "description": "", + "signature": "(strings: TemplateStringsArray, ...values: unknown[]) => Promise", + "examples": [] + }, + { + "name": "THREADPLANE_CLIENT_TOOL_EXECUTIONS_SCHEMA", + "kind": "const", + "description": "", + "signature": "\"\\nCREATE TABLE IF NOT EXISTS threadplane_client_tool_executions (\\n tenant_id text,\\n thread_id text NOT NULL,\\n tool_call_id text NOT NULL,\\n status text NOT NULL,\\n result jsonb,\\n created_at timestamptz NOT NULL DEFAULT now(),\\n updated_at timestamptz NOT NULL DEFAULT now(),\\n PRIMARY KEY (thread_id, tool_call_id)\\n);\\n\"", + "examples": [] + }, { "name": "bindClientTools", "kind": "function", @@ -379,6 +581,81 @@ }, "examples": [] }, + { + "name": "createInMemoryClientToolExecutionStore", + "kind": "function", + "description": "Non-persistent store useful for tests and Tier-0/Tier-1 local backends.", + "signature": "createInMemoryClientToolExecutionStore(): ClientToolExecutionStore", + "params": [], + "returns": { + "type": "ClientToolExecutionStore", + "description": "" + }, + "examples": [] + }, + { + "name": "createPostgresClientToolExecutionStore", + "kind": "function", + "description": "Create a Postgres-backed client-tool execution store from a supplied SQL tag.", + "signature": "createPostgresClientToolExecutionStore(sql: PostgresTaggedSql, opts: PostgresClientToolExecutionStoreOptions): ClientToolExecutionStore", + "params": [ + { + "name": "sql", + "type": "PostgresTaggedSql", + "description": "", + "optional": false + }, + { + "name": "opts", + "type": "PostgresClientToolExecutionStoreOptions", + "description": "", + "optional": true + } + ], + "returns": { + "type": "ClientToolExecutionStore", + "description": "" + }, + "examples": [] + }, + { + "name": "extractClientToolResultMessages", + "kind": "function", + "description": "Extract serialized client-tool results from LangGraph ToolMessages.", + "signature": "extractClientToolResultMessages(messages: readonly BaseMessage, MessageType>[]): ClientToolResultMessage[]", + "params": [ + { + "name": "messages", + "type": "readonly BaseMessage, MessageType>[]", + "description": "", + "optional": false + } + ], + "returns": { + "type": "ClientToolResultMessage[]", + "description": "" + }, + "examples": [] + }, + { + "name": "filterDuplicateClientToolResultMessages", + "kind": "function", + "description": "Remove duplicate client-tool result messages before server continuation logic sees them.", + "signature": "filterDuplicateClientToolResultMessages(input: object): BaseMessage, MessageType>[]", + "params": [ + { + "name": "input", + "type": "object", + "description": "", + "optional": false + } + ], + "returns": { + "type": "BaseMessage, MessageType>[]", + "description": "" + }, + "examples": [] + }, { "name": "hasClientToolCall", "kind": "function", @@ -442,6 +719,44 @@ }, "examples": [] }, + { + "name": "lookupClientToolExecutions", + "kind": "function", + "description": "Lookup prior client-tool executions for reload reconciliation.", + "signature": "lookupClientToolExecutions(input: object): Promise>", + "params": [ + { + "name": "input", + "type": "object", + "description": "", + "optional": false + } + ], + "returns": { + "type": "Promise>", + "description": "" + }, + "examples": [] + }, + { + "name": "recordClientToolResults", + "kind": "function", + "description": "Record first-seen client-tool results and identify duplicate redeliveries.", + "signature": "recordClientToolResults(input: RecordClientToolResultsInput): Promise", + "params": [ + { + "name": "input", + "type": "RecordClientToolResultsInput", + "description": "", + "optional": false + } + ], + "returns": { + "type": "Promise", + "description": "" + }, + "examples": [] + }, { "name": "routeAfterAgent", "kind": "function", diff --git a/docs/superpowers/plans/2026-07-08-client-tools-m3-durable-dedup-plan.md b/docs/superpowers/plans/2026-07-08-client-tools-m3-durable-dedup-plan.md new file mode 100644 index 000000000..05fcdfe6c --- /dev/null +++ b/docs/superpowers/plans/2026-07-08-client-tools-m3-durable-dedup-plan.md @@ -0,0 +1,267 @@ +# Client Tool Durable Dedup M3 Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Ship Tier 1 backend-authoritative client-tool result dedup in `@threadplane/middleware/langgraph`, including an injectable store interface, in-memory implementation, Postgres implementation, and reload lookup helpers. + +**Architecture:** Keep durable dedup in the backend middleware package. Graph authors wire the guard into their LangGraph node/route with a `threadId` and a `ClientToolExecutionStore`; browser packages remain unchanged. M3 records inbound client `ToolMessage` results by `toolCallId`, drops duplicate redeliveries before server continuation logic, and exposes lookup for reload reconciliation. Tier 2 claim-before-execute, `settle()`, batching, max-turns, and browser fallback markers stay out of scope. + +**Tech Stack:** TypeScript, Vitest, LangChain message classes, Nx, `postgres`-style tagged SQL supplied by consumers, generated API docs. + +--- + +## Source Of Truth + +- Spec: `docs/superpowers/specs/2026-07-07-client-tool-continuation-architecture-design.md` §7 and milestone M3. +- Store schema: nullable `tenant_id`, `thread_id`, `tool_call_id`, `status`, `result`, timestamps, first-ship PK `(thread_id, tool_call_id)`. +- No new browser dependencies. The Postgres helper accepts a supplied tagged SQL client; no new install is required in `@threadplane/chat`, `@threadplane/ag-ui`, or `@threadplane/langgraph`. +- M3 only: Tier 1 server guard and lookup. Do not implement Tier 2 pre-execution claims or fail-closed browser execution policy in this PR. + +## File Structure + +- Create: `libs/middleware/src/langgraph/client-tool-execution-store.ts` + - `ClientToolResult`, key/status/record types. + - `ClientToolExecutionStore` interface. + - `createInMemoryClientToolExecutionStore()` default implementation for deterministic tests and non-persistent backends. +- Create: `libs/middleware/src/langgraph/client-tool-result-guard.ts` + - Extract client `ToolMessage` results from LangGraph messages. + - `recordClientToolResults()` records first-seen tool-call results and reports duplicates. + - `filterDuplicateClientToolResultMessages()` returns messages with duplicate client-tool result redeliveries removed. + - `lookupClientToolExecutions()` wraps store lookup for reload reconciliation. +- Create: `libs/middleware/src/langgraph/postgres-client-tool-execution-store.ts` + - `THREADPLANE_CLIENT_TOOL_EXECUTIONS_SCHEMA` SQL string. + - `createPostgresClientToolExecutionStore(sql, opts?)` using a supplied `postgres`-style tagged client. +- Modify: `libs/middleware/src/langgraph/index.ts` + - Export M3 public API from the existing `/langgraph` entrypoint. +- Modify: `libs/middleware/README.md` + - Document Tier 1 guard wiring and Postgres store construction. +- Modify: `apps/website/content/docs/middleware/api/api-docs.json` + - Regenerate after new public exports. +- Test: `libs/middleware/src/client-tool-execution-store.spec.ts` +- Test: `libs/middleware/src/client-tool-result-guard.spec.ts` +- Test: `libs/middleware/src/postgres-client-tool-execution-store.spec.ts` + +--- + +### Task 1: Store Types And In-Memory Store + +**Files:** +- Create: `libs/middleware/src/client-tool-execution-store.spec.ts` +- Create: `libs/middleware/src/langgraph/client-tool-execution-store.ts` +- Modify: `libs/middleware/src/langgraph/index.ts` + +- [x] **Step 1: Write failing in-memory store tests** + +Cover: +- `claim()` returns `'claimed'` for a new `(threadId, toolCallId)`. +- A second `claim()` returns `{ status: 'executing' }`. +- `record()` stores `{ status: 'done', result }`. +- A later `claim()` returns `{ status: 'done', result }`. +- `lookup()` returns only requested known records. +- The returned lookup object cannot mutate internal store state. + +- [x] **Step 2: Run tests to verify red** + +Run: + +```bash +npx vitest run src/client-tool-execution-store.spec.ts --config vite.config.mts +``` + +Expected: FAIL because the module does not exist. + +- [x] **Step 3: Implement minimal store** + +Implement the interface and in-memory map. Use key format `${threadId}\0${toolCallId}` to avoid accidental collisions. + +- [x] **Step 4: Run tests to verify green** + +Run: + +```bash +npx vitest run src/client-tool-execution-store.spec.ts --config vite.config.mts +``` + +Expected: PASS. + +--- + +### Task 2: LangGraph ToolMessage Guard Helpers + +**Files:** +- Create: `libs/middleware/src/client-tool-result-guard.spec.ts` +- Create: `libs/middleware/src/langgraph/client-tool-result-guard.ts` +- Modify: `libs/middleware/src/langgraph/index.ts` + +- [x] **Step 1: Write failing extraction and record tests** + +Cover: +- Extracts `tool_call_id` from a LangChain `ToolMessage`. +- Converts `"Error: boom"` content to `{ ok: false, error: 'boom' }`. +- Converts JSON object content to `{ ok: true, value: object }`. +- Converts plain text content to `{ ok: true, value: string }`. +- Ignores non-tool messages and tool messages without an id. +- `recordClientToolResults()` claims and records first-seen result ids. +- A second call with the same result reports the id in `duplicateToolCallIds` and does not overwrite the first stored result. +- `filterDuplicateClientToolResultMessages()` removes only duplicate tool-result messages and leaves non-duplicates in order. +- `lookupClientToolExecutions()` delegates to store lookup. + +- [x] **Step 2: Run tests to verify red** + +Run: + +```bash +npx vitest run src/client-tool-result-guard.spec.ts --config vite.config.mts +``` + +Expected: FAIL because the guard module does not exist. + +- [x] **Step 3: Implement guard helpers** + +Implement helper signatures: + +```ts +recordClientToolResults(input: { + threadId: string; + messages: readonly BaseMessage[]; + store: ClientToolExecutionStore; +}): Promise<{ + recordedToolCallIds: string[]; + duplicateToolCallIds: string[]; +}>; + +filterDuplicateClientToolResultMessages(input: { + messages: readonly BaseMessage[]; + duplicateToolCallIds: ReadonlySet; +}): BaseMessage[]; + +lookupClientToolExecutions(input: { + threadId: string; + toolCallIds: readonly string[]; + store: ClientToolExecutionStore; +}): Promise>; +``` + +Keep this middleware-only. Do not change browser adapter pending or resolve behavior in M3. + +- [x] **Step 4: Run tests to verify green** + +Run: + +```bash +npx vitest run src/client-tool-result-guard.spec.ts --config vite.config.mts +``` + +Expected: PASS. + +--- + +### Task 3: Postgres Store + +**Files:** +- Create: `libs/middleware/src/postgres-client-tool-execution-store.spec.ts` +- Create: `libs/middleware/src/langgraph/postgres-client-tool-execution-store.ts` +- Modify: `libs/middleware/src/langgraph/index.ts` +- Modify: `libs/middleware/package.json` + +- [x] **Step 1: Write failing Postgres store tests** + +Use a fake tagged SQL function that records query text and returns scripted rows. Cover: +- `THREADPLANE_CLIENT_TOOL_EXECUTIONS_SCHEMA` contains the table, nullable `tenant_id`, and `PRIMARY KEY (thread_id, tool_call_id)`. +- `claim()` inserts `tenant_id`, `thread_id`, `tool_call_id`, `status='executing'` with `ON CONFLICT DO NOTHING`, returning `'claimed'` when insert returns a row. +- `claim()` reads the existing row on conflict and maps done rows back to `ClientToolExecutionRecord`. +- `record()` stores status `done` and a JSON result without overwriting an already-done row's result. +- `lookup()` returns a record keyed by `tool_call_id` for requested ids. + +- [x] **Step 2: Run tests to verify red** + +Run: + +```bash +npx vitest run src/postgres-client-tool-execution-store.spec.ts --config vite.config.mts +``` + +Expected: FAIL because the Postgres module does not exist. + +- [x] **Step 3: Implement Postgres store** + +Use a supplied `postgres`-style SQL tagged function and avoid importing the driver. Add `postgres` as an optional peer/dependency in `libs/middleware/package.json` only if TypeScript requires a package-level dependency for published consumers; otherwise keep the dependency surface unchanged. + +- [x] **Step 4: Run tests to verify green** + +Run: + +```bash +npx vitest run src/postgres-client-tool-execution-store.spec.ts --config vite.config.mts +``` + +Expected: PASS. + +--- + +### Task 4: Docs, API Docs, And Verification + +**Files:** +- Modify: `libs/middleware/README.md` +- Modify: `apps/website/content/docs/middleware/api/api-docs.json` + +- [x] **Step 1: Update middleware README** + +Add a short “Durable client-tool result guard” section showing: +- in-memory store creation, +- where to call `recordClientToolResults()`, +- how to drop duplicates with `filterDuplicateClientToolResultMessages()`, +- Postgres schema/setup with the exported SQL constant, +- note that Tier 2 claim-before-execute is not part of M3. + +- [x] **Step 2: Regenerate API docs** + +Run: + +```bash +npm run generate-api-docs +``` + +Expected: middleware docs include new M3 exports. + +- [x] **Step 3: Run middleware tests** + +Run: + +```bash +npx nx test middleware +``` + +Expected: PASS. + +- [x] **Step 4: Run middleware lint and build** + +Run: + +```bash +npx nx lint middleware +npx nx build middleware +``` + +Expected: both PASS. + +- [x] **Step 5: Run focused library guard** + +Run: + +```bash +npx nx run-many -t build --projects=chat,langgraph,ag-ui,render,a2ui,licensing,telemetry --configuration=production +``` + +Expected: PASS, preserving the fix-forwarded library CI gate. + +- [x] **Step 6: Diff audit** + +Run: + +```bash +git diff --check +(git diff --name-only; git ls-files --others --exclude-standard) | rg -v '^docs/superpowers/' | xargs rg -n "hashbrown|copilotkit|chatgpt|claude" || true +``` + +Expected: no whitespace errors; changed non-plan files contain none of the forbidden external names. diff --git a/libs/middleware/README.md b/libs/middleware/README.md index 53f853363..bc8460d8e 100644 --- a/libs/middleware/README.md +++ b/libs/middleware/README.md @@ -82,6 +82,59 @@ import { } from '@threadplane/middleware/langgraph'; ``` +## Durable client-tool result guard + +Tier 1 durable dedup records inbound client-tool `ToolMessage` results by +`tool_call_id` before the graph continues. A duplicate redelivery can then be +filtered before server continuation logic sees it. + +```ts +import { + createInMemoryClientToolExecutionStore, + filterDuplicateClientToolResultMessages, + recordClientToolResults, +} from '@threadplane/middleware/langgraph'; + +const clientToolExecutions = createInMemoryClientToolExecutionStore(); + +async function agent(state: typeof State.State, config: { configurable?: { thread_id?: string } }) { + const threadId = config.configurable?.thread_id ?? 'default-thread'; + const guard = await recordClientToolResults({ + threadId, + messages: state.messages, + store: clientToolExecutions, + }); + const messages = filterDuplicateClientToolResultMessages({ + messages: state.messages, + duplicateToolCallIds: new Set(guard.duplicateToolCallIds), + }); + + const llm = bindClientTools(baseLlm, SERVER_TOOLS, state); + const response = await llm.invoke(messages); + return { messages: [response] }; +} +``` + +For persistent storage, create the table once and pass a `postgres`-style SQL +tag to the Postgres store: + +```ts +import postgres from 'postgres'; +import { + THREADPLANE_CLIENT_TOOL_EXECUTIONS_SCHEMA, + createPostgresClientToolExecutionStore, +} from '@threadplane/middleware/langgraph'; + +const sql = postgres(process.env.DATABASE_URL!); +await sql.unsafe(THREADPLANE_CLIENT_TOOL_EXECUTIONS_SCHEMA); + +const clientToolExecutions = createPostgresClientToolExecutionStore(sql); +``` + +M3 is server-side Tier 1 only: it dedups delivered client-tool results and +supports lookup-based reload reconciliation. Pre-execution claims for +non-idempotent browser effects are a later opt-in layer. + ## Peer dependencies `@langchain/core` and `@langchain/langgraph`. The package has no runtime dependencies of its diff --git a/libs/middleware/src/client-tool-execution-store.spec.ts b/libs/middleware/src/client-tool-execution-store.spec.ts new file mode 100644 index 000000000..4646baf5d --- /dev/null +++ b/libs/middleware/src/client-tool-execution-store.spec.ts @@ -0,0 +1,56 @@ +// SPDX-License-Identifier: MIT +import { describe, expect, it } from 'vitest'; +import { createInMemoryClientToolExecutionStore } from './langgraph/client-tool-execution-store'; +import type { ClientToolExecutionRecord, ClientToolResult } from './langgraph/client-tool-execution-store'; + +const key = { threadId: 'thread-1', toolCallId: 'call-1' }; +const result: ClientToolResult = { ok: true, value: { temp: 72 } }; + +describe('createInMemoryClientToolExecutionStore', () => { + it('claims a new tool-call execution', async () => { + const store = createInMemoryClientToolExecutionStore(); + + await expect(store.claim(key)).resolves.toBe('claimed'); + }); + + it('returns executing for a second claim before a result is recorded', async () => { + const store = createInMemoryClientToolExecutionStore(); + + await store.claim(key); + + await expect(store.claim(key)).resolves.toEqual({ status: 'executing' }); + }); + + it('returns the stored done result after record', async () => { + const store = createInMemoryClientToolExecutionStore(); + + await store.claim(key); + await store.record(key, result); + + await expect(store.claim(key)).resolves.toEqual({ status: 'done', result }); + }); + + it('looks up only requested known records', async () => { + const store = createInMemoryClientToolExecutionStore(); + await store.claim(key); + await store.record(key, result); + await store.claim({ threadId: 'thread-1', toolCallId: 'call-2' }); + + await expect(store.lookup('thread-1', ['call-1', 'missing'])).resolves.toEqual({ + 'call-1': { status: 'done', result }, + }); + }); + + it('does not expose mutable internal records through lookup', async () => { + const store = createInMemoryClientToolExecutionStore(); + await store.claim(key); + await store.record(key, result); + + const first = await store.lookup('thread-1', ['call-1']); + (first['call-1'] as ClientToolExecutionRecord).status = 'failed'; + + await expect(store.lookup('thread-1', ['call-1'])).resolves.toEqual({ + 'call-1': { status: 'done', result }, + }); + }); +}); diff --git a/libs/middleware/src/client-tool-result-guard.spec.ts b/libs/middleware/src/client-tool-result-guard.spec.ts new file mode 100644 index 000000000..5c0478ec1 --- /dev/null +++ b/libs/middleware/src/client-tool-result-guard.spec.ts @@ -0,0 +1,120 @@ +// SPDX-License-Identifier: MIT +import { AIMessage, HumanMessage, ToolMessage } from '@langchain/core/messages'; +import { describe, expect, it } from 'vitest'; +import { createInMemoryClientToolExecutionStore } from './langgraph/client-tool-execution-store'; +import { + extractClientToolResultMessages, + filterDuplicateClientToolResultMessages, + lookupClientToolExecutions, + recordClientToolResults, +} from './langgraph/client-tool-result-guard'; + +describe('extractClientToolResultMessages', () => { + it('extracts tool_call_id from a LangChain ToolMessage', () => { + const [entry] = extractClientToolResultMessages([ + new ToolMessage({ content: '{"temp":72}', tool_call_id: 'call-1' }), + ]); + + expect(entry).toEqual({ + toolCallId: 'call-1', + result: { ok: true, value: { temp: 72 } }, + }); + }); + + it('converts error content to an error result', () => { + const [entry] = extractClientToolResultMessages([ + new ToolMessage({ content: 'Error: boom', tool_call_id: 'call-err' }), + ]); + + expect(entry.result).toEqual({ ok: false, error: 'boom' }); + }); + + it('converts plain text content to an ok string result', () => { + const [entry] = extractClientToolResultMessages([ + new ToolMessage({ content: 'plain text', tool_call_id: 'call-text' }), + ]); + + expect(entry.result).toEqual({ ok: true, value: 'plain text' }); + }); + + it('ignores non-tool messages and tool messages without an id', () => { + expect(extractClientToolResultMessages([ + new HumanMessage('hi'), + new AIMessage('hello'), + new ToolMessage({ content: 'x', tool_call_id: '' }), + ])).toEqual([]); + }); +}); + +describe('recordClientToolResults', () => { + it('claims and records first-seen result ids', async () => { + const store = createInMemoryClientToolExecutionStore(); + const messages = [new ToolMessage({ content: '{"ok":true}', tool_call_id: 'call-1' })]; + + await expect(recordClientToolResults({ + threadId: 'thread-1', + messages, + store, + })).resolves.toEqual({ + recordedToolCallIds: ['call-1'], + duplicateToolCallIds: [], + }); + + await expect(store.lookup('thread-1', ['call-1'])).resolves.toEqual({ + 'call-1': { status: 'done', result: { ok: true, value: { ok: true } } }, + }); + }); + + it('reports duplicate done ids without overwriting the first result', async () => { + const store = createInMemoryClientToolExecutionStore(); + const first = [new ToolMessage({ content: '{"first":true}', tool_call_id: 'call-1' })]; + const second = [new ToolMessage({ content: '{"second":true}', tool_call_id: 'call-1' })]; + + await recordClientToolResults({ threadId: 'thread-1', messages: first, store }); + + await expect(recordClientToolResults({ + threadId: 'thread-1', + messages: second, + store, + })).resolves.toEqual({ + recordedToolCallIds: [], + duplicateToolCallIds: ['call-1'], + }); + + await expect(store.lookup('thread-1', ['call-1'])).resolves.toEqual({ + 'call-1': { status: 'done', result: { ok: true, value: { first: true } } }, + }); + }); +}); + +describe('filterDuplicateClientToolResultMessages', () => { + it('removes only duplicate tool-result messages and keeps order for the rest', () => { + const human = new HumanMessage('hi'); + const duplicate = new ToolMessage({ content: 'dup', tool_call_id: 'call-1' }); + const fresh = new ToolMessage({ content: 'fresh', tool_call_id: 'call-2' }); + + expect(filterDuplicateClientToolResultMessages({ + messages: [human, duplicate, fresh], + duplicateToolCallIds: new Set(['call-1']), + })).toEqual([human, fresh]); + }); +}); + +describe('lookupClientToolExecutions', () => { + it('delegates reload reconciliation to the store lookup', async () => { + const store = createInMemoryClientToolExecutionStore(); + await recordClientToolResults({ + threadId: 'thread-1', + messages: [new ToolMessage({ content: '{"done":true}', tool_call_id: 'call-1' })], + store, + }); + + await expect(lookupClientToolExecutions({ + threadId: 'thread-1', + toolCallIds: ['call-1', 'missing'], + store, + })).resolves.toEqual({ + 'call-1': { status: 'done', result: { ok: true, value: { done: true } } }, + }); + }); +}); diff --git a/libs/middleware/src/langgraph/client-tool-execution-store.ts b/libs/middleware/src/langgraph/client-tool-execution-store.ts new file mode 100644 index 000000000..69d45224f --- /dev/null +++ b/libs/middleware/src/langgraph/client-tool-execution-store.ts @@ -0,0 +1,87 @@ +// SPDX-License-Identifier: MIT + +/** Serialized result for a browser-executed client tool. */ +export type ClientToolResult = + | { readonly ok: true; readonly value: unknown } + | { readonly ok: false; readonly error: string }; + +/** Durable identity for one client-tool execution on one thread. */ +export interface ClientToolExecutionKey { + readonly threadId: string; + readonly toolCallId: string; +} + +export type ClientToolExecutionStatus = 'executing' | 'done' | 'failed'; + +export type ClientToolExecutionRecord = + | { status: 'executing' } + | { status: 'done'; result: ClientToolResult } + | { status: 'failed'; result?: ClientToolResult }; + +/** Backend-authoritative store for client-tool execution deduplication. */ +export interface ClientToolExecutionStore { + /** Atomically claim a tool-call execution. Returns prior state when present. */ + claim(key: ClientToolExecutionKey): Promise<'claimed' | ClientToolExecutionRecord>; + /** Record the final client-tool result for a claimed or first-seen execution. */ + record(key: ClientToolExecutionKey, result: ClientToolResult): Promise; + /** Reload reconciliation: statuses/results for selected tool calls on a thread. */ + lookup( + threadId: string, + toolCallIds: readonly string[], + ): Promise>; +} + +/** Non-persistent store useful for tests and Tier-0/Tier-1 local backends. */ +export function createInMemoryClientToolExecutionStore(): ClientToolExecutionStore { + const records = new Map(); + + return { + async claim(key: ClientToolExecutionKey): Promise<'claimed' | ClientToolExecutionRecord> { + const existing = records.get(mapKey(key)); + if (existing) return cloneRecord(existing); + records.set(mapKey(key), { status: 'executing' }); + return 'claimed'; + }, + + async record(key: ClientToolExecutionKey, result: ClientToolResult): Promise { + records.set(mapKey(key), { status: 'done', result: cloneResult(result) }); + }, + + async lookup( + threadId: string, + toolCallIds: readonly string[], + ): Promise> { + const out: Record = {}; + for (const toolCallId of toolCallIds) { + const existing = records.get(mapKey({ threadId, toolCallId })); + if (existing) out[toolCallId] = cloneRecord(existing); + } + return out; + }, + }; +} + +function mapKey(key: ClientToolExecutionKey): string { + return `${key.threadId}\0${key.toolCallId}`; +} + +function cloneRecord(record: ClientToolExecutionRecord): ClientToolExecutionRecord { + if (record.status === 'done') return { status: 'done', result: cloneResult(record.result) }; + if (record.status === 'failed') { + return record.result + ? { status: 'failed', result: cloneResult(record.result) } + : { status: 'failed' }; + } + return { status: 'executing' }; +} + +function cloneResult(result: ClientToolResult): ClientToolResult { + if (!result.ok) return { ok: false, error: result.error }; + return { ok: true, value: structuredCloneIfAvailable(result.value) }; +} + +function structuredCloneIfAvailable(value: unknown): unknown { + return typeof structuredClone === 'function' + ? structuredClone(value) + : JSON.parse(JSON.stringify(value)); +} diff --git a/libs/middleware/src/langgraph/client-tool-result-guard.ts b/libs/middleware/src/langgraph/client-tool-result-guard.ts new file mode 100644 index 000000000..888008537 --- /dev/null +++ b/libs/middleware/src/langgraph/client-tool-result-guard.ts @@ -0,0 +1,118 @@ +// SPDX-License-Identifier: MIT +import type { BaseMessage } from './types.js'; +import type { + ClientToolExecutionRecord, + ClientToolExecutionStore, + ClientToolResult, +} from './client-tool-execution-store.js'; + +export interface ClientToolResultMessage { + readonly toolCallId: string; + readonly result: ClientToolResult; +} + +export interface RecordClientToolResultsInput { + readonly threadId: string; + readonly messages: readonly BaseMessage[]; + readonly store: ClientToolExecutionStore; +} + +export interface RecordClientToolResultsResult { + readonly recordedToolCallIds: string[]; + readonly duplicateToolCallIds: string[]; +} + +/** Extract serialized client-tool results from LangGraph ToolMessages. */ +export function extractClientToolResultMessages( + messages: readonly BaseMessage[], +): ClientToolResultMessage[] { + const out: ClientToolResultMessage[] = []; + for (const message of messages) { + const toolCallId = toolCallIdFromMessage(message); + if (!toolCallId) continue; + out.push({ toolCallId, result: resultFromMessageContent(message.content) }); + } + return out; +} + +/** Record first-seen client-tool results and identify duplicate redeliveries. */ +export async function recordClientToolResults( + input: RecordClientToolResultsInput, +): Promise { + const recordedToolCallIds: string[] = []; + const duplicateToolCallIds: string[] = []; + + for (const entry of extractClientToolResultMessages(input.messages)) { + const key = { threadId: input.threadId, toolCallId: entry.toolCallId }; + const claim = await input.store.claim(key); + if (claim === 'claimed') { + await input.store.record(key, entry.result); + recordedToolCallIds.push(entry.toolCallId); + continue; + } + if (claim.status === 'done') { + duplicateToolCallIds.push(entry.toolCallId); + continue; + } + await input.store.record(key, entry.result); + recordedToolCallIds.push(entry.toolCallId); + } + + return { recordedToolCallIds, duplicateToolCallIds }; +} + +/** Remove duplicate client-tool result messages before server continuation logic sees them. */ +export function filterDuplicateClientToolResultMessages(input: { + readonly messages: readonly BaseMessage[]; + readonly duplicateToolCallIds: ReadonlySet; +}): BaseMessage[] { + return input.messages.filter((message) => { + const toolCallId = toolCallIdFromMessage(message); + return !toolCallId || !input.duplicateToolCallIds.has(toolCallId); + }); +} + +/** Lookup prior client-tool executions for reload reconciliation. */ +export function lookupClientToolExecutions(input: { + readonly threadId: string; + readonly toolCallIds: readonly string[]; + readonly store: ClientToolExecutionStore; +}): Promise> { + return input.store.lookup(input.threadId, input.toolCallIds); +} + +function toolCallIdFromMessage(message: BaseMessage): string | undefined { + const raw = message as unknown as Record; + const direct = raw['tool_call_id'] ?? raw['toolCallId']; + if (typeof direct === 'string' && direct.length > 0) return direct; + const kwargs = raw['additional_kwargs']; + if (kwargs && typeof kwargs === 'object') { + const nested = (kwargs as Record)['tool_call_id']; + if (typeof nested === 'string' && nested.length > 0) return nested; + } + const lcKwargs = raw['lc_kwargs']; + if (lcKwargs && typeof lcKwargs === 'object') { + const nested = (lcKwargs as Record)['tool_call_id']; + if (typeof nested === 'string' && nested.length > 0) return nested; + } + return undefined; +} + +function resultFromMessageContent(content: unknown): ClientToolResult { + const text = contentToText(content); + if (text.startsWith('Error: ')) { + return { ok: false, error: text.slice('Error: '.length) }; + } + try { + return { ok: true, value: JSON.parse(text) }; + } catch { + return { ok: true, value: text }; + } +} + +function contentToText(content: unknown): string { + if (typeof content === 'string') return content; + if (Array.isArray(content)) return content.map(contentToText).join(''); + if (content == null) return ''; + return String(content); +} diff --git a/libs/middleware/src/langgraph/index.ts b/libs/middleware/src/langgraph/index.ts index 42ec31b0e..339dd251b 100644 --- a/libs/middleware/src/langgraph/index.ts +++ b/libs/middleware/src/langgraph/index.ts @@ -12,3 +12,33 @@ export { } from './middleware.js'; export { clientToolsChannel } from './channel.js'; export { clientToolsRouter } from './router.js'; +export { + createInMemoryClientToolExecutionStore, +} from './client-tool-execution-store.js'; +export type { + ClientToolExecutionKey, + ClientToolExecutionRecord, + ClientToolExecutionStatus, + ClientToolExecutionStore, + ClientToolResult, +} from './client-tool-execution-store.js'; +export { + extractClientToolResultMessages, + filterDuplicateClientToolResultMessages, + lookupClientToolExecutions, + recordClientToolResults, +} from './client-tool-result-guard.js'; +export type { + ClientToolResultMessage, + RecordClientToolResultsInput, + RecordClientToolResultsResult, +} from './client-tool-result-guard.js'; +export { + THREADPLANE_CLIENT_TOOL_EXECUTIONS_SCHEMA, + createPostgresClientToolExecutionStore, +} from './postgres-client-tool-execution-store.js'; +export type { + PostgresClientToolExecutionStoreOptions, + PostgresRow, + PostgresTaggedSql, +} from './postgres-client-tool-execution-store.js'; diff --git a/libs/middleware/src/langgraph/postgres-client-tool-execution-store.ts b/libs/middleware/src/langgraph/postgres-client-tool-execution-store.ts new file mode 100644 index 000000000..8bff28261 --- /dev/null +++ b/libs/middleware/src/langgraph/postgres-client-tool-execution-store.ts @@ -0,0 +1,108 @@ +// SPDX-License-Identifier: MIT +import type { + ClientToolExecutionKey, + ClientToolExecutionRecord, + ClientToolExecutionStore, + ClientToolResult, +} from './client-tool-execution-store.js'; + +export type PostgresRow = Record; +export type PostgresTaggedSql = ( + strings: TemplateStringsArray, + ...values: unknown[] +) => Promise; + +export const THREADPLANE_CLIENT_TOOL_EXECUTIONS_SCHEMA = ` +CREATE TABLE IF NOT EXISTS threadplane_client_tool_executions ( + tenant_id text, + thread_id text NOT NULL, + tool_call_id text NOT NULL, + status text NOT NULL, + result jsonb, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (thread_id, tool_call_id) +); +`; + +export interface PostgresClientToolExecutionStoreOptions { + readonly tenantId?: string | null; +} + +/** Create a Postgres-backed client-tool execution store from a supplied SQL tag. */ +export function createPostgresClientToolExecutionStore( + sql: PostgresTaggedSql, + opts: PostgresClientToolExecutionStoreOptions = {}, +): ClientToolExecutionStore { + const tenantId = opts.tenantId ?? null; + + return { + async claim(key: ClientToolExecutionKey): Promise<'claimed' | ClientToolExecutionRecord> { + const inserted = await sql` + INSERT INTO threadplane_client_tool_executions + (tenant_id, thread_id, tool_call_id, status) + VALUES (${tenantId}, ${key.threadId}, ${key.toolCallId}, 'executing') + ON CONFLICT (thread_id, tool_call_id) DO NOTHING + RETURNING status, result + `; + if (inserted.length > 0) return 'claimed'; + + const existing = await sql` + SELECT status, result + FROM threadplane_client_tool_executions + WHERE thread_id = ${key.threadId} + AND tool_call_id = ${key.toolCallId} + LIMIT 1 + `; + return rowToRecord(existing[0]); + }, + + async record(key: ClientToolExecutionKey, result: ClientToolResult): Promise { + await sql` + INSERT INTO threadplane_client_tool_executions + (tenant_id, thread_id, tool_call_id, status, result) + VALUES (${tenantId}, ${key.threadId}, ${key.toolCallId}, 'done', ${JSON.stringify(result)}::jsonb) + ON CONFLICT (thread_id, tool_call_id) DO UPDATE + SET status = 'done', + result = CASE + WHEN threadplane_client_tool_executions.status = 'done' + THEN threadplane_client_tool_executions.result + ELSE EXCLUDED.result + END, + updated_at = now() + `; + }, + + async lookup( + threadId: string, + toolCallIds: readonly string[], + ): Promise> { + if (toolCallIds.length === 0) return {}; + const rows = await sql` + SELECT tool_call_id, status, result + FROM threadplane_client_tool_executions + WHERE thread_id = ${threadId} + AND tool_call_id = ANY(${[...toolCallIds]}) + `; + const out: Record = {}; + for (const row of rows) { + if (typeof row['tool_call_id'] !== 'string') continue; + out[row['tool_call_id']] = rowToRecord(row); + } + return out; + }, + }; +} + +function rowToRecord(row: PostgresRow | undefined): ClientToolExecutionRecord { + const status = row?.['status']; + if (status === 'done') { + return { status: 'done', result: row?.['result'] as ClientToolResult }; + } + if (status === 'failed') { + return row?.['result'] + ? { status: 'failed', result: row['result'] as ClientToolResult } + : { status: 'failed' }; + } + return { status: 'executing' }; +} diff --git a/libs/middleware/src/postgres-client-tool-execution-store.spec.ts b/libs/middleware/src/postgres-client-tool-execution-store.spec.ts new file mode 100644 index 000000000..ec21ba862 --- /dev/null +++ b/libs/middleware/src/postgres-client-tool-execution-store.spec.ts @@ -0,0 +1,82 @@ +// SPDX-License-Identifier: MIT +import { describe, expect, it } from 'vitest'; +import { + THREADPLANE_CLIENT_TOOL_EXECUTIONS_SCHEMA, + createPostgresClientToolExecutionStore, + type PostgresTaggedSql, +} from './langgraph/postgres-client-tool-execution-store'; + +function makeSql(rows: unknown[][]): { sql: PostgresTaggedSql; queries: string[]; values: unknown[][] } { + const queries: string[] = []; + const values: unknown[][] = []; + const sql = (async (strings: TemplateStringsArray, ...params: unknown[]) => { + queries.push(strings.join('?')); + values.push(params); + return rows.shift() ?? []; + }) as PostgresTaggedSql; + return { sql, queries, values }; +} + +describe('THREADPLANE_CLIENT_TOOL_EXECUTIONS_SCHEMA', () => { + it('creates the client-tool execution table with first-ship single-tenant primary key', () => { + expect(THREADPLANE_CLIENT_TOOL_EXECUTIONS_SCHEMA).toContain('CREATE TABLE'); + expect(THREADPLANE_CLIENT_TOOL_EXECUTIONS_SCHEMA).toContain('threadplane_client_tool_executions'); + expect(THREADPLANE_CLIENT_TOOL_EXECUTIONS_SCHEMA).toContain('tenant_id'); + expect(THREADPLANE_CLIENT_TOOL_EXECUTIONS_SCHEMA).toContain('PRIMARY KEY (thread_id, tool_call_id)'); + }); +}); + +describe('createPostgresClientToolExecutionStore', () => { + it('claims a new execution with ON CONFLICT DO NOTHING', async () => { + const { sql, queries, values } = makeSql([[{ status: 'executing', result: null }]]); + const store = createPostgresClientToolExecutionStore(sql, { tenantId: 'tenant-1' }); + + await expect(store.claim({ threadId: 'thread-1', toolCallId: 'call-1' })).resolves.toBe('claimed'); + + expect(queries[0]).toContain('ON CONFLICT (thread_id, tool_call_id) DO NOTHING'); + expect(values[0]).toEqual(['tenant-1', 'thread-1', 'call-1']); + }); + + it('reads the existing row when claim conflicts', async () => { + const { sql } = makeSql([ + [], + [{ status: 'done', result: { ok: true, value: { temp: 72 } } }], + ]); + const store = createPostgresClientToolExecutionStore(sql); + + await expect(store.claim({ threadId: 'thread-1', toolCallId: 'call-1' })).resolves.toEqual({ + status: 'done', + result: { ok: true, value: { temp: 72 } }, + }); + }); + + it('records a done result without overwriting an already-done result', async () => { + const { sql, queries, values } = makeSql([[]]); + const store = createPostgresClientToolExecutionStore(sql); + const result = { ok: false as const, error: 'boom' }; + + await store.record({ threadId: 'thread-1', toolCallId: 'call-1' }, result); + + expect(queries[0]).toContain('ON CONFLICT (thread_id, tool_call_id) DO UPDATE'); + expect(queries[0]).toContain('WHEN threadplane_client_tool_executions.status ='); + expect(values[0]).toEqual([null, 'thread-1', 'call-1', JSON.stringify(result)]); + }); + + it('looks up records by requested tool_call_id', async () => { + const { sql, queries, values } = makeSql([ + [ + { tool_call_id: 'call-1', status: 'done', result: { ok: true, value: 'done' } }, + { tool_call_id: 'call-2', status: 'executing', result: null }, + ], + ]); + const store = createPostgresClientToolExecutionStore(sql); + + await expect(store.lookup('thread-1', ['call-1', 'call-2'])).resolves.toEqual({ + 'call-1': { status: 'done', result: { ok: true, value: 'done' } }, + 'call-2': { status: 'executing' }, + }); + + expect(queries[0]).toContain('tool_call_id = ANY'); + expect(values[0]).toEqual(['thread-1', ['call-1', 'call-2']]); + }); +});