From 6a494f9cb93f3abb4b750e88e64d21cec4ad867d Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Thu, 9 Jul 2026 22:29:41 -0700 Subject: [PATCH 1/2] Advertise NDJSON operation outputs as the arrays invocations return NDJSON endpoints (application/stream+json, x-ndjson, jsonl) declare the schema of one line, but the invoke path collects the stream and returns an array of parsed lines. Extraction stored the per-line schema unwrapped, so describe previews promised a single object that invocations never returned, pushing agents into runtime shape-probing (seen with Vercel getRuntimeLogs). Wrap the stored output schema in an array (with the truncation and raw-text caveats in its description) on both serve paths, share one NDJSON media-type set between extraction and invocation, and stale-mark affected connections via data migrations (sqlite ledger + cloud drizzle 0010) so existing catalogs rebuild with the corrected schemas. --- .changeset/ndjson-output-arrays.md | 5 + .../drizzle/0010_ndjson_output_arrays.sql | 23 + apps/cloud/drizzle/meta/0010_snapshot.json | 1290 +++++++++++++++++ apps/cloud/drizzle/meta/_journal.json | 7 + .../src/db/data-migrations.test.ts | 3 + .../host-cloudflare/src/db/data-migrations.ts | 4 + apps/host-selfhost/src/db/data-migrations.ts | 4 + apps/local/src/db/data-migrations.ts | 4 + e2e/scenarios/openapi-ndjson-output.test.ts | 250 ++++ packages/plugins/openapi/src/sdk/backing.ts | 3 +- .../plugins/openapi/src/sdk/extract.test.ts | 89 ++ packages/plugins/openapi/src/sdk/extract.ts | 20 +- packages/plugins/openapi/src/sdk/index.ts | 5 + packages/plugins/openapi/src/sdk/invoke.ts | 20 +- .../src/sdk/ndjson-output-migration.test.ts | 73 + .../src/sdk/ndjson-output-migration.ts | 90 ++ .../plugins/openapi/src/sdk/openapi-utils.ts | 33 + .../src/sdk/streaming-response.test.ts | 81 ++ 18 files changed, 1985 insertions(+), 19 deletions(-) create mode 100644 .changeset/ndjson-output-arrays.md create mode 100644 apps/cloud/drizzle/0010_ndjson_output_arrays.sql create mode 100644 apps/cloud/drizzle/meta/0010_snapshot.json create mode 100644 e2e/scenarios/openapi-ndjson-output.test.ts create mode 100644 packages/plugins/openapi/src/sdk/ndjson-output-migration.test.ts create mode 100644 packages/plugins/openapi/src/sdk/ndjson-output-migration.ts diff --git a/.changeset/ndjson-output-arrays.md b/.changeset/ndjson-output-arrays.md new file mode 100644 index 000000000..0b2d0c15e --- /dev/null +++ b/.changeset/ndjson-output-arrays.md @@ -0,0 +1,5 @@ +--- +"@executor-js/plugin-openapi": patch +--- + +Advertise NDJSON operation outputs as arrays. Endpoints declaring `application/stream+json`, `application/x-ndjson`, or `application/jsonl` responses (for example Vercel's runtime-logs) spec the schema of one line, but invocations return an array of parsed lines; describe previews now wrap the line schema in an array so generated code matches what actually comes back. Existing integrations with NDJSON operations are stale-marked once so their tool catalogs rebuild with the corrected schemas. diff --git a/apps/cloud/drizzle/0010_ndjson_output_arrays.sql b/apps/cloud/drizzle/0010_ndjson_output_arrays.sql new file mode 100644 index 000000000..e811f3b4f --- /dev/null +++ b/apps/cloud/drizzle/0010_ndjson_output_arrays.sql @@ -0,0 +1,23 @@ +-- Extraction used to persist an NDJSON operation's PER-LINE response schema +-- as the tool's output schema, while the invoke path returns an ARRAY of +-- parsed lines — so describe previews promised a single object invocations +-- never returned. The producer now wraps those schemas in an array. Persisted +-- `tool` rows can't be recognized as NDJSON-derived from their schema alone, +-- but the stored operation bindings kept the response content type, so the +-- affected connections are findable: stale-mark `tools_synced_at` and the +-- next read rebuilds their tool rows through the fixed producer. +-- +-- Effectively idempotent: re-running re-marks the same connections, and a +-- rebuild from unchanged bindings writes identical rows. +UPDATE "connection" c +SET "tools_synced_at" = NULL +FROM ( + SELECT DISTINCT ps."tenant", ps."data"::jsonb ->> 'integration' AS integration + FROM "plugin_storage" ps + WHERE ps."collection" = 'operation' + AND (ps."data"::text LIKE '%application/stream+json%' + OR ps."data"::text LIKE '%application/x-ndjson%' + OR ps."data"::text LIKE '%application/jsonl%') +) ndjson +WHERE c."tenant" = ndjson."tenant" + AND c."integration" = ndjson."integration"; diff --git a/apps/cloud/drizzle/meta/0010_snapshot.json b/apps/cloud/drizzle/meta/0010_snapshot.json new file mode 100644 index 000000000..5b5ced86d --- /dev/null +++ b/apps/cloud/drizzle/meta/0010_snapshot.json @@ -0,0 +1,1290 @@ +{ + "id": "c27a8958-ad73-4590-9b9b-d1b2584e0bc6", + "prevId": "b07ddbda-be11-4679-9f00-16d1b3b5e879", + "version": "7", + "dialect": "postgresql", + "tables": { + "public.accounts": { + "name": "accounts", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.memberships": { + "name": "memberships", + "schema": "", + "columns": { + "account_id": { + "name": "account_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "organization_id": { + "name": "organization_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "memberships_account_id_accounts_id_fk": { + "name": "memberships_account_id_accounts_id_fk", + "tableFrom": "memberships", + "tableTo": "accounts", + "columnsFrom": ["account_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "memberships_organization_id_organizations_id_fk": { + "name": "memberships_organization_id_organizations_id_fk", + "tableFrom": "memberships", + "tableTo": "organizations", + "columnsFrom": ["organization_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "memberships_account_id_organization_id_pk": { + "name": "memberships_account_id_organization_id_pk", + "columns": ["account_id", "organization_id"] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.organizations": { + "name": "organizations", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "slug": { + "name": "slug", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "organizations_slug_unique": { + "name": "organizations_slug_unique", + "columns": [ + { + "expression": "slug", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.blob": { + "name": "blob", + "schema": "", + "columns": { + "namespace": { + "name": "namespace", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "key": { + "name": "key", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "value": { + "name": "value", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "id": { + "name": "id", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "blob_id_uidx": { + "name": "blob_id_uidx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.connection": { + "name": "connection", + "schema": "", + "columns": { + "integration": { + "name": "integration", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "template": { + "name": "template", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "provider": { + "name": "provider", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "item_ids": { + "name": "item_ids", + "type": "json", + "primaryKey": false, + "notNull": true + }, + "identity_label": { + "name": "identity_label", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "description": { + "name": "description", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "last_health": { + "name": "last_health", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "tools_synced_at": { + "name": "tools_synced_at", + "type": "bigint", + "primaryKey": false, + "notNull": false + }, + "oauth_client": { + "name": "oauth_client", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "oauth_client_owner": { + "name": "oauth_client_owner", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "refresh_item_id": { + "name": "refresh_item_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "expires_at": { + "name": "expires_at", + "type": "bigint", + "primaryKey": false, + "notNull": false + }, + "oauth_scope": { + "name": "oauth_scope", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "oauth_token_url": { + "name": "oauth_token_url", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "provider_state": { + "name": "provider_state", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "owner": { + "name": "owner", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "subject": { + "name": "subject", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "connection_uidx": { + "name": "connection_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "owner", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "subject", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "integration", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "name", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.definition": { + "name": "definition", + "schema": "", + "columns": { + "integration": { + "name": "integration", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "connection": { + "name": "connection", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "plugin_id": { + "name": "plugin_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "schema": { + "name": "schema", + "type": "json", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "owner": { + "name": "owner", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "subject": { + "name": "subject", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "definition_uidx": { + "name": "definition_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "owner", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "subject", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "integration", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "connection", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "name", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.integration": { + "name": "integration", + "schema": "", + "columns": { + "slug": { + "name": "slug", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "plugin_id": { + "name": "plugin_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "description": { + "name": "description", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "config_revised_at": { + "name": "config_revised_at", + "type": "bigint", + "primaryKey": false, + "notNull": false + }, + "config": { + "name": "config", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "health_check": { + "name": "health_check", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "can_remove": { + "name": "can_remove", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": true + }, + "can_refresh": { + "name": "can_refresh", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "integration_uidx": { + "name": "integration_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "slug", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.oauth_client": { + "name": "oauth_client", + "schema": "", + "columns": { + "slug": { + "name": "slug", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "authorization_url": { + "name": "authorization_url", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "token_url": { + "name": "token_url", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "grant": { + "name": "grant", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "client_id": { + "name": "client_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "client_secret_item_id": { + "name": "client_secret_item_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "resource": { + "name": "resource", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "origin_kind": { + "name": "origin_kind", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "origin_integration": { + "name": "origin_integration", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "origin_issuer": { + "name": "origin_issuer", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "owner": { + "name": "owner", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "subject": { + "name": "subject", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "oauth_client_uidx": { + "name": "oauth_client_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "owner", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "subject", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "slug", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.oauth_session": { + "name": "oauth_session", + "schema": "", + "columns": { + "state": { + "name": "state", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "client_slug": { + "name": "client_slug", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "integration": { + "name": "integration", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "template": { + "name": "template", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "redirect_url": { + "name": "redirect_url", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "pkce_verifier": { + "name": "pkce_verifier", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "identity_label": { + "name": "identity_label", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "payload": { + "name": "payload", + "type": "json", + "primaryKey": false, + "notNull": true + }, + "expires_at": { + "name": "expires_at", + "type": "bigint", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "owner": { + "name": "owner", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "subject": { + "name": "subject", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "oauth_session_uidx": { + "name": "oauth_session_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "state", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.plugin_storage": { + "name": "plugin_storage", + "schema": "", + "columns": { + "plugin_id": { + "name": "plugin_id", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "collection": { + "name": "collection", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "key": { + "name": "key", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "data": { + "name": "data", + "type": "json", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "owner": { + "name": "owner", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "subject": { + "name": "subject", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "plugin_storage_uidx": { + "name": "plugin_storage_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "owner", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "subject", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "plugin_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "collection", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.private_executor_cloud_settings": { + "name": "private_executor_cloud_settings", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "version": { + "name": "version", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true, + "default": "'1.0.0'" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.tool": { + "name": "tool", + "schema": "", + "columns": { + "integration": { + "name": "integration", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "connection": { + "name": "connection", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "plugin_id": { + "name": "plugin_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "description": { + "name": "description", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "input_schema": { + "name": "input_schema", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "output_schema": { + "name": "output_schema", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "annotations": { + "name": "annotations", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "owner": { + "name": "owner", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "subject": { + "name": "subject", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "tool_uidx": { + "name": "tool_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "owner", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "subject", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "integration", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "connection", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "name", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.tool_policy": { + "name": "tool_policy", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "pattern": { + "name": "pattern", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "action": { + "name": "action", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "position": { + "name": "position", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "owner": { + "name": "owner", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "subject": { + "name": "subject", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "tool_policy_uidx": { + "name": "tool_policy_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "owner", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "subject", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + } + }, + "enums": {}, + "schemas": {}, + "sequences": {}, + "roles": {}, + "policies": {}, + "views": {}, + "_meta": { + "columns": {}, + "schemas": {}, + "tables": {} + } +} diff --git a/apps/cloud/drizzle/meta/_journal.json b/apps/cloud/drizzle/meta/_journal.json index f6a97b755..b37e01479 100644 --- a/apps/cloud/drizzle/meta/_journal.json +++ b/apps/cloud/drizzle/meta/_journal.json @@ -71,6 +71,13 @@ "when": 1783025157233, "tag": "0009_true_lake", "breakpoints": true + }, + { + "idx": 10, + "version": "7", + "when": 1783660669921, + "tag": "0010_ndjson_output_arrays", + "breakpoints": true } ] } diff --git a/apps/host-cloudflare/src/db/data-migrations.test.ts b/apps/host-cloudflare/src/db/data-migrations.test.ts index 615547db9..02dc9be27 100644 --- a/apps/host-cloudflare/src/db/data-migrations.test.ts +++ b/apps/host-cloudflare/src/db/data-migrations.test.ts @@ -240,6 +240,7 @@ describe("runCloudflareDataMigrations", () => { expect(yield* Effect.promise(() => runCloudflareDataMigrations(d1, bucket))).toEqual([ "2026-06-20-google-openapi-ownership", "2026-07-08-provider-service-split", + "2026-07-09-openapi-ndjson-output-arrays", ]); expect(yield* Effect.promise(() => runCloudflareDataMigrations(d1, bucket))).toEqual([]); @@ -283,6 +284,7 @@ describe("runCloudflareDataMigrations", () => { expect(yield* Effect.promise(() => runCloudflareDataMigrations(d1, bucket))).toEqual([ "2026-06-20-google-openapi-ownership", "2026-07-08-provider-service-split", + "2026-07-09-openapi-ndjson-output-arrays", ]); expect(yield* Effect.promise(() => runCloudflareDataMigrations(d1, bucket))).toEqual([]); @@ -333,6 +335,7 @@ describe("runCloudflareDataMigrations", () => { expect(yield* Effect.promise(() => runCloudflareDataMigrations(d1, bucket))).toEqual([ "2026-06-20-google-openapi-ownership", "2026-07-08-provider-service-split", + "2026-07-09-openapi-ndjson-output-arrays", ]); expect(yield* Effect.promise(() => runCloudflareDataMigrations(d1, bucket))).toEqual([]); diff --git a/apps/host-cloudflare/src/db/data-migrations.ts b/apps/host-cloudflare/src/db/data-migrations.ts index 5d449cadd..85f48b371 100644 --- a/apps/host-cloudflare/src/db/data-migrations.ts +++ b/apps/host-cloudflare/src/db/data-migrations.ts @@ -7,6 +7,7 @@ import { type SqliteDataMigration, type SqliteDataMigrationClient, } from "@executor-js/sdk"; +import { openApiNdjsonOutputDataMigration } from "@executor-js/plugin-openapi"; import { googleOpenApiOwnershipDataMigration } from "@executor-js/plugin-openapi/providers/google"; import { @@ -242,6 +243,9 @@ const cloudflareDataMigrations = (bucket: R2Bucket | undefined): readonly Sqlite }).pipe(Effect.asVoid); }), }, + // Stale-mark connections whose operations return NDJSON so their tool rows + // rebuild with array-wrapped output schemas (mirrors cloud's drizzle 0010). + openApiNdjsonOutputDataMigration, ]; export const runCloudflareDataMigrations = ( diff --git a/apps/host-selfhost/src/db/data-migrations.ts b/apps/host-selfhost/src/db/data-migrations.ts index e736dd995..295f77d5b 100644 --- a/apps/host-selfhost/src/db/data-migrations.ts +++ b/apps/host-selfhost/src/db/data-migrations.ts @@ -8,6 +8,7 @@ import { sqliteDataMigration, type SqliteDataMigration } from "@executor-js/sdk"; import { runSqliteAuthConfigMigration } from "@executor-js/sdk/http-auth"; import { + openApiNdjsonOutputDataMigration, openApiOutputSchemaDataMigration, openApiSpecBlobDataMigration, } from "@executor-js/plugin-openapi"; @@ -33,4 +34,7 @@ export const selfHostDataMigrations: readonly SqliteDataMigration[] = [ graphqlIntrospectionBlobDataMigration, googleOpenApiOwnershipDataMigration, providerServiceSplitDataMigration, + // Stale-mark connections whose operations return NDJSON so their tool rows + // rebuild with array-wrapped output schemas (mirrors cloud's drizzle 0010). + openApiNdjsonOutputDataMigration, ]; diff --git a/apps/local/src/db/data-migrations.ts b/apps/local/src/db/data-migrations.ts index fea0c4ef8..522a7321d 100644 --- a/apps/local/src/db/data-migrations.ts +++ b/apps/local/src/db/data-migrations.ts @@ -13,6 +13,7 @@ import { } from "@executor-js/sdk"; import { runSqliteAuthConfigMigration } from "@executor-js/sdk/http-auth"; import { + openApiNdjsonOutputDataMigration, openApiOutputSchemaDataMigration, openApiSpecBlobDataMigration, } from "@executor-js/plugin-openapi"; @@ -47,4 +48,7 @@ export const localDataMigrations: readonly SqliteDataMigration[] = [ // GC dead DCR oauth_client rows (old always-register duplicates) and backfill // the surviving DCR rows' origin_issuer from token_url (issue #1120, Part C). oauthClientGcSqliteMigration, + // Stale-mark connections whose operations return NDJSON so their tool rows + // rebuild with array-wrapped output schemas (mirrors cloud's drizzle 0010). + openApiNdjsonOutputDataMigration, ]; diff --git a/e2e/scenarios/openapi-ndjson-output.test.ts b/e2e/scenarios/openapi-ndjson-output.test.ts new file mode 100644 index 000000000..00e4d7e53 --- /dev/null +++ b/e2e/scenarios/openapi-ndjson-output.test.ts @@ -0,0 +1,250 @@ +// Cross-target: the advertised output type of an NDJSON streaming operation +// must match what invoking it actually returns. +// +// NDJSON endpoints (Vercel's getRuntimeLogs is the motivating real case) are +// spec'd with a PER-LINE schema under `application/stream+json`: the schema +// describes one log line, the body is many of them. Executor's invoke path +// already understands this — it collects the stream and returns an ARRAY of +// parsed rows. But extraction stores the per-line schema as the operation's +// outputSchema unchanged, so `tools.schema` / `tools.describe.tool()` advertise +// `data: { level; message; ... }` (a single object) while the runtime value is +// `Array<{ level; message; ... }>`. +// +// An agent that trusts the typedef writes broken code; an agent that has been +// burned falls back to Object.keys/JSON.stringify shape-probing, defeating the +// point of shipping type definitions at all. This scenario pins the contract: +// what describe advertises is what invoke returns. +import { randomBytes } from "node:crypto"; +import { createServer } from "node:http"; + +import { expect } from "@effect/vitest"; +import { Effect } from "effect"; +import { composePluginApi } from "@executor-js/api/server"; +import { openApiHttpPlugin } from "@executor-js/plugin-openapi/api"; +import { IntegrationSlug } from "@executor-js/sdk/shared"; + +import { scenario } from "../src/scenario"; +import { Api, Mcp, Target } from "../src/services"; +import type { McpSession } from "../src/surfaces/mcp"; + +const api = composePluginApi([openApiHttpPlugin()] as const); + +const unique = (prefix: string) => `${prefix}_${randomBytes(4).toString("hex")}`; + +// Three real-shaped runtime-log lines, one JSON document per line — exactly +// what Vercel's runtime-logs endpoint sends over application/stream+json. +const LOG_LINES = [ + { level: "info", message: "build started", rowId: "r1", timestampInMs: 1_700_000_000_000 }, + { level: "warning", message: "slow cold start", rowId: "r2", timestampInMs: 1_700_000_000_100 }, + { level: "error", message: "unhandled rejection", rowId: "r3", timestampInMs: 1_700_000_000_200 }, +] as const; + +/** A real 127.0.0.1 server answering GET /logs with an NDJSON body. */ +const serveNdjsonLogs = Effect.acquireRelease( + Effect.callback<{ readonly baseUrl: string; readonly close: () => void }>((resume) => { + const server = createServer((_request, response) => { + response.writeHead(200, { "content-type": "application/stream+json" }); + response.end(LOG_LINES.map((line) => JSON.stringify(line)).join("\n")); + }); + server.listen(0, "127.0.0.1", () => { + const address = server.address(); + const port = typeof address === "object" && address ? address.port : 0; + resume( + Effect.succeed({ + baseUrl: `http://127.0.0.1:${port}`, + close: () => { + server.close(); + server.closeAllConnections(); + }, + }), + ); + }); + }), + (server) => Effect.sync(server.close), +); + +// The response declares the schema of ONE line under application/stream+json — +// the same convention Vercel's spec uses for getRuntimeLogs. +const runtimeLogsSpec = (baseUrl: string): string => + JSON.stringify({ + openapi: "3.0.3", + info: { title: "Runtime Logs API", version: "1.0.0" }, + servers: [{ url: baseUrl }], + paths: { + "/logs": { + get: { + operationId: "getRuntimeLogs", + summary: "Stream runtime logs for a deployment", + responses: { + "200": { + description: "Log lines as newline-delimited JSON", + content: { + "application/stream+json": { + schema: { + type: "object", + properties: { + level: { type: "string", enum: ["info", "warning", "error"] }, + message: { type: "string" }, + rowId: { type: "string" }, + timestampInMs: { type: "number" }, + }, + required: ["level", "message", "rowId", "timestampInMs"], + }, + }, + }, + }, + }, + }, + }, + }, + }); + +// No securitySchemes in the spec ⇒ the integration is no-auth, so the agent +// can wire the connection itself over the core tool (template "none"). +const createNoAuthConnectionCode = (slug: string) => ` +const created = await tools.executor.coreTools.connections.create({ + owner: "org", + name: "main", + integration: ${JSON.stringify(slug)}, + template: "none", +}); +return created.ok ? { ok: true } : { ok: false, error: created.error }; +`; + +// Invoke through the sandbox exactly as an agent would, and report the shape +// of what came back next to the shape describe.tool advertised for it. +const invokeAndDescribeCode = (slug: string) => ` +const found = await tools.search({ namespace: ${JSON.stringify(slug)}, query: "logs", limit: 5 }); +const path = found.items[0]?.path; +if (!path) return { ok: false, error: "tool not found", items: found.items }; + +const described = await tools.describe.tool({ path }); +const outputTypeScript = described?.data?.outputTypeScript ?? described?.outputTypeScript ?? null; + +let t = tools; +for (const seg of path.split(".")) t = t[seg]; +const result = await t({}); +if (!result.ok) return { ok: false, error: result.error }; + +return { + ok: true, + path, + outputTypeScript, + runtimeIsArray: Array.isArray(result.data), + data: result.data, +}; +`; + +const removeConnectionsCode = (slug: string) => ` +const list = await tools.executor.coreTools.connections.list({}); +const mine = (list.ok ? list.data.connections : []).filter((c) => c.integration === ${JSON.stringify(slug)}); +for (const c of mine) { + await tools.executor.coreTools.connections.remove({ owner: c.owner, integration: c.integration, name: c.name }); +} +return { removed: mine.length }; +`; + +/** Run `execute`, auto-approving any policy-paused execution, and parse the + * sandbox's JSON return value. */ +const executeJson = (session: McpSession, code: string) => + Effect.gen(function* () { + let result = yield* session.call("execute", { code }); + let guard = 0; + while (result.text.includes("executionId:") && guard < 10) { + result = yield* session.approvePaused(result.text); + guard += 1; + } + expect(result.ok, `execute completed (got: ${result.text.slice(0, 400)})`).toBe(true); + return JSON.parse(result.text) as Record; + }); + +scenario( + "OpenAPI · an NDJSON operation's advertised output type matches what invoking it returns", + { timeout: 180_000 }, + Effect.scoped( + Effect.gen(function* () { + const target = yield* Target; + const mcp = yield* Mcp; + const { client } = yield* Api; + + const slug = unique("ndjson"); + const identity = yield* target.newIdentity(); + const session = mcp.session(identity); + const apiClient = yield* client(api, identity); + const upstream = yield* serveNdjsonLogs; + + yield* Effect.ensuring( + Effect.gen(function* () { + const added = yield* apiClient.openapi.addSpec({ + payload: { + spec: { kind: "blob", value: runtimeLogsSpec(upstream.baseUrl) }, + slug, + baseUrl: upstream.baseUrl, + }, + }); + expect(added.toolCount, "the NDJSON operation was extracted as a tool").toBe(1); + + const connected = yield* executeJson(session, createNoAuthConnectionCode(slug)); + expect( + connected.ok, + `the no-auth connection was created: ${JSON.stringify(connected)}`, + ).toBe(true); + + // What the agent sees before writing code: the schema view (the same + // compiled TypeScript previews tools.describe.tool() serves). + const tools = yield* apiClient.tools.list({ + query: { integration: IntegrationSlug.make(slug) }, + }); + expect(tools.length, "the tool is in the catalog").toBe(1); + const schema = yield* apiClient.tools.schema({ + query: { address: tools[0]!.address }, + }); + + // What the agent gets when it runs that code: the real invocation, + // through the sandbox, against the live NDJSON upstream. + const invoked = yield* executeJson(session, invokeAndDescribeCode(slug)); + expect(invoked.ok, `the tool answered over the wire: ${JSON.stringify(invoked)}`).toBe( + true, + ); + + // The invoke path parses NDJSON into an array of rows. This half of + // the contract already holds and documents the actual runtime shape. + expect(invoked.runtimeIsArray, "the runtime value is an array of log lines").toBe(true); + const rows = invoked.data as ReadonlyArray>; + expect(rows.length, "every NDJSON line became a row").toBe(LOG_LINES.length); + expect(rows[0], "rows are the parsed per-line objects").toMatchObject({ + level: "info", + message: "build started", + }); + + // THE CONTRACT UNDER TEST: the advertised type must describe that + // array. Today extraction stores the per-line schema unwrapped, so + // the typedef claims `data` is a single log-line object — the exact + // mismatch that pushes agents into JSON.stringify shape-probing. + const advertised = schema.outputTypeScript ?? ""; + expect( + /\[\]|Array (${rows.length} rows)`, + ).toBe(true); + + // And the sandbox's describe.tool must tell the same story — it is + // the surface agents actually consult before writing code. + const sandboxAdvertised = String(invoked.outputTypeScript ?? ""); + expect( + /\[\]|Array { : Option.match(binding.responseBody, { onNone: () => undefined, onSome: (responseBody) => - normalizeOpenApiRefs(Option.getOrUndefined(responseBody.schema)), + normalizeOpenApiRefs(outputSchemaFromResponseBody(responseBody)), }), annotations: annotationsForOperation(binding.method, binding.pathTemplate), }; diff --git a/packages/plugins/openapi/src/sdk/extract.test.ts b/packages/plugins/openapi/src/sdk/extract.test.ts index a10cd83bc..3bd12af11 100644 --- a/packages/plugins/openapi/src/sdk/extract.test.ts +++ b/packages/plugins/openapi/src/sdk/extract.test.ts @@ -87,4 +87,93 @@ describe("OpenAPI extract response bodies", () => { expect(responseBody?.contentType).toBe("application/json"); }), ); + + it.effect("wraps NDJSON response schemas in an array (the shape invoke returns)", () => + Effect.gen(function* () { + // NDJSON endpoints (e.g. Vercel's runtime-logs) declare the schema of + // ONE line under application/stream+json; the invoke path returns an + // array of parsed lines, so the advertised output schema must be that + // array — not the per-line object. + const lineSchema = { + type: "object", + properties: { level: { type: "string" }, message: { type: "string" } }, + }; + const doc = yield* parse( + JSON.stringify({ + openapi: "3.0.3", + info: { title: "Logs", version: "1.0.0" }, + servers: [{ url: "https://api.example.com" }], + paths: { + "/logs": { + get: { + operationId: "getRuntimeLogs", + responses: { + "200": { + description: "Log lines", + content: { "application/stream+json": { schema: lineSchema } }, + }, + }, + }, + }, + }, + }), + ); + + const result = yield* extract(doc); + const operation = result.operations.find((op) => op.operationId === "getRuntimeLogs"); + const outputSchema = Option.getOrUndefined(operation!.outputSchema) as Record< + string, + unknown + >; + expect(outputSchema.type).toBe("array"); + expect(outputSchema.items).toEqual(lineSchema); + // The caveats (truncation, raw-text fallback) ride the description into + // the compiled TypeScript preview. + expect(outputSchema.description).toContain("NDJSON"); + + // The binding keeps the RAW per-line schema: invoke decodes the body + // per line, and the serve path re-derives the array wrap from the + // content type. + const responseBody = Option.getOrUndefined(operation!.responseBody); + expect(responseBody?.contentType).toBe("application/stream+json"); + expect(Option.getOrUndefined(responseBody!.schema)).toEqual(lineSchema); + }), + ); + + it.effect("leaves plain JSON response schemas unwrapped", () => + Effect.gen(function* () { + const doc = yield* parse( + JSON.stringify({ + openapi: "3.0.3", + info: { title: "Plain", version: "1.0.0" }, + servers: [{ url: "https://api.example.com" }], + paths: { + "/thing": { + get: { + operationId: "getThing", + responses: { + "200": { + description: "One thing", + content: { + "application/json": { + schema: { type: "object", properties: { id: { type: "string" } } }, + }, + }, + }, + }, + }, + }, + }, + }), + ); + + const result = yield* extract(doc); + const operation = result.operations.find((op) => op.operationId === "getThing"); + const outputSchema = Option.getOrUndefined(operation!.outputSchema) as Record< + string, + unknown + >; + expect(outputSchema.type).toBe("object"); + }), + ); }); diff --git a/packages/plugins/openapi/src/sdk/extract.ts b/packages/plugins/openapi/src/sdk/extract.ts index ea3533342..6715ef3af 100644 --- a/packages/plugins/openapi/src/sdk/extract.ts +++ b/packages/plugins/openapi/src/sdk/extract.ts @@ -14,6 +14,8 @@ import { import { declaredContents, DocResolver, + isNdjsonMediaType, + ndjsonArrayOutputSchema, preferredResponseContent, type OperationObject, type ParameterObject, @@ -259,6 +261,22 @@ const extractResponseBody = ( return undefined; }; +/** + * Derive an operation's output schema from its response body. NDJSON bodies + * (`application/stream+json` and friends) are spec'd per LINE but returned by + * the invoke path as an array of parsed lines, so the advertised schema wraps + * the line schema in an array — otherwise describe previews promise a single + * object that invocations never return. Used by both the whole-tree extract + * and the serve path's stored-binding rebuild so the two stay in lockstep. + */ +export const outputSchemaFromResponseBody = ( + responseBody: OperationResponseBody, +): unknown | undefined => { + const schema = Option.getOrUndefined(responseBody.schema); + if (schema === undefined) return undefined; + return isNdjsonMediaType(responseBody.contentType) ? ndjsonArrayOutputSchema(schema) : schema; +}; + // --------------------------------------------------------------------------- // Input schema builder // --------------------------------------------------------------------------- @@ -508,7 +526,7 @@ export const extract = Effect.fn("OpenApi.extract")(function* (doc: ParsedDocume const responseBody = extractResponseBody(operation, r); const servers = operationServers(pathItem, operation, docServers); const inputSchema = buildInputSchema(parameters, requestBody, servers); - const outputSchema = responseBody ? Option.getOrUndefined(responseBody.schema) : undefined; + const outputSchema = responseBody ? outputSchemaFromResponseBody(responseBody) : undefined; const tags = (operation.tags ?? []).filter((t) => t.trim().length > 0); const operationPathTemplate = explicitPathTemplate(operation) ?? pathTemplate; diff --git a/packages/plugins/openapi/src/sdk/index.ts b/packages/plugins/openapi/src/sdk/index.ts index e3b412cb9..a4864fe6e 100644 --- a/packages/plugins/openapi/src/sdk/index.ts +++ b/packages/plugins/openapi/src/sdk/index.ts @@ -126,4 +126,9 @@ export { unwrapOpenApiTransportEnvelope, } from "./output-schema-migration"; +export { + openApiNdjsonOutputDataMigration, + runSqliteNdjsonOutputMigration, +} from "./ndjson-output-migration"; + export { openApiSpecBlobDataMigration } from "./spec-blob-migration"; diff --git a/packages/plugins/openapi/src/sdk/invoke.ts b/packages/plugins/openapi/src/sdk/invoke.ts index 5bbeb1517..62e483fc7 100644 --- a/packages/plugins/openapi/src/sdk/invoke.ts +++ b/packages/plugins/openapi/src/sdk/invoke.ts @@ -3,7 +3,7 @@ import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstab import type { ToolFileValue } from "@executor-js/sdk/core"; import { OpenApiInvocationError } from "./errors"; -import { resolveServerUrl } from "./openapi-utils"; +import { isNdjsonMediaType, NDJSON_MEDIA_TYPES, resolveServerUrl } from "./openapi-utils"; import { type EncodingObject, type OperationFileHint, @@ -178,12 +178,7 @@ const formatTimeout = (timeoutMs: number): string => const responseHeadersTimeoutMessage = (timeoutMs: number): string => `Upstream returned no response headers within ${formatTimeout(timeoutMs)}. The endpoint may be a live stream with no data to send (for example, runtime logs of an idle deployment); the request was aborted. Retry when the resource has activity, or use a non-streaming endpoint.`; -const STREAMING_RESPONSE_CONTENT_TYPES = new Set([ - "application/stream+json", - "application/x-ndjson", - "application/jsonl", - "text/event-stream", -]); +const STREAMING_RESPONSE_CONTENT_TYPES = new Set([...NDJSON_MEDIA_TYPES, "text/event-stream"]); const isStreamingResponseContentType = (ct: string | null | undefined): boolean => STREAMING_RESPONSE_CONTENT_TYPES.has(normalizeContentType(ct)); @@ -247,20 +242,11 @@ const parseStreamingJsonLines = (text: string, truncated: boolean): unknown => { return rows; }; -const NDJSON_CONTENT_TYPES = new Set([ - "application/stream+json", - "application/x-ndjson", - "application/jsonl", -]); - const decodeStreamingResponseBody = ( contentType: string | null, text: string, truncated: boolean, -): unknown => - NDJSON_CONTENT_TYPES.has(normalizeContentType(contentType)) - ? parseStreamingJsonLines(text, truncated) - : text; +): unknown => (isNdjsonMediaType(contentType) ? parseStreamingJsonLines(text, truncated) : text); export const collectStreamingBody = ( stream: Stream.Stream, diff --git a/packages/plugins/openapi/src/sdk/ndjson-output-migration.test.ts b/packages/plugins/openapi/src/sdk/ndjson-output-migration.test.ts new file mode 100644 index 000000000..fd199225c --- /dev/null +++ b/packages/plugins/openapi/src/sdk/ndjson-output-migration.test.ts @@ -0,0 +1,73 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Effect } from "effect"; +import type { SqliteDataMigrationClient } from "@executor-js/sdk/core"; + +import { runSqliteNdjsonOutputMigration } from "./ndjson-output-migration"; + +// A tiny scripted fake standing in for a libSQL client. +const makeFakeClient = ( + operationRows: Record[], + options?: { readonly missingTable?: string }, +) => { + const log: (string | { readonly sql: string; readonly args: readonly unknown[] })[] = []; + const client: SqliteDataMigrationClient = { + execute: (stmt) => { + log.push(stmt); + const sql = typeof stmt === "string" ? stmt : stmt.sql; + if (sql.includes("sqlite_master")) { + const table = typeof stmt === "string" ? "" : String(stmt.args[0]); + return Promise.resolve({ + rows: table === options?.missingTable ? [] : [{ name: table }], + }); + } + if (sql.includes("SELECT DISTINCT")) { + return Promise.resolve({ rows: operationRows }); + } + return Promise.resolve({ rows: [] }); + }, + }; + return { client, log }; +}; + +describe("runSqliteNdjsonOutputMigration", () => { + it.effect("stale-marks each affected integration's connections", () => + Effect.gen(function* () { + const { client, log } = makeFakeClient([ + { tenant: "t1", integration: "vercel_api" }, + { tenant: "t2", integration: "logs_api" }, + ]); + const marked = yield* runSqliteNdjsonOutputMigration(client); + expect(marked).toBe(2); + + const updates = log.filter( + (stmt) => typeof stmt !== "string" && stmt.sql.includes("tools_synced_at = NULL"), + ); + expect(updates.map((stmt) => (typeof stmt === "string" ? [] : stmt.args))).toEqual([ + ["t1", "vercel_api"], + ["t2", "logs_api"], + ]); + }), + ); + + it.effect("skips malformed rows and marks nothing when no operation is NDJSON", () => + Effect.gen(function* () { + const { client, log } = makeFakeClient([{ tenant: null, integration: "broken" }]); + const marked = yield* runSqliteNdjsonOutputMigration(client); + expect(marked).toBe(0); + expect( + log.some((stmt) => typeof stmt !== "string" && stmt.sql.includes("tools_synced_at")), + ).toBe(false); + }), + ); + + it.effect("treats a fresh database (missing tables) as nothing to migrate", () => + Effect.gen(function* () { + for (const missingTable of ["connection", "plugin_storage"]) { + const { client } = makeFakeClient([{ tenant: "t1", integration: "vercel_api" }], { + missingTable, + }); + expect(yield* runSqliteNdjsonOutputMigration(client)).toBe(0); + } + }), + ); +}); diff --git a/packages/plugins/openapi/src/sdk/ndjson-output-migration.ts b/packages/plugins/openapi/src/sdk/ndjson-output-migration.ts new file mode 100644 index 000000000..409d7f944 --- /dev/null +++ b/packages/plugins/openapi/src/sdk/ndjson-output-migration.ts @@ -0,0 +1,90 @@ +// --------------------------------------------------------------------------- +// Data migration: rebuild tool catalogs whose operations return NDJSON. +// +// Extraction used to persist an NDJSON operation's PER-LINE response schema as +// the tool's output schema, while the invoke path returns an ARRAY of parsed +// lines — so describe previews promised a single object that invocations never +// returned. The producer now wraps those schemas in an array; the per-line +// shape can't be recognized from a persisted `tool` row alone (it's just a +// JSON schema), but the stored operation bindings kept the response content +// type, so the affected connections are findable and the catalog machinery +// already knows how to rebuild them: stale-mark `tools_synced_at` and the next +// read re-produces the rows through the fixed producer. Mirrors the cloud +// drizzle migration (apps/cloud/drizzle/0010_ndjson_output_arrays.sql). +// +// Effectively idempotent: re-running re-marks the same connections, and a +// rebuild from unchanged bindings writes identical rows. +// --------------------------------------------------------------------------- + +import { Effect } from "effect"; +import { DataMigrationError, type SqliteDataMigrationClient } from "@executor-js/sdk/core"; + +const MIGRATION_NAME = "2026-07-09-openapi-ndjson-output-arrays"; + +// Substring probes for the NDJSON media types (NDJSON_MEDIA_TYPES in +// openapi-utils.ts). The binding JSON inside `plugin_storage.data` may be a +// nested object or a re-encoded string, so a LIKE over the raw row text is the +// shape-agnostic match; false positives only cause a harmless extra rebuild. +const NDJSON_TYPE_PROBES = [ + "application/stream+json", + "application/x-ndjson", + "application/jsonl", +] as const; + +const execute = ( + client: SqliteDataMigrationClient, + stmt: string | { readonly sql: string; readonly args: readonly unknown[] }, +) => + Effect.tryPromise({ + try: () => client.execute(stmt), + catch: (cause) => new DataMigrationError({ migration: MIGRATION_NAME, cause }), + }); + +const tableExists = (client: SqliteDataMigrationClient, table: string) => + Effect.map( + execute(client, { + sql: "SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?", + args: [table], + }), + (result) => result.rows.length > 0, + ); + +/** Stale-mark every connection whose integration has at least one stored + * operation binding with an NDJSON response, so the next read rebuilds its + * tool rows (with array-wrapped output schemas). Returns the number of + * connections marked. Fresh databases may lack either table — nothing to + * migrate. */ +export const runSqliteNdjsonOutputMigration = ( + client: SqliteDataMigrationClient, +): Effect.Effect => + Effect.gen(function* () { + for (const table of ["connection", "plugin_storage"]) { + if (!(yield* tableExists(client, table))) return 0; + } + + const probes = NDJSON_TYPE_PROBES.map(() => "ps.data LIKE ?").join(" OR "); + const affected = yield* execute(client, { + sql: `SELECT DISTINCT ps.tenant AS tenant, json_extract(ps.data, '$.integration') AS integration + FROM plugin_storage ps + WHERE ps.collection = 'operation' AND (${probes})`, + args: NDJSON_TYPE_PROBES.map((probe) => `%${probe}%`), + }); + + let marked = 0; + for (const row of affected.rows) { + if (typeof row.tenant !== "string" || typeof row.integration !== "string") continue; + yield* execute(client, { + sql: "UPDATE connection SET tools_synced_at = NULL WHERE tenant = ? AND integration = ?", + args: [row.tenant, row.integration], + }); + marked += 1; + } + return marked; + }); + +/** Registry entry for the boot-time data-migration ledger. */ +export const openApiNdjsonOutputDataMigration = { + name: MIGRATION_NAME, + run: (client: SqliteDataMigrationClient) => + runSqliteNdjsonOutputMigration(client).pipe(Effect.asVoid), +}; diff --git a/packages/plugins/openapi/src/sdk/openapi-utils.ts b/packages/plugins/openapi/src/sdk/openapi-utils.ts index 39ed375e6..2e85d489d 100644 --- a/packages/plugins/openapi/src/sdk/openapi-utils.ts +++ b/packages/plugins/openapi/src/sdk/openapi-utils.ts @@ -128,3 +128,36 @@ export const preferredResponseContent = ( entries[0]; return pick ? { mediaType: pick[0], media: pick[1] } : undefined; }; + +// --------------------------------------------------------------------------- +// NDJSON responses +// --------------------------------------------------------------------------- + +export const normalizeMediaType = (mediaType: string | null | undefined): string => + mediaType?.split(";")[0]?.trim().toLowerCase() ?? ""; + +/** Media types whose bodies are newline-delimited JSON documents. The invoke + * path collects these streams and returns an ARRAY of parsed lines, so every + * schema surface must describe that array — the spec convention is to declare + * the schema of ONE line (e.g. Vercel's runtime-logs endpoint). */ +export const NDJSON_MEDIA_TYPES: ReadonlySet = new Set([ + "application/stream+json", + "application/x-ndjson", + "application/jsonl", +]); + +export const isNdjsonMediaType = (mediaType: string | null | undefined): boolean => + NDJSON_MEDIA_TYPES.has(normalizeMediaType(mediaType)); + +/** Wrap a per-line NDJSON response schema into the array the runtime actually + * returns. The description rides into the compiled TypeScript preview as a + * JSDoc comment, so agents learn the truncation and raw-text caveats where + * they read the type. */ +export const ndjsonArrayOutputSchema = (lineSchema: unknown): Record => ({ + type: "array", + items: lineSchema, + description: + "Parsed NDJSON stream: one array item per line. The stream may be truncated " + + "(`x-executor-stream: truncated` response header); a body that is not valid " + + "NDJSON is returned as the raw string instead.", +}); diff --git a/packages/plugins/openapi/src/sdk/streaming-response.test.ts b/packages/plugins/openapi/src/sdk/streaming-response.test.ts index 8c7f9a894..82e1ec958 100644 --- a/packages/plugins/openapi/src/sdk/streaming-response.test.ts +++ b/packages/plugins/openapi/src/sdk/streaming-response.test.ts @@ -273,4 +273,85 @@ describe("OpenAPI streaming responses", () => { expect(text).not.toMatch(/x-executor-stream/); }), ); + + it.effect("advertises an NDJSON operation's output as the array invoke returns", () => + Effect.gen(function* () { + // The spec declares the schema of ONE line (Vercel's runtime-logs + // convention); the runtime returns an array of parsed lines. The + // advertised output schema must describe that array on BOTH serve + // paths — the add-time compile and the stored-binding rebuild — or + // agents write code against a shape that never comes back. + const lineSchema = { + type: "object", + properties: { level: { type: "string" }, message: { type: "string" } }, + }; + const server = yield* startRawServer((res) => { + res.writeHead(200, { "content-type": "application/stream+json" }); + res.end( + `${JSON.stringify({ level: "info", message: "a" })}\n` + + `${JSON.stringify({ level: "error", message: "b" })}\n`, + ); + }); + + const executor = yield* createExecutor(makeTestConfig({ plugins: testPlugins() })); + yield* executor.openapi.addSpec({ + spec: { + kind: "blob", + value: JSON.stringify({ + openapi: "3.0.3", + info: { title: "Runtime Logs", version: "1.0.0" }, + servers: [{ url: server.baseUrl }], + paths: { + "/logs": { + get: { + operationId: "getLogs", + tags: ["logs"], + responses: { + "200": { + description: "Log lines", + content: { "application/stream+json": { schema: lineSchema } }, + }, + }, + }, + }, + }, + }), + }, + slug: "ndjson_logs", + baseUrl: server.baseUrl, + }); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: IntegrationSlug.make("ndjson_logs"), + template: AuthTemplateSlug.make("apiKey"), + value: "token", + }); + const address = ToolAddress.make("tools.ndjson_logs.org.main.logs.getLogs"); + + const schema = yield* executor.tools.schema(address); + const outputSchema = schema?.outputSchema as Record; + expect(outputSchema.type).toBe("array"); + expect(outputSchema.items).toEqual(lineSchema); + + // Force a catalog rebuild (the stored-binding serve path) to re-derive + // the same array wrap from the persisted content type. + yield* executor.connections.refresh({ + owner: "org", + integration: IntegrationSlug.make("ndjson_logs"), + name: ConnectionName.make("main"), + }); + const rebuilt = yield* executor.tools.schema(address); + const rebuiltOutput = rebuilt?.outputSchema as Record; + expect(rebuiltOutput.type).toBe("array"); + expect(rebuiltOutput.items).toEqual(lineSchema); + + // And the runtime value is exactly that array. + const result = unwrapInvocation(yield* executor.execute(address, {})); + expect(result.data).toEqual([ + { level: "info", message: "a" }, + { level: "error", message: "b" }, + ]); + }), + ); }); From 69499196d7e9928626bd40e646ffaf57acbec5c8 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Thu, 9 Jul 2026 22:39:08 -0700 Subject: [PATCH 2/2] Drop em-dashes from new comments --- apps/cloud/drizzle/0010_ndjson_output_arrays.sql | 2 +- e2e/scenarios/openapi-ndjson-output.test.ts | 12 ++++++------ packages/plugins/openapi/src/sdk/extract.test.ts | 2 +- packages/plugins/openapi/src/sdk/extract.ts | 2 +- .../openapi/src/sdk/ndjson-output-migration.ts | 4 ++-- packages/plugins/openapi/src/sdk/openapi-utils.ts | 2 +- .../openapi/src/sdk/streaming-response.test.ts | 2 +- 7 files changed, 13 insertions(+), 13 deletions(-) diff --git a/apps/cloud/drizzle/0010_ndjson_output_arrays.sql b/apps/cloud/drizzle/0010_ndjson_output_arrays.sql index e811f3b4f..804bc5882 100644 --- a/apps/cloud/drizzle/0010_ndjson_output_arrays.sql +++ b/apps/cloud/drizzle/0010_ndjson_output_arrays.sql @@ -1,6 +1,6 @@ -- Extraction used to persist an NDJSON operation's PER-LINE response schema -- as the tool's output schema, while the invoke path returns an ARRAY of --- parsed lines — so describe previews promised a single object invocations +-- parsed lines, so describe previews promised a single object invocations -- never returned. The producer now wraps those schemas in an array. Persisted -- `tool` rows can't be recognized as NDJSON-derived from their schema alone, -- but the stored operation bindings kept the response content type, so the diff --git a/e2e/scenarios/openapi-ndjson-output.test.ts b/e2e/scenarios/openapi-ndjson-output.test.ts index 00e4d7e53..564ecc723 100644 --- a/e2e/scenarios/openapi-ndjson-output.test.ts +++ b/e2e/scenarios/openapi-ndjson-output.test.ts @@ -4,7 +4,7 @@ // NDJSON endpoints (Vercel's getRuntimeLogs is the motivating real case) are // spec'd with a PER-LINE schema under `application/stream+json`: the schema // describes one log line, the body is many of them. Executor's invoke path -// already understands this — it collects the stream and returns an ARRAY of +// already understands this: it collects the stream and returns an ARRAY of // parsed rows. But extraction stores the per-line schema as the operation's // outputSchema unchanged, so `tools.schema` / `tools.describe.tool()` advertise // `data: { level; message; ... }` (a single object) while the runtime value is @@ -31,7 +31,7 @@ const api = composePluginApi([openApiHttpPlugin()] as const); const unique = (prefix: string) => `${prefix}_${randomBytes(4).toString("hex")}`; -// Three real-shaped runtime-log lines, one JSON document per line — exactly +// Three real-shaped runtime-log lines, one JSON document per line, exactly // what Vercel's runtime-logs endpoint sends over application/stream+json. const LOG_LINES = [ { level: "info", message: "build started", rowId: "r1", timestampInMs: 1_700_000_000_000 }, @@ -63,7 +63,7 @@ const serveNdjsonLogs = Effect.acquireRelease( (server) => Effect.sync(server.close), ); -// The response declares the schema of ONE line under application/stream+json — +// The response declares the schema of ONE line under application/stream+json, // the same convention Vercel's spec uses for getRuntimeLogs. const runtimeLogsSpec = (baseUrl: string): string => JSON.stringify({ @@ -219,7 +219,7 @@ scenario( // THE CONTRACT UNDER TEST: the advertised type must describe that // array. Today extraction stores the per-line schema unwrapped, so - // the typedef claims `data` is a single log-line object — the exact + // the typedef claims `data` is a single log-line object, the exact // mismatch that pushes agents into JSON.stringify shape-probing. const advertised = schema.outputTypeScript ?? ""; expect( @@ -229,7 +229,7 @@ scenario( ` runtime: Array<{ level; message; rowId; timestampInMs }> (${rows.length} rows)`, ).toBe(true); - // And the sandbox's describe.tool must tell the same story — it is + // And the sandbox's describe.tool must tell the same story: it is // the surface agents actually consult before writing code. const sandboxAdvertised = String(invoked.outputTypeScript ?? ""); expect( @@ -238,7 +238,7 @@ scenario( ` advertised: ${sandboxAdvertised}`, ).toBe(true); }), - // Selfhost shares one workspace identity — leaked connections fail + // Selfhost shares one workspace identity; leaked connections fail // other scenarios' zero-state assertions. Effect.gen(function* () { yield* executeJson(session, removeConnectionsCode(slug)).pipe(Effect.ignore); diff --git a/packages/plugins/openapi/src/sdk/extract.test.ts b/packages/plugins/openapi/src/sdk/extract.test.ts index 3bd12af11..7b521d464 100644 --- a/packages/plugins/openapi/src/sdk/extract.test.ts +++ b/packages/plugins/openapi/src/sdk/extract.test.ts @@ -93,7 +93,7 @@ describe("OpenAPI extract response bodies", () => { // NDJSON endpoints (e.g. Vercel's runtime-logs) declare the schema of // ONE line under application/stream+json; the invoke path returns an // array of parsed lines, so the advertised output schema must be that - // array — not the per-line object. + // array, not the per-line object. const lineSchema = { type: "object", properties: { level: { type: "string" }, message: { type: "string" } }, diff --git a/packages/plugins/openapi/src/sdk/extract.ts b/packages/plugins/openapi/src/sdk/extract.ts index 6715ef3af..5dafb4e4d 100644 --- a/packages/plugins/openapi/src/sdk/extract.ts +++ b/packages/plugins/openapi/src/sdk/extract.ts @@ -265,7 +265,7 @@ const extractResponseBody = ( * Derive an operation's output schema from its response body. NDJSON bodies * (`application/stream+json` and friends) are spec'd per LINE but returned by * the invoke path as an array of parsed lines, so the advertised schema wraps - * the line schema in an array — otherwise describe previews promise a single + * the line schema in an array; otherwise describe previews promise a single * object that invocations never return. Used by both the whole-tree extract * and the serve path's stored-binding rebuild so the two stay in lockstep. */ diff --git a/packages/plugins/openapi/src/sdk/ndjson-output-migration.ts b/packages/plugins/openapi/src/sdk/ndjson-output-migration.ts index 409d7f944..0e20dec14 100644 --- a/packages/plugins/openapi/src/sdk/ndjson-output-migration.ts +++ b/packages/plugins/openapi/src/sdk/ndjson-output-migration.ts @@ -3,7 +3,7 @@ // // Extraction used to persist an NDJSON operation's PER-LINE response schema as // the tool's output schema, while the invoke path returns an ARRAY of parsed -// lines — so describe previews promised a single object that invocations never +// lines, so describe previews promised a single object that invocations never // returned. The producer now wraps those schemas in an array; the per-line // shape can't be recognized from a persisted `tool` row alone (it's just a // JSON schema), but the stored operation bindings kept the response content @@ -52,7 +52,7 @@ const tableExists = (client: SqliteDataMigrationClient, table: string) => /** Stale-mark every connection whose integration has at least one stored * operation binding with an NDJSON response, so the next read rebuilds its * tool rows (with array-wrapped output schemas). Returns the number of - * connections marked. Fresh databases may lack either table — nothing to + * connections marked. Fresh databases may lack either table; nothing to * migrate. */ export const runSqliteNdjsonOutputMigration = ( client: SqliteDataMigrationClient, diff --git a/packages/plugins/openapi/src/sdk/openapi-utils.ts b/packages/plugins/openapi/src/sdk/openapi-utils.ts index 2e85d489d..6723de421 100644 --- a/packages/plugins/openapi/src/sdk/openapi-utils.ts +++ b/packages/plugins/openapi/src/sdk/openapi-utils.ts @@ -138,7 +138,7 @@ export const normalizeMediaType = (mediaType: string | null | undefined): string /** Media types whose bodies are newline-delimited JSON documents. The invoke * path collects these streams and returns an ARRAY of parsed lines, so every - * schema surface must describe that array — the spec convention is to declare + * schema surface must describe that array: the spec convention is to declare * the schema of ONE line (e.g. Vercel's runtime-logs endpoint). */ export const NDJSON_MEDIA_TYPES: ReadonlySet = new Set([ "application/stream+json", diff --git a/packages/plugins/openapi/src/sdk/streaming-response.test.ts b/packages/plugins/openapi/src/sdk/streaming-response.test.ts index 82e1ec958..54a2c5667 100644 --- a/packages/plugins/openapi/src/sdk/streaming-response.test.ts +++ b/packages/plugins/openapi/src/sdk/streaming-response.test.ts @@ -279,7 +279,7 @@ describe("OpenAPI streaming responses", () => { // The spec declares the schema of ONE line (Vercel's runtime-logs // convention); the runtime returns an array of parsed lines. The // advertised output schema must describe that array on BOTH serve - // paths — the add-time compile and the stored-binding rebuild — or + // paths (the add-time compile and the stored-binding rebuild) or // agents write code against a shape that never comes back. const lineSchema = { type: "object",