diff --git a/.changeset/28040-connector-pull-input-schema.md b/.changeset/28040-connector-pull-input-schema.md new file mode 100644 index 00000000000..d03f9faad9e --- /dev/null +++ b/.changeset/28040-connector-pull-input-schema.md @@ -0,0 +1,22 @@ +--- +'@objectstack/service-automation': patch +--- + +fix(service-automation)!: a connector pull whose input breaks the read action's declared `inputSchema` is refused before the action is called, with `VALIDATION_ERROR` naming each parameter (#28040) + +Clause-②: yes (narrowing) +Metadata-protocol: none + + + +**BREAKING** (an accept-set narrowing at run time), shipped as `patch`: nothing a metadata author writes changes shape (`Metadata-protocol: none`), so the level follows the kind, and the banner and the ADR-0087 disposition stay. + +**FROM.** `pullConnectorSource` (the door a `job`'s `pull` runs) built the action's input from `connectorSource.input` plus the watermark parameter and called the connector's handler with it unchecked. The check the `connector_action` flow node gained in #27995 did not cover this door, so a pull whose input missed a parameter the action declares `required` still reached the remote API; for an OpenAPI `getPet`-style action the path placeholder was sent literally. + +**TO.** The input the pull is about to send, with the watermark parameter added, is judged against the action's `inputSchema` by the same check the `connector_action` node runs (one shared helper, so the two doors judge alike: `required`, `type` with OpenAPI 3.0 `nullable`, descent through `properties` and `items`; keywords outside that subset never refuse). A failing input throws `ConnectorPullError` with `code: 'VALIDATION_ERROR'`, `status: 400` and the new `reason: 'input_schema_mismatch'`, naming each problem by its dotted path (`missing required 'query.since'`, `'query.limit' must be integer, got string`); a text value where a number or boolean is declared adds a hint to write it as that JSON type (no CEL runs on this door, so there is no cast to suggest). The action is not called and nothing is written, so the job run is recorded `failed` and its `retryPolicy` applies. An action that declares no `inputSchema` pulls whatever it is given, as before. The `connector_action` node's behaviour and messages are unchanged. + +**TypeScript.** `ConnectorPullRefusalReason` gains `'input_schema_mismatch'`, and `ConnectorPullRegistry` now requires `getConnectorDescriptors()` (where the action's `inputSchema` is read). The automation engine already has it; a hand-built registry passed to `pullConnectorSource` adds it. + +**Who sees it.** A `mapping` whose `connectorSource.input` omits a parameter the read action declares `required`, or sends one of a type the action does not declare: its pull now fails before the call instead of the remote answering it. A required parameter that only the watermark supplies is absent on the first pull (nothing is stored yet), so that pull is refused too. + +**Fix.** Make `connectorSource.input` satisfy the action's `inputSchema`, which `GET /api/v1/automation/connectors` lists: supply every required parameter, and write each one as the declared type (`{ query: { limit: 50 } }`, not `{ query: { limit: '50' } }`). The watermark parameter is sent as the value stored in its target field, so the action must declare that parameter with the stored field's type (a text or datetime field ⇒ `string`, the ISO form a pull sends for a date), and must not declare it `required` unless `connectorSource.input` also gives it a first value. diff --git a/content/docs/automation/jobs.mdx b/content/docs/automation/jobs.mdx index 9b4e81df2ff..0c2a4c5c713 100644 --- a/content/docs/automation/jobs.mdx +++ b/content/docs/automation/jobs.mdx @@ -260,7 +260,8 @@ export const OrdersPullJob = defineJob({ [Data sync is defined on the target](https://github.com/objectstack-ai/objectstack/blob/main/packages/spec/docs/SYNC_ARCHITECTURE.md#data-sync-is-defined-on-the-target) for the watermark and the one-response limit. - **How a run is recorded.** A pull the platform refuses — the mapping is gone, - the connector is degraded, the upstream answered `ok: false` — rejects, so the + the connector is degraded, the input breaks the read action's declared + `inputSchema`, the upstream answered `ok: false` — rejects, so the run is `failed` and the `retryPolicy` applies. A pull whose rows the import runner refused completes as `degraded`, with the counts as its reason; retrying would refuse the same rows. Otherwise the run is `success`, including a pull diff --git a/packages/services/service-automation/src/builtin/connector-nodes.ts b/packages/services/service-automation/src/builtin/connector-nodes.ts index 3e42d7711f7..fa94d26e2e2 100644 --- a/packages/services/service-automation/src/builtin/connector-nodes.ts +++ b/packages/services/service-automation/src/builtin/connector-nodes.ts @@ -5,6 +5,7 @@ import { defineActionDescriptor } from '@objectstack/spec/automation'; import type { AutomationEngine, ConnectorActionContext } from '../engine.js'; import { refuseNode } from '../guard-refusal.js'; import { describeHold } from '../plugin-extension-api.js'; +import { findConnectorActionInputSchema, judgeConnectorInput, TEXT_FOR_SCALAR_HINT } from '../connector-input-schema.js'; import { resolveValueSlotMap } from './value-slot-map.js'; /** @@ -128,38 +129,24 @@ export function registerConnectorNodes(engine: AutomationEngine, ctx: PluginCont // hint: the EVALUATED input is judged against it here, after the // value slots resolved and before the handler is called, so a call // missing a required parameter (or carrying one of the wrong type) - // fails this step instead of reaching the remote API, whose answer - // to a malformed call is the remote's to choose. Read from the - // designer-facing descriptor — `inputSchema` is projected from the - // stored def unchanged, and it is the very schema an author picks - // the action by on `GET /connectors`, so what is enforced is what - // was shown. A `runtime` failure, not a guard: the input can come + // fails this step instead of reaching the remote API. The judgment + // is `connector-input-schema.ts`'s, the one the pull door runs too + // (#28040). A `runtime` failure, not a guard: the input can come // from run data, so a `fault` edge may handle it. - const inputSchema = engine - .getConnectorDescriptors() - .find((d) => d.name === cfg.connectorId) - ?.actions.find((a) => a.key === cfg.actionId)?.inputSchema; - const violations: InputSchemaViolations = { lines: [], textForScalar: false }; - collectInputSchemaViolations(inputSchema, input, '', violations); - if (violations.lines.length > 0) { - const shown = violations.lines.slice(0, MAX_NAMED_VIOLATIONS); - const more = violations.lines.length - shown.length; + const verdict = judgeConnectorInput( + findConnectorActionInputSchema(engine, cfg.connectorId, cfg.actionId), + input, + ); + if (verdict) { return { success: false, code: 'VALIDATION_ERROR', error: `connector_action '${node.id}': input for '${cfg.connectorId}.${cfg.actionId}' ` - + `does not match the action's inputSchema — ${shown.join('; ')}` - + (more > 0 ? `; and ${more} more` : '') + + `does not match the action's inputSchema — ${verdict.violations}` + `. Nothing was sent. Fix connectorConfig.input so it satisfies the inputSchema ` + `GET /api/v1/automation/connectors lists for this action.` - // Text is never converted here: the executor cannot tell - // a provider that serializes a section to strings from - // one that sends JSON, so the cast is the author's. - + (violations.textForScalar - ? ` A text value for an integer, number or boolean parameter is not converted: ` - + `cast it in a CEL envelope with int(), double() or bool().` - : ''), + + (verdict.textForScalar ? TEXT_FOR_SCALAR_HINT : ''), }; } @@ -196,103 +183,3 @@ export function registerConnectorNodes(engine: AutomationEngine, ctx: PluginCont ctx.logger.info('[Connector] 1 built-in node executor registered (connector_action)'); } - -/** What one judgment found: each violation as a sentence, and whether any was text where a number or boolean is declared. */ -interface InputSchemaViolations { - lines: string[]; - textForScalar: boolean; -} - -/** How many violations one refusal names before it summarises the rest as a count. */ -const MAX_NAMED_VIOLATIONS = 5; - -/** JSON Schema `type` names this check can judge; any other name leaves the value unjudged. */ -const JUDGED_TYPES = new Set(['string', 'number', 'integer', 'boolean', 'object', 'array', 'null']); - -function isPlainRecord(value: unknown): value is Record { - if (value === null || typeof value !== 'object' || Array.isArray(value)) return false; - const proto = Object.getPrototypeOf(value); - return proto === Object.prototype || proto === null; -} - -/** The value's JSON kind, or `undefined` for one with no JSON spelling to judge (a `Date`, a `Map`, …). */ -function jsonKindOf(value: unknown): string | undefined { - if (value === null) return 'null'; - if (Array.isArray(value)) return 'array'; - if (isPlainRecord(value)) return 'object'; - if (typeof value === 'bigint') return 'number'; - if (typeof value === 'string' || typeof value === 'number' || typeof value === 'boolean') return typeof value; - return undefined; -} - -function matchesType(type: string, value: unknown, kind: string): boolean { - if (type === 'integer') return typeof value === 'bigint' || (typeof value === 'number' && Number.isInteger(value)); - return type === kind; -} - -/** - * Collect where `value` breaks `schema` (#27995), each as a sentence naming the - * parameter by its dotted path. Judges a deliberate SUBSET of JSON Schema — - * `required`, `type` (with OpenAPI 3.0 `nullable`), and descent through - * `properties` and `items` — because the schemas reaching here are not ours: - * an OpenAPI document's operation (which may keep an unresolved `$ref`), an - * MCP server's `tools/list` entry (any draft), or a code connector's literal. - * A keyword outside the subset (`$ref`, `oneOf`/`anyOf`/`allOf`, `enum`, - * `format`, bounds, `additionalProperties`, …) never refuses a call: the - * remote still judges those, as it did before. A missing piece is the narrow - * direction on purpose — a call this check refuses is one the declared - * schema itself says is wrong under every draft. `null` on an optional - * property counts as absent (see the property loop), and text is never - * coerced to the number or boolean a parameter declares. - */ -function collectInputSchemaViolations(schema: unknown, value: unknown, path: string, out: InputSchemaViolations): void { - if (!isPlainRecord(schema)) return; - const kind = jsonKindOf(value); - if (kind === undefined) return; - const declared = typeof schema.type === 'string' ? [schema.type] : Array.isArray(schema.type) ? schema.type : []; - const types = declared.filter((t): t is string => typeof t === 'string'); - const judgeable = types.length > 0 && types.length === declared.length && types.every((t) => JUDGED_TYPES.has(t)); - if ( - judgeable - && !(kind === 'null' && schema.nullable === true) - && !types.some((t) => matchesType(t, value, kind)) - ) { - out.lines.push(`'${path || 'input'}' must be ${types.join(' or ')}, got ${kind}`); - if (kind === 'string' && types.some((t) => t === 'integer' || t === 'number' || t === 'boolean')) { - out.textForScalar = true; - } - return; - } - if (kind === 'object') { - const record = value as Record; - const properties = isPlainRecord(schema.properties) ? schema.properties : undefined; - const required = new Set( - Array.isArray(schema.required) ? schema.required.filter((k): k is string => typeof k === 'string') : [], - ); - for (const key of required) { - if (!Object.prototype.hasOwnProperty.call(record, key) || record[key] === undefined) { - const at = path ? `${path}.${key}` : key; - // A missing required SECTION names the parameters it requires - // too, so the first refusal already carries the parameter. - const sub = properties?.[key]; - const inner = isPlainRecord(sub) && Array.isArray(sub.required) - ? sub.required.filter((k): k is string => typeof k === 'string').map((k) => `'${at}.${k}'`) - : []; - out.lines.push(`missing required '${at}'${inner.length > 0 ? ` (with ${inner.join(', ')})` : ''}`); - } - } - if (properties) { - for (const [key, sub] of Object.entries(properties)) { - if (!Object.prototype.hasOwnProperty.call(record, key) || record[key] === undefined) continue; - // `null` on an OPTIONAL property is read as absent: a provider - // that serializes a section drops it, so the wire equals - // omission, which the schema permits. On a REQUIRED property - // it is still judged — it would be sent as a placeholder. - if (record[key] === null && !required.has(key)) continue; - collectInputSchemaViolations(sub, record[key], path ? `${path}.${key}` : key, out); - } - } - } else if (kind === 'array' && isPlainRecord(schema.items)) { - (value as unknown[]).forEach((item, i) => collectInputSchemaViolations(schema.items, item, `${path}[${i}]`, out)); - } -} diff --git a/packages/services/service-automation/src/connector-input-schema.ts b/packages/services/service-automation/src/connector-input-schema.ts new file mode 100644 index 00000000000..ff5952ddba5 --- /dev/null +++ b/packages/services/service-automation/src/connector-input-schema.ts @@ -0,0 +1,171 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * The connector action input check (#27995, #28040) — ONE judgment, shared by + * every door that dispatches a connector action: the `connector_action` flow + * node (`builtin/connector-nodes.ts`) and the connector pull door + * (`connector-pull.ts`). The action's declared `inputSchema` is a contract, + * not a hint: a call missing a required parameter (or carrying one of the + * wrong type) is refused before the handler is called, instead of reaching + * the remote API, whose answer to a malformed call is the remote's to choose. + * + * The schema is read from the designer-facing descriptor — + * `inputSchema` is projected from the stored def unchanged, and it is the very + * schema an author picks the action by on `GET /connectors`, so what is + * enforced is what was shown. Each door words its own refusal around the + * sentence {@link judgeConnectorInput} returns. + */ + +import type { ConnectorDescriptor } from '@objectstack/spec/integration'; + +/** Where a door reads an action's declared `inputSchema` — the automation engine's descriptor list. */ +export interface ConnectorInputSchemaSource { + getConnectorDescriptors(): ConnectorDescriptor[]; +} + +/** What one judgment found: the violations as one sentence, and whether any was text where a number or boolean is declared. */ +export interface ConnectorInputVerdict { + /** Up to {@link MAX_NAMED_VIOLATIONS} violations joined by `; `, then `; and N more` when there are others. */ + violations: string; + /** A text value met an `integer` / `number` / `boolean` parameter — the door appends {@link TEXT_FOR_SCALAR_HINT}. */ + textForScalar: boolean; +} + +/** + * The cast hint a refusal appends when {@link ConnectorInputVerdict.textForScalar} + * is set. Text is never converted here: the check cannot tell a provider that + * serializes a section to strings from one that sends JSON, so the cast is the + * author's. + */ +export const TEXT_FOR_SCALAR_HINT = + ` A text value for an integer, number or boolean parameter is not converted: ` + + `cast it in a CEL envelope with int(), double() or bool().`; + +/** The `inputSchema` `connectorId.actionId` declares, or `undefined` when it declares none (or is not registered). */ +export function findConnectorActionInputSchema( + source: ConnectorInputSchemaSource, + connectorId: string, + actionId: string, +): unknown { + return source + .getConnectorDescriptors() + .find((d) => d.name === connectorId) + ?.actions.find((a) => a.key === actionId)?.inputSchema; +} + +/** + * Judge `input` — the input the door is about to hand the handler, after + * every value it adds — against `inputSchema`. `undefined` means nothing is + * wrong (or the action declares no schema), so the call may be dispatched. + */ +export function judgeConnectorInput(inputSchema: unknown, input: unknown): ConnectorInputVerdict | undefined { + const found: InputSchemaViolations = { lines: [], textForScalar: false }; + collectInputSchemaViolations(inputSchema, input, '', found); + if (found.lines.length === 0) return undefined; + const shown = found.lines.slice(0, MAX_NAMED_VIOLATIONS); + const more = found.lines.length - shown.length; + return { + violations: shown.join('; ') + (more > 0 ? `; and ${more} more` : ''), + textForScalar: found.textForScalar, + }; +} + +/** What one judgment found: each violation as a sentence, and whether any was text where a number or boolean is declared. */ +interface InputSchemaViolations { + lines: string[]; + textForScalar: boolean; +} + +/** How many violations one refusal names before it summarises the rest as a count. */ +const MAX_NAMED_VIOLATIONS = 5; + +/** JSON Schema `type` names this check can judge; any other name leaves the value unjudged. */ +const JUDGED_TYPES = new Set(['string', 'number', 'integer', 'boolean', 'object', 'array', 'null']); + +function isPlainRecord(value: unknown): value is Record { + if (value === null || typeof value !== 'object' || Array.isArray(value)) return false; + const proto = Object.getPrototypeOf(value); + return proto === Object.prototype || proto === null; +} + +/** The value's JSON kind, or `undefined` for one with no JSON spelling to judge (a `Date`, a `Map`, …). */ +function jsonKindOf(value: unknown): string | undefined { + if (value === null) return 'null'; + if (Array.isArray(value)) return 'array'; + if (isPlainRecord(value)) return 'object'; + if (typeof value === 'bigint') return 'number'; + if (typeof value === 'string' || typeof value === 'number' || typeof value === 'boolean') return typeof value; + return undefined; +} + +function matchesType(type: string, value: unknown, kind: string): boolean { + if (type === 'integer') return typeof value === 'bigint' || (typeof value === 'number' && Number.isInteger(value)); + return type === kind; +} + +/** + * Collect where `value` breaks `schema` (#27995), each as a sentence naming the + * parameter by its dotted path. Judges a deliberate SUBSET of JSON Schema — + * `required`, `type` (with OpenAPI 3.0 `nullable`), and descent through + * `properties` and `items` — because the schemas reaching here are not ours: + * an OpenAPI document's operation (which may keep an unresolved `$ref`), an + * MCP server's `tools/list` entry (any draft), or a code connector's literal. + * A keyword outside the subset (`$ref`, `oneOf`/`anyOf`/`allOf`, `enum`, + * `format`, bounds, `additionalProperties`, …) never refuses a call: the + * remote still judges those, as it did before. A missing piece is the narrow + * direction on purpose — a call this check refuses is one the declared + * schema itself says is wrong under every draft. `null` on an optional + * property counts as absent (see the property loop), and text is never + * coerced to the number or boolean a parameter declares. + */ +function collectInputSchemaViolations(schema: unknown, value: unknown, path: string, out: InputSchemaViolations): void { + if (!isPlainRecord(schema)) return; + const kind = jsonKindOf(value); + if (kind === undefined) return; + const declared = typeof schema.type === 'string' ? [schema.type] : Array.isArray(schema.type) ? schema.type : []; + const types = declared.filter((t): t is string => typeof t === 'string'); + const judgeable = types.length > 0 && types.length === declared.length && types.every((t) => JUDGED_TYPES.has(t)); + if ( + judgeable + && !(kind === 'null' && schema.nullable === true) + && !types.some((t) => matchesType(t, value, kind)) + ) { + out.lines.push(`'${path || 'input'}' must be ${types.join(' or ')}, got ${kind}`); + if (kind === 'string' && types.some((t) => t === 'integer' || t === 'number' || t === 'boolean')) { + out.textForScalar = true; + } + return; + } + if (kind === 'object') { + const record = value as Record; + const properties = isPlainRecord(schema.properties) ? schema.properties : undefined; + const required = new Set( + Array.isArray(schema.required) ? schema.required.filter((k): k is string => typeof k === 'string') : [], + ); + for (const key of required) { + if (!Object.prototype.hasOwnProperty.call(record, key) || record[key] === undefined) { + const at = path ? `${path}.${key}` : key; + // A missing required SECTION names the parameters it requires + // too, so the first refusal already carries the parameter. + const sub = properties?.[key]; + const inner = isPlainRecord(sub) && Array.isArray(sub.required) + ? sub.required.filter((k): k is string => typeof k === 'string').map((k) => `'${at}.${k}'`) + : []; + out.lines.push(`missing required '${at}'${inner.length > 0 ? ` (with ${inner.join(', ')})` : ''}`); + } + } + if (properties) { + for (const [key, sub] of Object.entries(properties)) { + if (!Object.prototype.hasOwnProperty.call(record, key) || record[key] === undefined) continue; + // `null` on an OPTIONAL property is read as absent: a provider + // that serializes a section drops it, so the wire equals + // omission, which the schema permits. On a REQUIRED property + // it is still judged — it would be sent as a placeholder. + if (record[key] === null && !required.has(key)) continue; + collectInputSchemaViolations(sub, record[key], path ? `${path}.${key}` : key, out); + } + } + } else if (kind === 'array' && isPlainRecord(schema.items)) { + (value as unknown[]).forEach((item, i) => collectInputSchemaViolations(schema.items, item, `${path}[${i}]`, out)); + } +} diff --git a/packages/services/service-automation/src/connector-pull.test.ts b/packages/services/service-automation/src/connector-pull.test.ts index 925efedbeb5..ad897637ab1 100644 --- a/packages/services/service-automation/src/connector-pull.test.ts +++ b/packages/services/service-automation/src/connector-pull.test.ts @@ -59,6 +59,8 @@ function harness(opts: { provider?: string | undefined; result?: unknown; stored?: Array>; + /** The `inputSchema` the action declares on `GET /connectors`; absent ⇒ none declared. */ + inputSchema?: Record; } = {}): Harness { const writes: unknown[] = []; const handler = vi.fn(async () => (opts.result ?? { status: 200, ok: true, body: [] }) as Record); @@ -82,6 +84,10 @@ function harness(opts: { getConnectorOrigin: (name) => name === 'crm_api' ? ('origin' in opts ? opts.origin : 'declarative') : undefined, getConnectorDegradedReason: () => opts.degraded, + getConnectorDescriptors: () => [{ + name: 'crm_api', label: 'CRM', type: 'api', origin: 'declarative', state: 'ready', + actions: [{ key: 'request', label: 'HTTP Request', inputSchema: opts.inputSchema }], + }], }, providerOf: (name) => (name === 'crm_api' ? ('provider' in opts ? opts.provider : 'rest') : undefined), logger: silent, @@ -285,3 +291,104 @@ describe('[#20919] connector pull — one action call, the watermark read from t expect(res.watermark?.from).toBeUndefined(); }); }); + +/** + * [#28040] The pull door judges the input it is about to send against the + * action's declared `inputSchema` — the check the `connector_action` node + * runs (#27995), through the same helper — and refuses BEFORE dispatch. The + * pin is the upstream call count: a refused pull never calls the action. + */ +describe('[#28040] connector pull — the input is checked against the action\'s inputSchema before dispatch', () => { + /** An action whose `query.since` is required and whose `query.limit` is an integer. */ + const SINCE_REQUIRED = { + type: 'object', + properties: { + method: { type: 'string' }, + path: { type: 'string' }, + query: { + type: 'object', + properties: { since: { type: 'string' }, limit: { type: 'integer' } }, + required: ['since'], + }, + }, + required: ['path', 'query'], + }; + + it('an input missing a required parameter is VALIDATION_ERROR 400 naming it, and the action is never called', async () => { + const h = harness({ + inputSchema: SINCE_REQUIRED, + mapping: baseMapping({ + connectorSource: { connector: 'crm_api', action: 'request', input: { path: '/contacts', query: {} } }, + }), + }); + const err = await refusal(h); + expectRefused(err, { code: 'VALIDATION_ERROR', status: 400, reason: 'input_schema_mismatch' }); + expect(err.message).toContain("missing required 'query.since'"); + expect(h.handler).toHaveBeenCalledTimes(0); + expect(h.writes).toEqual([]); + }); + + it('a missing required section names the parameters it requires', async () => { + const h = harness({ inputSchema: SINCE_REQUIRED }); + const err = await refusal(h); + expectRefused(err, { code: 'VALIDATION_ERROR', status: 400, reason: 'input_schema_mismatch' }); + expect(err.message).toContain("missing required 'query' (with 'query.since')"); + expect(h.handler).toHaveBeenCalledTimes(0); + expect(h.writes).toEqual([]); + }); + + it('a parameter of the wrong type is refused the same way, and text is not converted to the integer', async () => { + const h = harness({ + inputSchema: SINCE_REQUIRED, + mapping: baseMapping({ + connectorSource: { + connector: 'crm_api', action: 'request', + input: { path: '/contacts', query: { since: '2026-01-01', limit: '50' } }, + }, + }), + }); + const err = await refusal(h); + expectRefused(err, { code: 'VALIDATION_ERROR', status: 400, reason: 'input_schema_mismatch' }); + expect(err.message).toContain("'query.limit' must be integer, got string"); + // The pull door runs no CEL, so its hint says what JSON to write, not which cast to use. + expect(err.message).toContain('as a JSON number or boolean'); + expect(err.message).not.toMatch(/CEL|int\(\)|double\(\)|bool\(\)/); + expect(h.handler).toHaveBeenCalledTimes(0); + expect(h.writes).toEqual([]); + }); + + it('the input judged is the one sent: a watermark that supplies the required parameter pulls', async () => { + const h = harness({ + inputSchema: SINCE_REQUIRED, + mapping: baseMapping({ + connectorSource: { + connector: 'crm_api', action: 'request', input: { path: '/contacts', query: { limit: 50 } }, + watermark: { field: 'updated_at', param: 'since' }, + }, + }), + stored: [{ synced_at: '2026-01-02T00:00:00.000Z' }], + }); + const res = await pullConnectorSource(h.deps, { mapping: 'crm_contacts' }); + expect(h.handler).toHaveBeenCalledTimes(1); + expect(h.handler.mock.calls[0][0]).toEqual({ + path: '/contacts', query: { limit: 50, since: '2026-01-02T00:00:00.000Z' }, + }); + expect(res.pulled).toBe(0); + }); + + it('a valid input still pulls and writes', async () => { + const h = harness({ + inputSchema: SINCE_REQUIRED, + mapping: baseMapping({ + connectorSource: { + connector: 'crm_api', action: 'request', input: { path: '/contacts', query: { since: '2026-01-01' } }, + }, + }), + result: { status: 200, ok: true, body: [{ id: 'c1', name: 'Ada', updated_at: '2026-01-02' }] }, + }); + const res = await pullConnectorSource(h.deps, { mapping: 'crm_contacts' }); + expect(h.handler).toHaveBeenCalledTimes(1); + expect(res.pulled).toBe(1); + expect(h.writes).toHaveLength(1); + }); +}); diff --git a/packages/services/service-automation/src/connector-pull.ts b/packages/services/service-automation/src/connector-pull.ts index 6208562d93c..89f14ba01ec 100644 --- a/packages/services/service-automation/src/connector-pull.ts +++ b/packages/services/service-automation/src/connector-pull.ts @@ -55,6 +55,11 @@ import { import type { AutomationContext, Logger } from '@objectstack/spec/contracts'; import type { ConnectorOrigin } from '@objectstack/spec/integration'; import type { ConnectorActionHandler } from './engine.js'; +import { + findConnectorActionInputSchema, + judgeConnectorInput, + type ConnectorInputSchemaSource, +} from './connector-input-schema.js'; /** The connector providers a v1 pull reads through (the ruling's "rest / openapi only"). */ export const CONNECTOR_PULL_PROVIDERS: readonly string[] = Object.freeze(['rest', 'openapi']); @@ -72,6 +77,7 @@ export type ConnectorPullRefusalReason = | 'unsupported_transform' | 'watermark_field_unmapped' | 'input_query_not_object' + | 'input_schema_mismatch' | 'mapping_target_refused' | 'upstream_not_ok' | 'records_not_array' @@ -108,8 +114,12 @@ export interface ConnectorPullProtocol extends ImportProtocolLike { getObjectSchema?(objectName: string, environmentId?: string): Promise; } -/** The connector registry a pull resolves through — the automation engine's. */ -export interface ConnectorPullRegistry { +/** + * The connector registry a pull resolves through — the automation engine's. + * Its descriptor list is where the action's declared `inputSchema` is read + * (#28040), the read the `connector_action` node makes. + */ +export interface ConnectorPullRegistry extends ConnectorInputSchemaSource { resolveConnectorAction(connectorId: string, actionId: string): ConnectorActionHandler | undefined; getConnectorOrigin(name: string): ConnectorOrigin | undefined; getConnectorDegradedReason(name: string): string | undefined; @@ -375,6 +385,27 @@ export async function pullConnectorSource( if (watermark && watermark.from !== undefined) { input.query = { ...(isPlainObject(input.query) ? input.query : {}), [watermark.param]: watermark.from }; } + // [#28040] The input about to be sent — the watermark included — is judged + // against the action's declared `inputSchema` by the check the + // `connector_action` node runs (#27995), so a pull missing a required + // parameter is refused here instead of reaching the remote API. The cast + // hint is the pull door's own: `connectorSource.input` is JSON spread as + // written and the watermark is the stored value, so no CEL envelope runs + // here and the node's `int()` / `double()` / `bool()` advice cannot apply. + const verdict = judgeConnectorInput(findConnectorActionInputSchema(registry, connector, action), input); + if (verdict) { + return refuse( + 'VALIDATION_ERROR', 400, 'input_schema_mismatch', + `Mapping "${mappingName}": input for '${connector}.${action}' does not match the action's inputSchema — ` + + `${verdict.violations}. Nothing was pulled. Fix connectorSource.input so it satisfies the inputSchema ` + + 'GET /api/v1/automation/connectors lists for this action.' + + (verdict.textForScalar + ? ' A text value for an integer, number or boolean parameter is not converted: write it in ' + + "connectorSource.input as a JSON number or boolean (50, not '50'); a watermark parameter " + + "carries the stored target field's value, so its declared type must be that field's." + : ''), + ); + } logger.info( `[Automation] connector pull: mapping "${mappingName}" ← ${connector}.${action}` + (watermark ? ` (watermark ${watermark.target} from ${watermark.from === undefined ? 'the start' : String(watermark.from)})` : ''), diff --git a/packages/spec/docs/SYNC_ARCHITECTURE.md b/packages/spec/docs/SYNC_ARCHITECTURE.md index d4df7399b77..dab2b3a83eb 100644 --- a/packages/spec/docs/SYNC_ARCHITECTURE.md +++ b/packages/spec/docs/SYNC_ARCHITECTURE.md @@ -219,6 +219,10 @@ Complete, production-grade integration with external systems. Includes authentic > same per-row verdicts. Every binding failure (a plugin-registered connector, an > `ok: false` answer, a non-array at `recordsPath`, an `update` / `upsert` with no > `upsertKey`, an unmapped `watermark.field`) is refused before anything is written. +> The input the call would send, the watermark parameter included, is judged against +> the read action's declared `inputSchema` first, as a `connector_action` flow node's +> is: a missing required parameter or a wrong type is refused with `VALIDATION_ERROR` +> naming it, and the action is not called. > > - **Watermark — read from the target.** Nothing stores it: the starting point sent as > `query[watermark.param]` is the highest value already stored in the target field a