From 07c862fe85d24da577ec93aa52188831fcb3a0af Mon Sep 17 00:00:00 2001 From: Nicholas Wright Date: Thu, 8 Oct 2026 18:10:40 -0600 Subject: [PATCH 1/2] feat: call DataGrout's REST API instead of tools Each operation is now one request to a resource endpoint (answers, transforms, memories, skill runs) with named fields in the response. The node no longer names tools or reads a tool result envelope. Slow work is followed through the task the API returns: the node asks the server to hold the request with Prefer: wait, then polls the task's location until the result is ready or Wait for Result runs out. An unverified answer is data (verified: false) rather than a node failure. Transform can split its result into one item per row. Memory Name defaults to "default". OAuth2 credentials use the server-less routes, since the token identifies the server. --- CHANGELOG.md | 62 ++-- README.md | 48 ++- credentials/DataGroutApi.credentials.ts | 19 +- nodes/DataGrout/DataGrout.node.ts | 197 +++--------- nodes/DataGrout/pure.ts | 220 +++++-------- nodes/DataGrout/transport.ts | 261 ++++++--------- test/pure.test.ts | 407 ++++++++---------------- test/session.test.ts | 10 +- 8 files changed, 426 insertions(+), 798 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ec42d0f..19987d2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,45 +1,49 @@ ## [0.4.0] — unreleased **Breaking.** The node is now organised around DataGrout operations instead of -raw tool calls, and talks to DataGrout's JSON-RPC API rather than MCP. Existing +raw tool calls, and calls DataGrout's REST API rather than MCP. Existing workflows using the old *DataGrout MCP* node need to be repointed at the new **DataGrout** node. ### Added -- **DataGrout Trigger** — starts a workflow when DataGrout pushes an event, over - one multiplexed WebSocket. Subscribes once to the server's event stream and - filters by event (`run.completed`, `task.completed`, `task.failed`, - `tool_call.failed`), emits one item per event, - filters connection lifecycle frames so a reconnect does not fire the workflow, - holds the connection open with a keepalive, and reopens a dropped one with a - backing-off delay. Requires Node 22+. -- **Answer → Ask** — a question in plain language returns a verified answer, a - caveat when DataGrout could not verify it, and a certificate URL. +- **Answer → Ask** — a question in plain language returns an answer, whether + DataGrout's verifier accepted it, a certificate showing what ran, and a skill + id to run the same work again. - **Data → Transform** — server-side compute over inline data or a reference, - returning one item per record. + returning one item per row, or the whole response with **Split Rows Into + Items** off. - **Memory → Remember / Recall** — facts that persist across executions, and - recall that includes what can be inferred from them. -- **Skill → Run** — re-run previously verified work. -- **API Token authentication** alongside OAuth2, selectable on the node. -- The transport a node needs is switched on automatically the first time a - workflow uses it — JSON-RPC on the node's first call, WebSocket when the - trigger starts — so nothing needs configuring in the DataGrout dashboard. - Both are logged, so a change to your server's settings is never silent. - Testing a credential only reads, and changes nothing. + recall that includes what can be inferred from them. Memory Name defaults to + `default`, the memory DataGrout uses when none is named. +- **Skill → Run** — re-run previously verified work, by a saved skill's id or the + Skill ID an Ask returned. +- **DataGrout Trigger** — starts a workflow when DataGrout pushes an event, over + one multiplexed WebSocket. Subscribes once to the server's event stream, + filters by event (`run.completed`, `task.completed`, `task.failed`, + `tool_call.failed`), emits one item per event, filters connection lifecycle + frames so a reconnect does not fire the workflow, holds the connection open + with a keepalive, and reopens a dropped one with a backing-off delay. It + switches on the WebSocket transport when it starts, and logs that it did. + Requires Node 22+. +- **API Token authentication** alongside OAuth2, selectable on the node. With + OAuth2 there is still nothing to fill in: DataGrout resolves the server from + the connected account. ### Changed -- Tool names are gone from the interface. Operations are named for what they do. -- A loop-guard intervention now fails the node instead of arriving as the answer. - DataGrout answers an identical call repeated with nothing changing in between - with an explanation rather than a result, and two of those replies are not - marked as errors — so a downstream node expecting rows used to receive - `{loop_detected: true, …}` as its data. The message says what happened and how - to proceed. -- A tool-level failure reported inside a success response now fails the node, - instead of returning an empty answer marked verified. +- Tool names and tool result handling are gone from the node. Each operation is + one REST call whose response has named fields, and slow work is followed by + polling the task the API returns. +- An answer DataGrout could not verify is returned as `verified: false`, with + what the plan found beside it, instead of failing the node — so a workflow can + branch on it. +- Errors carry DataGrout's own message: a repeated identical request the loop + guard declined, a skill that is unknown or has expired, a connected system + failing. +- Testing the API Token credential only reads, and costs nothing. ### Removed -- The **DataGrout MCP** node and its `List Tools` / `Execute Tool` operations. +- The **DataGrout MCP** node: its `List Tools` / `Execute Tool` operations and + the **Tools to Allow** filter. ## [0.3.1] - 2026-08-17 diff --git a/README.md b/README.md index 233152f..8dff372 100644 --- a/README.md +++ b/README.md @@ -15,13 +15,16 @@ Two nodes: | **DataGrout** | Ask questions, compute over data, store and recall facts, re-run verified work. Also works on an AI Agent's Tool connector. | | **DataGrout Trigger** | Start a workflow when a DataGrout run or background task finishes or fails. | +The DataGrout node calls DataGrout's [REST API](https://library.datagrout.ai/rest-api). No runtime dependencies. ## What you get that a plain HTTP call does not -- **A verified answer, or an admission.** Every answer carries whether DataGrout - could verify it and a certificate URL showing the steps that produced it. A - result that failed verification arrives as an error, not as a confident null. +- **A verified answer, or an admission.** Every answer says whether DataGrout's + verifier accepted it, with a certificate showing the steps that produced it. + An answer it could not verify arrives as `verified: false`, with `answer` empty + and what the plan found under `unverified_answer` — never as a confident + answer — so a workflow can branch on it instead of trusting it. - **Memory that outlives the execution.** Facts stored in one run are still there in the next, and in other workflows pointed at the same memory. - **Compute that stays next to the data.** A large result never has to travel @@ -36,7 +39,18 @@ No runtime dependencies. | **Answer** | Ask | Ask in plain language. DataGrout plans, runs and verifies, and returns the answer plus a certificate URL. | | **Data** | Transform | Describe the result you want; DataGrout computes it on its own servers and returns records, one item per row. | | **Memory** | Recall / Remember | Store facts as workflows run, then ask what is known — including what can be inferred from it. | -| **Skill** | Run | Re-run work DataGrout has already verified. No planning, so it is fast and repeatable. | +| **Skill** | Run | Re-run work DataGrout has already verified — a saved skill (`skill_…`), or the Skill ID an Ask returned. No planning, so it is fast and repeatable. | + +### What Ask returns + +| Field | | +|---|---| +| `answer` | The answer, or empty when it could not be verified | +| `verified` | Whether DataGrout's verifier accepted it | +| `unverified_answer`, `caveat` | What the plan found, and why it could not be verified | +| `certificate` | `id` and `url` of the certificate showing exactly what ran | +| `skill_id` | Run this verified work again with **Skill → Run**. An id beginning `skill_` does not expire; when `skill_expires` is true it lasts an hour. | +| `usage` | Credits charged and remaining | ### Why Transform is worth using @@ -115,18 +129,15 @@ Choose either on the node's **Authentication** field. | Gateway Base URL | Leave as-is unless you self-host the gateway | **DataGrout OAuth2 API** — for instances that prefer an OAuth flow. Create the -credential, click **Connect my account**, and there is nothing to fill in. +credential, click **Connect my account**, and there is nothing to fill in: the +token itself names the server it was granted. -Testing a credential only reads — it lists the tools your server exposes and -changes nothing. +Testing a credential only reads — it fetches the server's name, changes nothing +and costs nothing. -DataGrout enables transports per server, and a new server has only MCP switched -on. Rather than sending you to the dashboard, these nodes switch on the transport -they need the first time a workflow actually uses it: the DataGrout node enables -JSON-RPC on its first call, and the trigger enables WebSocket when it starts. -Both are idempotent, and both are written to the n8n log so a change to your -server's settings is never silent. Until then, a transport you have deliberately -left off stays off. +The trigger listens over DataGrout's WebSocket transport, which a new server has +switched off. It switches it on when the trigger starts, and writes that to the +n8n log, so a change to your server's settings is never silent. ## Using it with an AI Agent @@ -137,9 +148,12 @@ question. Replace a field with a fixed value to pin it. ## Options - **Wait for Result** — DataGrout moves slow work to the background; the node - waits this long for the finished result. Set 0 to return immediately. -- **Full Response** — return everything DataGrout sent rather than the tidied - answer. + waits this long for the finished result. Set 0 to return immediately with the + task to follow. + +On Transform, **Split Rows Into Items** (on by default) outputs one item per row +of the result. Turn it off to get the whole response as one item, with its +certificate and reference. On the trigger: diff --git a/credentials/DataGroutApi.credentials.ts b/credentials/DataGroutApi.credentials.ts index bd41d28..cdf542f 100644 --- a/credentials/DataGroutApi.credentials.ts +++ b/credentials/DataGroutApi.credentials.ts @@ -49,22 +49,15 @@ export class DataGroutApi implements ICredentialType { }, }; - // A credential test answers "are these credentials valid?", so it reads and - // changes nothing. `tools.list` settles both halves of the credential: a bad - // token is rejected by the gateway's auth plug (401/403), and an unknown - // Server ID is a 404. - // - // A server that has not yet switched on the JSON-RPC transport answers HTTP - // 200 with a JSON-RPC error in the body, which n8n reads as a pass — rightly, - // because that is a server setting rather than a bad credential. The node - // switches the transport on when a workflow actually runs, not when someone - // clicks Test. + // A credential test answers "are these credentials valid?", so it reads, + // changes nothing and costs nothing. The server endpoint settles both halves + // of the credential: a bad token is rejected by the gateway's auth plug + // (401/403), and an unknown Server ID is a 404. test: ICredentialTestRequest = { request: { baseURL: '={{$credentials.baseUrl}}', - url: '=/servers/{{$credentials.serverId}}/rpc', - method: 'POST', - body: { jsonrpc: '2.0', id: 1, method: 'tools.list', params: {} }, + url: '=/api/v1/servers/{{$credentials.serverId}}', + method: 'GET', }, }; } diff --git a/nodes/DataGrout/DataGrout.node.ts b/nodes/DataGrout/DataGrout.node.ts index b364efb..b44edb0 100644 --- a/nodes/DataGrout/DataGrout.node.ts +++ b/nodes/DataGrout/DataGrout.node.ts @@ -7,23 +7,8 @@ import type { } from 'n8n-workflow'; import { NodeConnectionTypes, NodeOperationError } from 'n8n-workflow'; -import { - parseJsonData, - parseJsonObject, - payloadError, - shapeAnswer, - toolPayload, - unwrapData, -} from './pure'; -import { DEFAULT_TASK_WAIT_MS, callTool } from './transport'; - -// The tools behind each operation. Users never see these — they pick an -// operation, and the node knows which DataGrout capability serves it. -const ASK = 'data-grout@1/discovery.plan@1'; -const TRANSFORM = 'data-grout@1/prism.refract@1'; -const REMEMBER = 'data-grout@1/logic.remember@1'; -const RECALL = 'data-grout@1/logic.query@1'; -const RUN_SKILL = 'data-grout@1/discovery.perform@1'; +import { apiPath, parseJsonData, parseJsonObject, rowsOf } from './pure'; +import { DEFAULT_WAIT_MS, call } from './transport'; // Free-text fields default to $fromAI so the node works on an AI Agent's tool // connector with no extra setup; a workflow user overwrites the field with a @@ -176,6 +161,15 @@ export class DataGrout implements INodeType { displayOptions: { show: { resource: ['data'], inputMode: ['cacheRef'] } }, description: 'The reference returned by an earlier DataGrout step', }, + { + displayName: 'Split Rows Into Items', + name: 'splitRows', + type: 'boolean', + default: true, + displayOptions: { show: { resource: ['data'] } }, + description: + 'Whether to output one item per row of the result. Turn off to get the whole response as one item, with its certificate and reference.', + }, // ── Memory ────────────────────────────────────────────────── { @@ -224,10 +218,11 @@ export class DataGrout implements INodeType { displayName: 'Memory Name', name: 'namespace', type: 'string', - default: '', + required: true, + default: 'default', displayOptions: { show: { resource: ['memory'] } }, description: - 'Keeps separate sets of facts apart — e.g. one per customer or project. Leave empty for the default memory.', + 'Keeps separate sets of facts apart — e.g. one per customer or project. "default" is the memory DataGrout uses when none is named.', }, // ── Skill ─────────────────────────────────────────────────── @@ -255,7 +250,8 @@ export class DataGrout implements INodeType { required: true, default: '', displayOptions: { show: { resource: ['skill'] } }, - description: 'The skill to run, as returned by an earlier Ask', + description: + 'A saved skill\'s ID (skill_…), or the Skill ID an earlier Ask returned. A saved skill does not expire.', }, { displayName: 'Inputs', @@ -278,17 +274,9 @@ export class DataGrout implements INodeType { displayName: 'Wait for Result (Ms)', name: 'waitForResult', type: 'number', - default: DEFAULT_TASK_WAIT_MS, - description: - 'DataGrout moves slow work to the background; the node waits this long for the finished result. Set 0 to return immediately.', - }, - { - displayName: 'Full Response', - name: 'fullResponse', - type: 'boolean', - default: false, + default: DEFAULT_WAIT_MS, description: - 'Whether to return everything DataGrout sent, instead of the tidied answer', + 'DataGrout moves slow work to the background; the node waits this long for the finished result. Set 0 to return immediately with the task to follow.', }, ], }, @@ -303,61 +291,25 @@ export class DataGrout implements INodeType { try { const resource = this.getNodeParameter('resource', i) as string; const options = this.getNodeParameter('options', i, {}) as IDataObject; - const waitMs = (options.waitForResult as number) ?? DEFAULT_TASK_WAIT_MS; - const full = (options.fullResponse as boolean) ?? false; + const waitMs = (options.waitForResult as number) ?? DEFAULT_WAIT_MS; - const json = (raw: unknown, field: string) => { - try { - return parseJsonObject(raw); - } catch (e) { - throw new NodeOperationError(this.getNode(), `${field}: ${(e as Error).message}`, { - itemIndex: i, - }); + const text = (name: string, label: string): string => { + const value = String(this.getNodeParameter(name, i, '') ?? '').trim(); + if (!value) { + throw new NodeOperationError(this.getNode(), `${label} is empty`, { itemIndex: i }); } + return value; }; - let result: IDataObject; + const emit = (json: IDataObject) => out.push({ json, pairedItem: { item: i } }); if (resource === 'answer') { - const question = String(this.getNodeParameter('question', i, '') ?? '').trim(); - if (!question) { - throw new NodeOperationError(this.getNode(), 'Question is empty', { itemIndex: i }); - } - result = await callTool( - this, - ASK, - { goal: question, execute: true, verify: true, lean: true, head: true }, - i, - waitMs, - ); - const payload = toolPayload(result); - const failure = payloadError(payload); - if (failure) { - throw new NodeOperationError(this.getNode(), `DataGrout: ${failure}`, { - itemIndex: i, - }); - } - out.push({ - json: full ? payload : shapeAnswer(payload), - pairedItem: { item: i }, - }); - continue; - } - - if (resource === 'data') { - const goal = String(this.getNodeParameter('goal', i, '') ?? '').trim(); - if (!goal) { - throw new NodeOperationError(this.getNode(), 'Goal is empty', { itemIndex: i }); - } - const mode = this.getNodeParameter('inputMode', i, 'payload') as string; - const args: IDataObject = { goal }; + emit(await call(this, 'POST', '/answers', { question: text('question', 'Question') }, i, waitMs)); + } else if (resource === 'data') { + const body: IDataObject = { goal: text('goal', 'Goal') }; - if (mode === 'cacheRef') { - const ref = String(this.getNodeParameter('cacheRef', i, '') ?? '').trim(); - if (!ref) { - throw new NodeOperationError(this.getNode(), 'Reference is empty', { itemIndex: i }); - } - args.cache_ref = ref; + if (this.getNodeParameter('inputMode', i, 'payload') === 'cacheRef') { + body.reference = text('cacheRef', 'Reference'); } else { let data: unknown; try { @@ -367,87 +319,44 @@ export class DataGrout implements INodeType { itemIndex: i, }); } - args.payload = data ?? items[i].json; + body.data = (data ?? items[i].json) as IDataObject; } - result = await callTool(this, TRANSFORM, args, i, waitMs); + const transform = await call(this, 'POST', '/transforms', body, i, waitMs); + const split = this.getNodeParameter('splitRows', i, true) as boolean; - const transformFailure = payloadError(toolPayload(result)); - if (transformFailure) { - throw new NodeOperationError(this.getNode(), `DataGrout: ${transformFailure}`, { - itemIndex: i, - }); - } - - if (!full) { - const value = unwrapData(toolPayload(result)); - if (Array.isArray(value)) { - for (const row of value) { - out.push({ - json: (row && typeof row === 'object' && !Array.isArray(row) - ? row - : { result: row }) as IDataObject, - pairedItem: { item: i }, - }); - } - } else { - out.push({ - json: (value && typeof value === 'object' && !Array.isArray(value) - ? value - : { result: value }) as IDataObject, - pairedItem: { item: i }, - }); - } - continue; + if (split && transform.status !== 'running') { + rowsOf(transform.result).forEach(emit); + } else { + emit(transform); } } else if (resource === 'memory') { - const operation = this.getNodeParameter('operation', i) as string; - const namespace = String(this.getNodeParameter('namespace', i, '') ?? '').trim(); - const args: IDataObject = {}; - if (namespace) args.namespace = namespace; + const name = text('namespace', 'Memory Name'); - if (operation === 'remember') { - const statement = String(this.getNodeParameter('statement', i, '') ?? '').trim(); - if (!statement) { - throw new NodeOperationError(this.getNode(), 'Fact is empty', { itemIndex: i }); - } - args.statement = statement; - result = await callTool(this, REMEMBER, args, i, waitMs); + if (this.getNodeParameter('operation', i) === 'remember') { + const statement = text('statement', 'Fact'); + emit(await call(this, 'POST', apiPath('memories', name, 'facts'), { statement }, i, waitMs)); } else { - const question = String( - this.getNodeParameter('memoryQuestion', i, '') ?? '', - ).trim(); - if (!question) { - throw new NodeOperationError(this.getNode(), 'Question is empty', { itemIndex: i }); - } - args.question = question; - result = await callTool(this, RECALL, args, i, waitMs); + const question = text('memoryQuestion', 'Question'); + emit(await call(this, 'POST', apiPath('memories', name, 'recall'), { question }, i, waitMs)); } } else if (resource === 'skill') { - const handle = String(this.getNodeParameter('skillHandle', i, '') ?? '').trim(); - if (!handle) { - throw new NodeOperationError(this.getNode(), 'Skill is empty', { itemIndex: i }); + let inputs: IDataObject | undefined; + try { + inputs = parseJsonObject(this.getNodeParameter('skillInputs', i, '')); + } catch (e) { + throw new NodeOperationError(this.getNode(), `Inputs: ${(e as Error).message}`, { + itemIndex: i, + }); } - const inputs = json(this.getNodeParameter('skillInputs', i, ''), 'Inputs'); - const args: IDataObject = { skill_handle: handle }; - if (inputs) args.args = inputs; - result = await callTool(this, RUN_SKILL, args, i, waitMs); + + const path = apiPath('skills', text('skillHandle', 'Skill'), 'runs'); + emit(await call(this, 'POST', path, inputs ? { inputs } : {}, i, waitMs)); } else { throw new NodeOperationError(this.getNode(), `Unknown resource "${resource}"`, { itemIndex: i, }); } - - const payload = toolPayload(result); - const failure = payloadError(payload); - if (failure) { - throw new NodeOperationError(this.getNode(), `DataGrout: ${failure}`, { itemIndex: i }); - } - - out.push({ - json: full ? result : payload, - pairedItem: { item: i }, - }); } catch (error) { if (this.continueOnFail()) { out.push({ json: { error: (error as Error).message }, pairedItem: { item: i } }); diff --git a/nodes/DataGrout/pure.ts b/nodes/DataGrout/pure.ts index 62fa753..fd3a9fa 100644 --- a/nodes/DataGrout/pure.ts +++ b/nodes/DataGrout/pure.ts @@ -5,179 +5,109 @@ import type { IDataObject } from 'n8n-workflow'; // unit-tested directly. // ──────────────────────────────────────────────────────────────────── -/** JSON-RPC error code DataGrout returns for a malformed or refused request. */ -export const INVALID_REQUEST = -32600; +// ── REST API ───────────────────────────────────────────────────────── + +export const DEFAULT_GATEWAY = 'https://gateway.datagrout.ai'; /** - * DataGrout gates transports per server: `interaction_config.enabled_protocols` - * defaults to `["mcp"]`, and the JSON-RPC endpoint refuses calls until - * `jsonrpc` is added. The refusal is recognisable, which lets the node enable - * the protocol itself rather than sending the user to a dashboard toggle. + * The gateway's origin from either credential: the token credential stores a + * base URL, the OAuth2 credential the `/connect` URL it authorises against. */ -export function isProtocolDisabled(error: IDataObject | undefined): boolean { - if (!error) return false; - const code = error.code; - const message = String(error.message ?? '').toLowerCase(); - return code === INVALID_REQUEST && message.includes('json-rpc not enabled'); +export function gatewayOrigin(credentials: IDataObject): string { + for (const candidate of [credentials.baseUrl, credentials.serverUrl]) { + if (typeof candidate !== 'string' || !candidate.trim()) continue; + try { + return new URL(candidate.trim()).origin; + } catch { + // not a URL; try the next + } + } + return DEFAULT_GATEWAY; } /** - * A long-running DataGrout call detaches to a background task and answers with - * `{status: "detached", task_ref}`. Returns that reference, else undefined. + * Where the API lives for a credential. An access token belongs to one server, + * which the path names. An OAuth token already names the server it was + * granted, so its requests use the server-less form and the gateway resolves + * the server from the token. */ -export function detachedTaskRef(result: IDataObject): string | undefined { - const sc = (result.structuredContent as IDataObject) ?? result ?? {}; - if (sc.status === 'detached' && typeof sc.task_ref === 'string') return sc.task_ref; - return undefined; +export function apiBase(origin: string, serverId: unknown): string { + const id = typeof serverId === 'string' ? serverId.trim() : ''; + return id ? `${origin}/api/v1/servers/${encodeURIComponent(id)}` : `${origin}/api/v1`; } +/** DataGrout holds a request for at most this long when asked to wait. */ +export const MAX_SERVER_WAIT_SECONDS = 30; + /** - * The task record inside a `tasks.wait` reply. A direct tool call returns it at - * the top of the payload; the perform wrapper nests it under `.result`. Both - * shapes occur live, so support both. + * A path under the server's REST API, with each segment encoded — a memory name + * or skill id typed by a user must not be able to reach a different endpoint. */ -export function taskRecord(payload: IDataObject | undefined): IDataObject { - const p = payload ?? {}; - if (typeof p.completed !== 'undefined' || p.task_ref) return p; - return (p.result as IDataObject) ?? {}; -} - -/** Unwrap the tool payload from a JSON-RPC `tools.call` result. */ -export function toolPayload(result: IDataObject): IDataObject { - const sc = result.structuredContent; - if (sc && typeof sc === 'object' && !Array.isArray(sc)) { - const inner = (sc as IDataObject).result; - if (inner && typeof inner === 'object' && !Array.isArray(inner)) return inner as IDataObject; - return sc as IDataObject; - } - return result; +export function apiPath(...segments: string[]): string { + return segments.map((segment) => `/${encodeURIComponent(segment)}`).join(''); } /** - * Some tools answer with a whole nested tool response under `data` — the - * computed rows sit in `data.structuredContent.data`, or as a JSON string in - * `data.content[0].text` (live-verified against prism.refract over JSON-RPC). - * Dig the useful value out so a workflow gets records, not an envelope. + * The `Prefer: wait=N` value (RFC 7240) asking DataGrout to hold a request until + * slow work finishes, or undefined when the caller does not want to wait. */ -export function unwrapData(payload: IDataObject): unknown { - const nested = payload.data; - if (!nested || typeof nested !== 'object' || Array.isArray(nested)) return payload; - const inner = nested as IDataObject; - - const structured = inner.structuredContent; - if (structured && typeof structured === 'object' && !Array.isArray(structured)) { - const value = (structured as IDataObject).data ?? (structured as IDataObject).result; - if (value !== undefined) return value; - } - - const content = inner.content; - if (Array.isArray(content)) { - const block = content.find((c) => (c as IDataObject)?.type === 'text') as - | IDataObject - | undefined; - if (typeof block?.text === 'string') { - try { - return JSON.parse(block.text); - } catch { - return block.text; - } - } - } - - return payload; +export function waitPreference(waitMs: number): string | undefined { + if (!(waitMs > 0)) return undefined; + return `wait=${Math.min(MAX_SERVER_WAIT_SECONDS, Math.ceil(waitMs / 1000))}`; } /** - * The failure message when a tool payload represents an error, else undefined. - * - * DataGrout reports tool-level failures inside a 200 response — an `error` - * string with a `status_code`. Without this check the node would hand a - * workflow a null answer marked verified (live-caught 2026-08-27), which is - * the one outcome this integration must never produce. + * Where to poll background work, from a `202 Accepted` body. Only a path on the + * API is accepted: it is joined to the gateway's own base URL, so a value that + * pointed anywhere else could carry the credential with it. */ +export function taskLocation(body: unknown): string | undefined { + const location = (body as IDataObject | undefined)?.location; + if (typeof location !== 'string') return undefined; + return /^\/api\/v1\/(servers\/[^/]+\/)?tasks\/[^/]+$/.test(location) ? location : undefined; +} + /** - * The message for a DataGrout loop-guard intervention, else undefined. - * - * DataGrout guards against repeated identical calls: an identical read issued - * several times with nothing changing in between, or a write repeated with the - * same arguments, is answered with an explanation rather than run again. - * - * Two of those replies are not marked as errors, so without this check they - * reach a workflow as though they were the answer — a downstream node expecting - * rows receives `{loop_detected: true, …}` instead (live-caught 2026-09-01). - * Nothing ran, so the honest outcome is a failure that says what happened. + * The message to show for an error response. The API's own errors carry + * `{error: {message}}`; the gateway's rate limiter and authentication use a + * flatter shape, and a proxy in between may send plain text. */ -export function interventionNotice(payload: IDataObject | undefined): string | undefined { - if (!payload || payload.loop_detected !== true) return undefined; - - const count = payload.execution_count; - const times = typeof count === 'number' ? `${count} times` : 'more than once'; - - // These read as the tail of the node's own "DataGrout: " prefix, so they do - // not repeat the product name. - if (payload.action === 'confirm_or_change_args') { - return ( - `this action was already called ${times} in this session with identical ` + - 'arguments, so it was not repeated and the earlier result still stands. Change the ' + - 'arguments if it genuinely needs to happen again.' - ); +export function apiErrorMessage(body: unknown, status: number): string { + if (body && typeof body === 'object') { + const b = body as IDataObject; + const error = b.error; + if (error && typeof error === 'object') { + const message = (error as IDataObject).message; + if (typeof message === 'string' && message) return message; + } + if (typeof b.message === 'string' && b.message) return b.message; + if (typeof error === 'string' && error) return error; } - - return ( - `this identical call ran ${times} with nothing changing in between, so the loop ` + - 'guard stopped it and no new result was produced. Vary the input, or raise the ' + - "loop-guard thresholds in this server's interaction settings." - ); + if (typeof body === 'string' && body.trim() && body.length <= 300) return body.trim(); + return `DataGrout returned HTTP ${status}`; } -export function payloadError(payload: IDataObject | undefined): string | undefined { - if (!payload) return undefined; - - for (const candidate of [payload, payload.data as IDataObject]) { - if (!candidate || typeof candidate !== 'object' || Array.isArray(candidate)) continue; - - const intervention = interventionNotice(candidate); - if (intervention) return intervention; - - const err = candidate.error; - if (typeof err === 'string' && err.trim()) return err; - if (err && typeof err === 'object') { - const message = (err as IDataObject).message; - return typeof message === 'string' ? message : JSON.stringify(err); - } - - const status = candidate.status_code; - if (typeof status === 'number' && status >= 400) { - return `DataGrout returned status ${status}`; - } +/** The stable error type, for continueOnFail output a workflow can branch on. */ +export function apiErrorType(body: unknown): string | undefined { + const error = (body as IDataObject | undefined)?.error; + if (error && typeof error === 'object') { + const type = (error as IDataObject).type; + if (typeof type === 'string') return type; } - - return undefined; + return typeof error === 'string' ? error : undefined; } /** - * Shape an Ask result for a workflow: the answer, whether DataGrout could - * verify it, and the certificate that shows exactly what ran. Everything the - * planner returned is preserved under `details` so nothing is lost. + * A transform's result as workflow items: one per row when it is a list of + * records, otherwise a single item. */ -export function shapeAnswer(payload: IDataObject): IDataObject { - const ctc = (payload.ctc as IDataObject) ?? {}; - const confidence = payload.answer_confidence; - - const failed = payloadError(payload) !== undefined; - - const out: IDataObject = { - answer: payload.result ?? payload.answer ?? null, - verified: !failed && confidence !== 'unverified', - details: payload, - }; - - if (typeof payload.hint === 'string' && confidence === 'unverified') out.caveat = payload.hint; - if (typeof ctc.url === 'string') out.certificateUrl = ctc.url; - if (typeof ctc.id === 'string') out.certificateId = ctc.id; - if (typeof payload.skill_handle === 'string') out.skill = payload.skill_handle; +export function rowsOf(result: unknown): IDataObject[] { + const asItem = (value: unknown): IDataObject => + value && typeof value === 'object' && !Array.isArray(value) + ? (value as IDataObject) + : ({ result: value } as IDataObject); - return out; + return Array.isArray(result) ? result.map(asItem) : [asItem(result)]; } /** @@ -265,7 +195,7 @@ export function matchesEvent(event: string, wanted: string[] | undefined): boole export function exampleEvent(event = 'run.completed'): IDataObject { const examples: Record = { 'run.completed': { - run_id: 79566, + run_id: 4127, execution_id: 'exec_1a2b3c', status: 'success', tool_name: 'discovery.plan', @@ -285,7 +215,7 @@ export function exampleEvent(event = 'run.completed'): IDataObject { status: 'failed', }, 'tool_call.failed': { - run_id: 79566, + run_id: 4127, execution_id: 'exec_1a2b3c', tool_name: 'logic.query', error: 'namespace not found', diff --git a/nodes/DataGrout/transport.ts b/nodes/DataGrout/transport.ts index d1689b2..5daadd7 100644 --- a/nodes/DataGrout/transport.ts +++ b/nodes/DataGrout/transport.ts @@ -1,204 +1,133 @@ import type { IDataObject, IExecuteFunctions, + IHttpRequestMethods, IHttpRequestOptions, - ILoadOptionsFunctions, + JsonObject, } from 'n8n-workflow'; -import { NodeApiError, NodeOperationError, sleep } from 'n8n-workflow'; +import { NodeApiError, sleep } from 'n8n-workflow'; -import { detachedTaskRef, isProtocolDisabled, taskRecord } from './pure'; +import { + apiBase, + apiErrorMessage, + apiErrorType, + gatewayOrigin, + taskLocation, + waitPreference, +} from './pure'; // ──────────────────────────────────────────────────────────────────── -// DataGrout JSON-RPC 2.0 transport. +// Client for DataGrout's REST API. // -// Built on n8n's own `helpers.httpRequestWithAuthentication` so the node +// Built on n8n's own `helpers.httpRequestWithAuthentication`, so the node // carries no runtime dependencies and inherits the instance's proxy and // TLS configuration. // ──────────────────────────────────────────────────────────────────── -export const DEFAULT_TIMEOUT_MS = 60_000; -export const DEFAULT_TASK_WAIT_MS = 120_000; +export const DEFAULT_WAIT_MS = 120_000; -type Ctx = IExecuteFunctions | ILoadOptionsFunctions; +const REQUEST_TIMEOUT_MS = 60_000; +const POLL_INTERVAL_MS = 2_000; -const TASKS_WAIT = 'data-grout@1/tasks.wait@1'; - -/** Which credential the node is configured to use. */ -async function authContext(ctx: Ctx, itemIndex = 0) { - const kind = - ((ctx as IExecuteFunctions).getNodeParameter?.('authentication', itemIndex, 'apiToken') as - | string - | undefined) ?? 'apiToken'; - const name = kind === 'oAuth2' ? 'dataGroutOAuth2Api' : 'dataGroutApi'; - const credentials = await ctx.getCredentials(name); - const baseUrl = String(credentials.baseUrl ?? 'https://gateway.datagrout.ai').replace(/\/$/, ''); - return { name, baseUrl, serverId: String(credentials.serverId ?? '') }; +interface Connection { + credential: string; + origin: string; + base: string; } -async function request( - ctx: Ctx, - options: IHttpRequestOptions, - credentialName: string, -): Promise { - return (await ctx.helpers.httpRequestWithAuthentication.call( - ctx, - credentialName, - options, - )) as IDataObject; +async function connection(ctx: IExecuteFunctions, itemIndex: number): Promise { + const kind = ctx.getNodeParameter('authentication', itemIndex, 'apiToken') as string; + const credential = kind === 'oAuth2' ? 'dataGroutOAuth2Api' : 'dataGroutApi'; + const credentials = await ctx.getCredentials(credential, itemIndex); + const origin = gatewayOrigin(credentials); + return { credential, origin, base: apiBase(origin, credentials.serverId) }; } -/** - * Turn the JSON-RPC transport on for this server. DataGrout ships with only - * MCP enabled, and this endpoint is idempotent and self-service, so the node - * can enable what it needs instead of asking the user to find a toggle. - * - * This changes a setting on the user's own server, so it is logged rather than - * done silently. It happens only when a workflow is actually running — never - * from a credential test — so that leaving the transport switched off stays - * meaningful right up to the moment someone asks DataGrout to do something. - */ -export async function enableJsonRpc(ctx: Ctx, itemIndex = 0): Promise { - const { name, baseUrl, serverId } = await authContext(ctx, itemIndex); - await request( - ctx, - { - method: 'POST', - url: `${baseUrl}/servers/${serverId}/interaction/enable_protocol`, - body: { protocol: 'jsonrpc' }, - json: true, - timeout: DEFAULT_TIMEOUT_MS, - }, - name, - ); - ctx.logger.info( - `[DataGrout] switched on the JSON-RPC transport for server ${serverId}, which was not enabled`, - ); -} - -let rpcId = 0; - -/** - * One JSON-RPC call. If DataGrout reports the transport is disabled, enable it - * and retry once — that turns a confusing first-run failure into a no-op. - */ -export async function rpc( - ctx: Ctx, - method: string, - params: IDataObject, - itemIndex = 0, - timeoutMs = DEFAULT_TIMEOUT_MS, -): Promise { - const { name, baseUrl, serverId } = await authContext(ctx, itemIndex); - - if (!serverId) { - throw new NodeOperationError( - ctx.getNode(), - 'The DataGrout credential is missing its Server ID', - { itemIndex }, - ); - } - - const send = async (): Promise => - await request( - ctx, - { - method: 'POST', - url: `${baseUrl}/servers/${serverId}/rpc`, - body: { jsonrpc: '2.0', id: ++rpcId, method, params }, - json: true, - timeout: timeoutMs, - }, - name, - ); - - let response = await send(); - - if (isProtocolDisabled(response.error as IDataObject)) { - await enableJsonRpc(ctx, itemIndex); - response = await send(); - } - - if (response.error) { - const err = response.error as IDataObject; - throw new NodeApiError(ctx.getNode(), response as never, { - message: `DataGrout: ${err.message ?? 'request failed'}`, - description: typeof err.data === 'string' ? err.data : undefined, +async function send( + ctx: IExecuteFunctions, + credential: string, + options: IHttpRequestOptions, + itemIndex: number, +): Promise<{ status: number; body: unknown }> { + const response = (await ctx.helpers.httpRequestWithAuthentication.call(ctx, credential, { + ...options, + json: true, + timeout: REQUEST_TIMEOUT_MS, + returnFullResponse: true, + ignoreHttpStatusErrors: true, + })) as { statusCode: number; body: unknown }; + + if (response.statusCode >= 400) { + const type = apiErrorType(response.body); + throw new NodeApiError(ctx.getNode(), (response.body ?? {}) as JsonObject, { + message: `DataGrout: ${apiErrorMessage(response.body, response.statusCode)}`, + description: type ? `Error type: ${type}` : undefined, + httpCode: String(response.statusCode), itemIndex, }); } - return (response.result as IDataObject) ?? {}; + return { status: response.statusCode, body: response.body }; } /** - * Call a DataGrout tool and return its payload, collecting the result of a - * background task if the call detaches. The caller never sees a task - * reference — a slow request simply takes longer. + * Call the API and return the finished resource. + * + * Slow work moves to a background task and the API answers 202. DataGrout is + * asked to hold the request for part of `waitMs` (`Prefer: wait`); if the work + * is still running, its task is polled for the rest. Past the deadline the + * node returns a `running` item naming the task rather than failing, so a + * workflow can pick the result up later. */ -export async function callTool( - ctx: Ctx, - toolName: string, - args: IDataObject, - itemIndex = 0, - waitMs = DEFAULT_TASK_WAIT_MS, -): Promise { - const result = await rpc(ctx, 'tools.call', { name: toolName, arguments: args }, itemIndex); - - if (result.isError) { - const content = (result.content as IDataObject[]) ?? []; - const text = (content.find((c) => c.type === 'text')?.text as string) ?? 'Tool failed'; - throw new NodeOperationError(ctx.getNode(), `DataGrout: ${text}`, { itemIndex }); - } - - const taskRef = detachedTaskRef((result.structuredContent as IDataObject) ?? result); - if (!taskRef || waitMs <= 0) return result; - - return await collect(ctx, taskRef, itemIndex, waitMs); -} - -async function collect( - ctx: Ctx, - taskRef: string, +export async function call( + ctx: IExecuteFunctions, + method: IHttpRequestMethods, + path: string, + body: IDataObject | undefined, itemIndex: number, - waitMs: number, + waitMs = DEFAULT_WAIT_MS, ): Promise { - const deadline = Date.now() + waitMs; - let ref = taskRef; + const { credential, origin, base } = await connection(ctx, itemIndex); + const deadline = Date.now() + Math.max(0, waitMs); + const prefer = waitPreference(waitMs); + + const first = await send( + ctx, + credential, + { method, url: base + path, body, headers: prefer ? { Prefer: prefer } : undefined }, + itemIndex, + ); + if (first.status !== 202) return (first.body ?? {}) as IDataObject; - while (Date.now() < deadline) { - const waited = await rpc( - ctx, - 'tools.call', - { name: TASKS_WAIT, arguments: { task_ref: ref } }, + const accepted = (first.body ?? {}) as IDataObject; + const location = taskLocation(accepted); + if (!location) { + throw new NodeApiError(ctx.getNode(), accepted as JsonObject, { + message: 'DataGrout: the work moved to the background but no task was given to follow', itemIndex, - ); - const task = taskRecord((waited.structuredContent as IDataObject) ?? waited); + }); + } - if (task.completed === true && task.result && typeof task.result === 'object') { - return { structuredContent: task.result as IDataObject }; - } - if (task.status === 'failed' || (task.error && task.completed === true)) { - return waited; + while (Date.now() + POLL_INTERVAL_MS <= deadline) { + await sleep(POLL_INTERVAL_MS); + const task = (await send(ctx, credential, { method: 'GET', url: origin + location }, itemIndex)) + .body as IDataObject; + + if (task.status === 'completed') return (task.result ?? {}) as IDataObject; + if (task.status === 'failed' || task.status === 'cancelled') { + throw new NodeApiError(ctx.getNode(), task as JsonObject, { + message: `DataGrout: ${apiErrorMessage(task, 422)}`, + itemIndex, + }); } - ref = (task.task_ref as string) ?? ref; - await sleep(1000); } - // Out of budget. Say so plainly rather than returning a task stub the - // caller has no way to act on. return { - structuredContent: { - status: 'running', - note: - 'DataGrout is still working on this in the background. Increase "Wait for Result" ' + - 'on this node, or run it again shortly — finished work is reused, so the retry is fast.', - }, + status: 'running', + task_id: accepted.task_id, + location: origin + location, + note: + 'DataGrout is still working on this in the background. Increase "Wait for Result" on ' + + 'this node, or fetch the task from its location later.', }; } - -/** Tool names the server exposes, for the credential test and dropdowns. */ -export async function listToolNames(ctx: Ctx): Promise { - const result = await rpc(ctx, 'tools.list', {}); - const tools = (result.tools as IDataObject[]) ?? []; - return tools.map((t) => String(t.name ?? '')).filter(Boolean); -} diff --git a/test/pure.test.ts b/test/pure.test.ts index 8588e06..0007826 100644 --- a/test/pure.test.ts +++ b/test/pure.test.ts @@ -2,150 +2,24 @@ import { describe, it } from 'node:test'; import assert from 'node:assert/strict'; import { - INVALID_REQUEST, - detachedTaskRef, - interventionNotice, - isProtocolDisabled, + apiBase, + apiErrorMessage, + apiErrorType, + apiPath, + gatewayOrigin, matchesEvent, parseJsonData, parseJsonObject, - payloadError, - shapeAnswer, + rowsOf, subscribeAck, subscriptionEvent, - taskRecord, - toolPayload, - unwrapData, + taskLocation, + waitPreference, } from '../nodes/DataGrout/pure.ts'; // DataGrout servers ship with only MCP enabled; the node detects the refusal // and turns JSON-RPC on itself, so this predicate gates the whole first-run // experience. -describe('isProtocolDisabled', () => { - it('recognises the transport-disabled refusal', () => { - assert.equal( - isProtocolDisabled({ code: INVALID_REQUEST, message: 'JSON-RPC not enabled for this server' }), - true, - ); - }); - - it('is case-insensitive on the message', () => { - assert.equal( - isProtocolDisabled({ code: INVALID_REQUEST, message: 'json-rpc NOT ENABLED for this server' }), - true, - ); - }); - - it('does not fire on other invalid-request errors', () => { - assert.equal(isProtocolDisabled({ code: INVALID_REQUEST, message: 'Invalid params' }), false); - }); - - it('does not fire on a different code with a similar message', () => { - assert.equal(isProtocolDisabled({ code: -32000, message: 'JSON-RPC not enabled' }), false); - }); - - it('handles a missing error', () => { - assert.equal(isProtocolDisabled(undefined), false); - }); -}); - -describe('detachedTaskRef', () => { - it('finds the reference when work detached', () => { - assert.equal(detachedTaskRef({ status: 'detached', task_ref: 'task_abc' }), 'task_abc'); - }); - - it('looks inside structuredContent too', () => { - assert.equal( - detachedTaskRef({ structuredContent: { status: 'detached', task_ref: 'task_xyz' } }), - 'task_xyz', - ); - }); - - it('returns undefined for a completed call', () => { - assert.equal(detachedTaskRef({ status: 'ready' }), undefined); - }); - - it('ignores a non-string reference', () => { - assert.equal(detachedTaskRef({ status: 'detached', task_ref: 7 }), undefined); - }); -}); - -// Two envelope shapes occur live: a direct tool call puts the record at the -// top, the perform wrapper nests it under .result. -describe('taskRecord', () => { - it('reads a top-level record', () => { - const p = { completed: true, status: 'completed', result: { answer: 1 } }; - assert.deepEqual(taskRecord(p), p); - }); - - it('reads a nested record', () => { - const inner = { completed: false, status: 'working' }; - assert.deepEqual(taskRecord({ result: inner }), inner); - }); - - it('treats a bare reference as the record', () => { - const p = { task_ref: 'task_abc' }; - assert.deepEqual(taskRecord(p), p); - }); - - it('copes with nothing', () => { - assert.deepEqual(taskRecord(undefined), {}); - assert.deepEqual(taskRecord({}), {}); - }); -}); - -describe('toolPayload', () => { - it('unwraps the doubly-wrapped result', () => { - const inner = { answer: 'yes', executed: true }; - assert.deepEqual(toolPayload({ structuredContent: { result: inner } }), inner); - }); - - it('falls back to structuredContent when there is no inner result', () => { - const sc = { status: 'ready', plan: {} }; - assert.deepEqual(toolPayload({ structuredContent: sc }), sc); - }); - - it('passes a bare result through', () => { - const bare = { rows: [1, 2] }; - assert.deepEqual(toolPayload(bare), bare); - }); -}); - -describe('shapeAnswer', () => { - it('surfaces the answer, verification and certificate', () => { - const shaped = shapeAnswer({ - result: [{ account: 'Acme' }], - answer_confidence: 'verified', - ctc: { id: 'ctc_abc', url: 'https://ctc.datagrout.ai/certs/ctc_abc' }, - skill_handle: 'vs_abc', - }); - assert.deepEqual(shaped.answer, [{ account: 'Acme' }]); - assert.equal(shaped.verified, true); - assert.equal(shaped.certificateUrl, 'https://ctc.datagrout.ai/certs/ctc_abc'); - assert.equal(shaped.certificateId, 'ctc_abc'); - assert.equal(shaped.skill, 'vs_abc'); - }); - - it('marks an unverified answer and carries the caveat', () => { - const shaped = shapeAnswer({ - result: [], - answer_confidence: 'unverified', - hint: 'Could not confirm the Jira leg ran', - }); - assert.equal(shaped.verified, false); - assert.equal(shaped.caveat, 'Could not confirm the Jira leg ran'); - }); - - it('never loses the original payload', () => { - const payload = { result: 1, extra: 'kept' }; - assert.deepEqual(shapeAnswer(payload).details, payload); - }); - - it('treats an absent confidence as verified', () => { - assert.equal(shapeAnswer({ result: 1 }).verified, true); - }); -}); - describe('parseJsonObject', () => { it('parses a typed JSON string', () => { assert.deepEqual(parseJsonObject('{"a":1}'), { a: 1 }); @@ -192,92 +66,6 @@ describe('parseJsonData', () => { // The exact envelope prism.refract returned over JSON-RPC, live-captured // 2026-08-27: the computed rows arrive as a JSON string inside a nested tool // response, which a workflow must never be handed raw. -describe('unwrapData', () => { - const rows = [ - { region: 'south', total_amount: 25 }, - { region: 'north', total_amount: 15 }, - ]; - - it('parses rows out of a nested text content block', () => { - const payload = { - _dg: { tool: 'data-grout@1/prism.refract@1' }, - data: { - content: [{ type: 'text', text: JSON.stringify(rows), annotations: {} }], - status_code: 200, - }, - status_code: 200, - }; - assert.deepEqual(unwrapData(payload), rows); - }); - - it('prefers a nested structuredContent value when present', () => { - const payload = { data: { structuredContent: { data: rows }, content: [] } }; - assert.deepEqual(unwrapData(payload), rows); - }); - - it('accepts a nested result key too', () => { - const payload = { data: { structuredContent: { result: rows } } }; - assert.deepEqual(unwrapData(payload), rows); - }); - - it('returns non-JSON text as text rather than throwing', () => { - const payload = { data: { content: [{ type: 'text', text: 'no rows found' }] } }; - assert.equal(unwrapData(payload), 'no rows found'); - }); - - it('leaves a payload without a nested response untouched', () => { - const payload = { count: 2, facts: [] }; - assert.equal(unwrapData(payload), payload); - }); - - it('ignores a non-object data field', () => { - const payload = { data: 'plain' }; - assert.equal(unwrapData(payload), payload); - }); -}); - -// DataGrout reports tool failures inside a 200 response. Live-caught: an Ask -// whose plan 400'd came back as a null answer marked verified — the exact -// confidently-wrong outcome this integration exists to avoid. -describe('payloadError', () => { - it('finds a top-level error string', () => { - assert.equal(payloadError({ error: 'No payload provided', status_code: 400 }), 'No payload provided'); - }); - - it('finds an error nested under data', () => { - assert.equal(payloadError({ data: { error: 'Bad request', status_code: 400 } }), 'Bad request'); - }); - - it('reads an error object message', () => { - assert.equal(payloadError({ error: { message: 'boom' } }), 'boom'); - }); - - it('falls back to a bad status code alone', () => { - assert.equal(payloadError({ status_code: 503 }), 'DataGrout returned status 503'); - }); - - it('is silent on success', () => { - assert.equal(payloadError({ status_code: 200, data: { status_code: 200 } }), undefined); - assert.equal(payloadError({ count: 2, facts: [] }), undefined); - assert.equal(payloadError(undefined), undefined); - }); - - it('ignores an empty error string', () => { - assert.equal(payloadError({ error: ' ' }), undefined); - }); -}); - -describe('shapeAnswer never marks a failure verified', () => { - it('reports verified false when the payload carries an error', () => { - const shaped = shapeAnswer({ error: 'No payload provided', status_code: 400 }); - assert.equal(shaped.verified, false); - assert.equal(shaped.answer, null); - }); -}); - -// Frame shapes captured from the live WebSocket transport 2026-08-27. The -// lifecycle case matters most: session.ready arrives on every connect, and -// emitting it would start the workflow whenever the socket reconnected. describe('subscriptionEvent', () => { it('emits a subscription event', () => { const frame = { @@ -378,77 +166,138 @@ describe('matchesEvent', () => { // interventions are not flagged as errors, so they would otherwise be handed to // a workflow as the answer. The read-tier one below is the exact payload a live // run produced on 2026-09-01. -describe('interventionNotice', () => { - it('catches the read-tier loop that arrives as a success', () => { - const live = { - action: 'possible_loop', - consequence_tier: 'read', - execution_count: 4, - loop_detected: true, - }; - const notice = interventionNotice(live); - assert.ok(notice); - assert.match(notice, /loop\s+guard stopped it/); - assert.match(notice, /4 times/); - }); - - it('catches the write-confirmation gate', () => { - const notice = interventionNotice({ - loop_detected: true, - consequence_tier: 'write', - execution_count: 2, - action: 'confirm_or_change_args', - original_receipt_id: 'rcpt_1', - }); - assert.ok(notice); - assert.match(notice, /was not repeated/); - assert.match(notice, /2 times/); + +// ── The REST API client's helpers ──────────────────────────────────────── + +describe('apiPath', () => { + it('joins segments under the API', () => { + assert.equal(apiPath('memories', 'warehouses', 'facts'), '/memories/warehouses/facts'); }); - it('catches a write-tier loop', () => { - const notice = interventionNotice({ - loop_detected: true, - consequence_tier: 'write', - execution_count: 3, - action: 'use_previous_result', - }); - assert.ok(notice); + // A memory name or skill id is typed by a user; it must not be able to reach + // a different endpoint. + it('encodes each segment so a name cannot change the path', () => { + assert.equal(apiPath('memories', '../tasks/x', 'facts'), '/memories/..%2Ftasks%2Fx/facts'); + assert.equal(apiPath('skills', 'a b?c', 'runs'), '/skills/a%20b%3Fc/runs'); + }); +}); + +describe('waitPreference', () => { + it('asks DataGrout to hold the request, in whole seconds', () => { + assert.equal(waitPreference(12_500), 'wait=13'); }); - it('copes with a missing count', () => { - const notice = interventionNotice({ loop_detected: true, action: 'possible_loop' }); - assert.match(notice as string, /more than once/); + it('never asks for longer than DataGrout will hold a request', () => { + assert.equal(waitPreference(120_000), 'wait=30'); }); - it('is silent on an ordinary payload', () => { - assert.equal(interventionNotice({ result: 42 }), undefined); - assert.equal(interventionNotice({ loop_detected: false }), undefined); - assert.equal(interventionNotice(undefined), undefined); + it('asks for nothing when the node should not wait', () => { + assert.equal(waitPreference(0), undefined); + assert.equal(waitPreference(-1), undefined); }); }); -describe('payloadError surfaces an intervention as a failure', () => { - it('fails rather than handing the intervention to the workflow', () => { - const live = { - action: 'possible_loop', - consequence_tier: 'read', - execution_count: 4, - loop_detected: true, - }; - assert.match(payloadError(live) as string, /loop\s+guard stopped it/); +describe('taskLocation', () => { + it('accepts a task path in either form', () => { + assert.equal( + taskLocation({ location: '/api/v1/servers/5c3e9f10/tasks/task_abc' }), + '/api/v1/servers/5c3e9f10/tasks/task_abc', + ); + assert.equal(taskLocation({ location: '/api/v1/tasks/task_abc' }), '/api/v1/tasks/task_abc'); }); - it('finds one nested under data', () => { - assert.ok(payloadError({ data: { loop_detected: true, action: 'possible_loop' } })); + // It is joined to the gateway's base URL; anything else could carry the + // credential to another host or endpoint. + it('refuses anything that is not a task path on the API', () => { + assert.equal(taskLocation({ location: 'https://elsewhere.example.com/tasks/x' }), undefined); + assert.equal(taskLocation({ location: '//elsewhere.example.com/api/v1/tasks/x' }), undefined); + assert.equal(taskLocation({ location: '/api/v1/tasks/x/../../answers' }), undefined); + assert.equal(taskLocation({ location: '/oauth/token' }), undefined); + assert.equal(taskLocation({}), undefined); }); +}); - it('never reports an intervention as a verified answer', () => { - const shaped = shapeAnswer({ - loop_detected: true, - action: 'possible_loop', - execution_count: 5, - answer_confidence: 'high', - }); - assert.equal(shaped.verified, false); +describe('apiErrorMessage', () => { + it("reads the API's own error", () => { + const body = { error: { code: 409, type: 'loop_detected', message: 'This identical call ran 4 times.' } }; + assert.equal(apiErrorMessage(body, 409), 'This identical call ran 4 times.'); + }); + + it("reads the gateway's authentication error", () => { + assert.equal(apiErrorMessage({ error: { code: 401, message: 'Authentication required' } }, 401), 'Authentication required'); + }); + + it("reads the rate limiter's flatter shape", () => { + const body = { error: 'rate_limited', message: 'Too many requests. Limit: 100 per 60s.' }; + assert.equal(apiErrorMessage(body, 429), 'Too many requests. Limit: 100 per 60s.'); + }); + + it('falls back to the status when there is nothing readable', () => { + assert.equal(apiErrorMessage(undefined, 502), 'DataGrout returned HTTP 502'); + assert.equal(apiErrorMessage(''.repeat(100), 502), 'DataGrout returned HTTP 502'); + }); + + it('uses a short plain-text body from a proxy', () => { + assert.equal(apiErrorMessage('Bad Gateway', 502), 'Bad Gateway'); + }); +}); + +describe('apiErrorType', () => { + it('reads the stable type', () => { + assert.equal(apiErrorType({ error: { type: 'insufficient_credits' } }), 'insufficient_credits'); + assert.equal(apiErrorType({ error: 'rate_limited' }), 'rate_limited'); + assert.equal(apiErrorType({}), undefined); + }); +}); + +describe('rowsOf', () => { + it('makes one item per record', () => { + assert.deepEqual(rowsOf([{ city: 'North', total: 7 }, { city: 'South', total: 5 }]), [ + { city: 'North', total: 7 }, + { city: 'South', total: 5 }, + ]); + }); + + it('wraps values that are not records', () => { + assert.deepEqual(rowsOf([1, 2]), [{ result: 1 }, { result: 2 }]); + assert.deepEqual(rowsOf(42), [{ result: 42 }]); + }); + + it('keeps a single record as one item', () => { + assert.deepEqual(rowsOf({ total: 42 }), [{ total: 42 }]); + }); +}); + +describe('gatewayOrigin', () => { + it("reads the token credential's base URL", () => { + assert.equal(gatewayOrigin({ baseUrl: 'https://gateway.datagrout.ai/' }), 'https://gateway.datagrout.ai'); + }); + + it("reads the OAuth2 credential's /connect URL", () => { + assert.equal(gatewayOrigin({ serverUrl: 'https://gateway.datagrout.ai/connect' }), 'https://gateway.datagrout.ai'); + }); + + it('defaults to the hosted gateway', () => { + assert.equal(gatewayOrigin({}), 'https://gateway.datagrout.ai'); + assert.equal(gatewayOrigin({ baseUrl: 'not a url' }), 'https://gateway.datagrout.ai'); + }); +}); + +describe('apiBase', () => { + it('names the server when the credential has one', () => { + assert.equal( + apiBase('https://gateway.datagrout.ai', '5c3e9f10-7b2a-4d8e-9a61-2f0c4b7d8e31'), + 'https://gateway.datagrout.ai/api/v1/servers/5c3e9f10-7b2a-4d8e-9a61-2f0c4b7d8e31', + ); + }); + + // An OAuth token already names its server; the gateway resolves it. + it('uses the server-less form when it does not', () => { + assert.equal(apiBase('https://gateway.datagrout.ai', undefined), 'https://gateway.datagrout.ai/api/v1'); + assert.equal(apiBase('https://gateway.datagrout.ai', ' '), 'https://gateway.datagrout.ai/api/v1'); + }); + + it('encodes the id', () => { + assert.equal(apiBase('https://g.example.com', 'a/b'), 'https://g.example.com/api/v1/servers/a%2Fb'); }); }); diff --git a/test/session.test.ts b/test/session.test.ts index 36648d1..2747275 100644 --- a/test/session.test.ts +++ b/test/session.test.ts @@ -167,13 +167,13 @@ function tickingHarness(limit: number) { return { h, state }; } -// Frames captured from wss://gateway.datagrout.ai on 2026-08-27. +// Frame shapes captured from wss://gateway.datagrout.ai on 2026-08-27, with every id and value replaced. const SESSION_READY = { jsonrpc: '2.0', method: 'notification', params: { event: 'session.ready', - data: { session_id: 'ws_nH_7D6NqvPeoPNFW', subprotocol: 'datagrout-jsonrpc.v1' }, + data: { session_id: 'ws_Fict0000Session', subprotocol: 'datagrout-jsonrpc.v1' }, }, }; @@ -190,7 +190,7 @@ const event = (subscription: string, name: string, data: Record }); const RUN_COMPLETED = { - run_id: 79566, + run_id: 4127, execution_id: null, status: 'success', tool_name: 'data-grout@1/prism.refract@1', @@ -222,8 +222,8 @@ describe('startEventSession — connecting', () => { it('records the subscription id from the ack', () => { const h = harness(); h.socket().open(); - h.socket().deliver(ack('sub_9Pfi7pPohro')); - assert.ok(h.logs.info.some((m) => m.includes('sub_9Pfi7pPohro'))); + h.socket().deliver(ack('sub_Fict0001')); + assert.ok(h.logs.info.some((m) => m.includes('sub_Fict0001'))); h.close(); }); }); From 551dea8e295cff461d27f013e9190161133bae76 Mon Sep 17 00:00:00 2001 From: Nicholas Wright Date: Tue, 13 Oct 2026 09:12:29 -0600 Subject: [PATCH 2/2] fix: the trigger works with OAuth2 and switches on only WebSocket An OAuth2 credential has no Server ID, so activating the trigger with one failed. The trigger now asks the gateway which server the token belongs to, and reads the credential again before dialling so the socket gets a refreshed token. Activation also switched on the JSON-RPC transport, which only the old action node needed. It now enables WebSocket alone. The activation steps move beside the socket session so they can be tested without the n8n runtime. --- nodes/DataGrout/pure.ts | 70 +++++++++++ .../DataGroutTrigger/DataGroutTrigger.node.ts | 80 +++--------- test/trigger.test.ts | 118 ++++++++++++++++++ 3 files changed, 205 insertions(+), 63 deletions(-) create mode 100644 test/trigger.test.ts diff --git a/nodes/DataGrout/pure.ts b/nodes/DataGrout/pure.ts index fd3a9fa..c3e91d5 100644 --- a/nodes/DataGrout/pure.ts +++ b/nodes/DataGrout/pure.ts @@ -435,3 +435,73 @@ export function startEventSession(options: SessionOptions): Session { }, }; } + +// ── Trigger activation ────────────────────────────────────────────── + +export interface ActivationOptions { + credentials: IDataObject; + /** An authenticated request through n8n, which refreshes an expired OAuth2 token. */ + request: (method: 'GET' | 'POST', url: string, body?: IDataObject) => Promise; + /** The credentials as stored now, after any refresh `request` caused. */ + readCredentials: () => Promise; + log: Pick; +} + +export interface EventSocketTarget { + url: string; + token: string; +} + +function bearerOf(credentials: IDataObject): string | undefined { + const oauth = credentials.oauthTokenData as IDataObject | undefined; + const token = oauth?.access_token ?? credentials.apiToken; + return typeof token === 'string' && token ? token : undefined; +} + +/** + * Everything the trigger does before it dials: find the server, make sure it + * accepts WebSockets, and pick up the token to present. + */ +export async function prepareEventSocket(options: ActivationOptions): Promise { + const { credentials, request, log } = options; + const origin = gatewayOrigin(credentials); + + // An OAuth2 credential has no Server ID: the server is the one the user + // picked at consent, and the token names it. + let serverId = String(credentials.serverId ?? '').trim(); + if (!serverId) { + const found = await request('GET', `${origin}/api/v1/server`).then( + (server) => ({ id: String(server.id ?? '') }), + (error: Error) => ({ error }), + ); + if ('error' in found) { + throw new Error( + `Could not find which DataGrout server this credential belongs to: ${found.error.message}`, + ); + } + serverId = found.id; + if (!serverId) throw new Error('DataGrout did not say which server this credential belongs to'); + } + const server = `${origin}/servers/${encodeURIComponent(serverId)}`; + + // DataGrout gates transports per server. Switching WebSocket on here keeps + // setup out of the dashboard; it changes the user's own server, so it is + // logged, and it happens only when they activate the trigger. + try { + const result = await request('POST', `${server}/interaction/enable_protocol`, { protocol: 'ws' }); + const enabled = result?.enabled_protocols; + log.info( + `[DataGrout Trigger] WebSocket transport enabled on server ${serverId}` + + (Array.isArray(enabled) ? ` (now: ${enabled.join(', ')})` : ''), + ); + } catch (error) { + log.warn( + `[DataGrout Trigger] could not enable the WebSocket transport: ${(error as Error).message}`, + ); + } + + const token = bearerOf(await options.readCredentials()); + if (!token) throw new Error('The DataGrout credential has no token'); + + return { url: `${server.replace(/^http/, 'ws')}/ws`, token }; +} diff --git a/nodes/DataGroutTrigger/DataGroutTrigger.node.ts b/nodes/DataGroutTrigger/DataGroutTrigger.node.ts index 4f895bd..666437b 100644 --- a/nodes/DataGroutTrigger/DataGroutTrigger.node.ts +++ b/nodes/DataGroutTrigger/DataGroutTrigger.node.ts @@ -11,6 +11,7 @@ import { EVENTS_TOPIC, SUBPROTOCOL, exampleEvent, + prepareEventSocket, startEventSession, } from '../DataGrout/pure'; @@ -28,17 +29,6 @@ import { // read parameters, supply a real socket, hand n8n its two callbacks. // ──────────────────────────────────────────────────────────────────── -type Credentials = { - apiToken?: string; - serverId?: string; - baseUrl?: string; - oauthTokenData?: { access_token?: string }; -}; - -function bearer(credentials: Credentials): string | undefined { - return credentials.oauthTokenData?.access_token ?? credentials.apiToken; -} - // A trigger starts workflows and is never called by an agent. The property's // type only accepts `true`, so it is omitted rather than set to false. // eslint-disable-next-line @n8n/community-nodes/node-usable-as-tool @@ -165,7 +155,6 @@ export class DataGroutTrigger implements INodeType { const authentication = this.getNodeParameter('authentication', 'apiToken') as string; const credentialName = authentication === 'oAuth2' ? 'dataGroutOAuth2Api' : 'dataGroutApi'; - const credentials = (await this.getCredentials(credentialName)) as Credentials; const wanted = this.getNodeParameter('events', []) as string[]; const options = this.getNodeParameter('options', {}) as IDataObject; @@ -173,55 +162,20 @@ export class DataGroutTrigger implements INodeType { const reconnect = (options.reconnect as boolean) ?? true; const topic = String(options.topic ?? EVENTS_TOPIC).trim() || EVENTS_TOPIC; - const token = bearer(credentials); - if (!token) { - throw new NodeOperationError(this.getNode(), 'The DataGrout credential has no token'); - } - - const baseUrl = String(credentials.baseUrl ?? 'https://gateway.datagrout.ai').replace( - /\/$/, - '', - ); - const serverId = String(credentials.serverId ?? ''); - if (!serverId) { - throw new NodeOperationError( - this.getNode(), - 'The DataGrout credential is missing its Server ID', - ); - } - - // DataGrout gates transports per server, so make sure the WebSocket - // transport is on before dialling. Idempotent, and it keeps setup out of - // the dashboard. - // - // This changes a setting on the user's own server, so it is logged rather - // than done silently — and it happens on activation, when the user has - // asked for this trigger to run, rather than when they test a credential. - for (const protocol of ['ws', 'jsonrpc']) { - try { - const result = (await this.helpers.httpRequestWithAuthentication.call( - this, - credentialName, - { - method: 'POST', - url: `${baseUrl}/servers/${serverId}/interaction/enable_protocol`, - body: { protocol }, - json: true, - }, - )) as IDataObject; - const enabled = result?.enabled_protocols; - this.logger.info( - `[DataGrout Trigger] "${protocol}" transport enabled on server ${serverId}` + - (Array.isArray(enabled) ? ` (now: ${enabled.join(', ')})` : ''), - ); - } catch (error) { - this.logger.warn( - `[DataGrout Trigger] could not enable "${protocol}": ${(error as Error).message}`, - ); - } - } - - const wsUrl = `${baseUrl.replace(/^http/, 'ws')}/servers/${serverId}/ws`; + const target = await prepareEventSocket({ + credentials: await this.getCredentials(credentialName), + request: async (method, url, body) => + (await this.helpers.httpRequestWithAuthentication.call(this, credentialName, { + method, + url, + body, + json: true, + })) as IDataObject, + readCredentials: async () => await this.getCredentials(credentialName), + log: this.logger, + }).catch((error: Error) => { + throw new NodeOperationError(this.getNode(), error); + }); const session = startEventSession({ topic, @@ -235,8 +189,8 @@ export class DataGroutTrigger implements INodeType { // the bearer token reaches the upgrade request, which the standard API // gives no way to do. connect: () => - new WebSocket(wsUrl, { - headers: { Authorization: `Bearer ${token}` }, + new WebSocket(target.url, { + headers: { Authorization: `Bearer ${target.token}` }, protocols: [SUBPROTOCOL], } as unknown as string[]), openState: WebSocket.OPEN, diff --git a/test/trigger.test.ts b/test/trigger.test.ts new file mode 100644 index 0000000..109ec05 --- /dev/null +++ b/test/trigger.test.ts @@ -0,0 +1,118 @@ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; + +import { prepareEventSocket } from '../nodes/DataGrout/pure.ts'; +import type { ActivationOptions } from '../nodes/DataGrout/pure.ts'; + +// ──────────────────────────────────────────────────────────────────── +// What the trigger does before it dials: which server it resolves, what +// it changes on that server, and which token reaches the socket. +// ──────────────────────────────────────────────────────────────────── + +const SERVER = '5c3e9f10-7b2a-4d8e-9a61-2f0c4b7d8e31'; + +interface Request { + method: string; + url: string; + body?: unknown; +} + +function activation(opts: { + credentials: Array>; + respond?: (request: Request) => unknown; +}) { + const requests: Request[] = []; + const logged: string[] = []; + let reads = 0; + const read = async () => opts.credentials[Math.min(reads++, opts.credentials.length - 1)]; + + const options = async (): Promise => ({ + credentials: await read(), + request: async (method, url, body) => { + const request = { method, url, body }; + requests.push(request); + return (opts.respond?.(request) ?? {}) as Record; + }, + readCredentials: read, + log: { info: (m) => logged.push(m), warn: (m) => logged.push(m) }, + }); + + return { run: async () => prepareEventSocket(await options()), requests, logged }; +} + +describe('prepareEventSocket', () => { + it('switches on only the WebSocket transport', async () => { + const { run, requests } = activation({ credentials: [{ apiToken: 'tok_example', serverId: SERVER }] }); + + await run(); + + assert.deepEqual(requests, [ + { + method: 'POST', + url: `https://gateway.datagrout.ai/servers/${SERVER}/interaction/enable_protocol`, + body: { protocol: 'ws' }, + }, + ]); + }); + + it('dials the credential’s own server and base URL with its token', async () => { + const { run, requests } = activation({ + credentials: [{ apiToken: 'tok_example', serverId: SERVER, baseUrl: 'https://gw.example.com/' }], + }); + + const target = await run(); + + assert.equal(requests.some((r) => r.method === 'GET'), false); + assert.deepEqual(target, { url: `wss://gw.example.com/servers/${SERVER}/ws`, token: 'tok_example' }); + }); + + it('asks the gateway which server an OAuth2 credential belongs to', async () => { + const { run, requests } = activation({ + credentials: [ + { serverUrl: 'https://gateway.datagrout.ai/connect', oauthTokenData: { access_token: 'stale' } }, + { serverUrl: 'https://gateway.datagrout.ai/connect', oauthTokenData: { access_token: 'fresh' } }, + ], + respond: (r) => (r.method === 'GET' ? { id: SERVER, name: 'Example' } : {}), + }); + + const target = await run(); + + assert.deepEqual(requests[0], { method: 'GET', url: 'https://gateway.datagrout.ai/api/v1/server', body: undefined }); + assert.equal(target.url, `wss://gateway.datagrout.ai/servers/${SERVER}/ws`); + // The lookup may have refreshed the token; the socket gets the current one. + assert.equal(target.token, 'fresh'); + }); + + it('fails with a reason when the server cannot be found', async () => { + const { run } = activation({ + credentials: [{ oauthTokenData: { access_token: 'tok' } }], + respond: (r) => { + if (r.method === 'GET') throw new Error('401 Unauthorized'); + return {}; + }, + }); + + await assert.rejects(run(), /Could not find which DataGrout server.*401 Unauthorized/); + }); + + it('goes on to dial when the transport could not be switched on, and says so', async () => { + const { run, logged } = activation({ + credentials: [{ apiToken: 'tok_example', serverId: SERVER }], + respond: (r) => { + if (r.method === 'POST') throw new Error('403 Forbidden'); + return {}; + }, + }); + + const target = await run(); + + assert.equal(target.token, 'tok_example'); + assert.ok(logged.some((m) => m.includes('could not enable the WebSocket transport: 403 Forbidden'))); + }); + + it('refuses a credential with no token', async () => { + const { run } = activation({ credentials: [{ serverId: SERVER }] }); + + await assert.rejects(run(), /has no token/); + }); +});