fix(service-automation)!: a connector pull whose input breaks the action's declared inputSchema is refused before dispatch

.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
+
+<!-- adr-0087: not-required (no-migration-prescription) No key of any schema is added, removed, renamed or re-shaped, and no export, type or error code is removed or renamed: `MappingSchema.connectorSource` and `ConnectorActionSchema.inputSchema` keep their shapes, and the refusal reuses the existing `VALIDATION_ERROR` code the pull door already answers its other input faults with. What narrows is the run-time dispatch of the connector pull: a call the action's own declared schema says is wrong is no longer sent upstream. Nothing stored is rewritten or re-judged at rest, so there is nothing for `objectstack migrate meta` to rewrite and no tombstone. -->
+
+**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.

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

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<string, unknown> {
-    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<string, unknown>;
-        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));
-    }
-}