From 67180114a2791c1e02d3624f43cc82adaf471f31 Mon Sep 17 00:00:00 2001 From: Miya Date: Tue, 22 Sep 2026 08:39:02 +0200 Subject: [PATCH 1/3] feat(sdk): dispatch hosted extension handlers Session-Id: 01a0c4a6-dd65-7ce1-a90e-de1b0b4e86c3 Session-Id: 01a0c4a6-dd65-7ce1-a90e-de1b0b4e86c3 --- docs/CLOUD.md | 10 +++-- packages/sdk/src/authored-flow-executor.ts | 5 ++- packages/sdk/src/flow-extension-loader.ts | 48 ++++++++++++++++++++ packages/sdk/src/plugin-manifest.ts | 1 + packages/sdk/tests/authored-flow.test.ts | 51 +++++++++++++++++++++- 5 files changed, 109 insertions(+), 6 deletions(-) diff --git a/docs/CLOUD.md b/docs/CLOUD.md index 1fc6bc839..d3056358e 100644 --- a/docs/CLOUD.md +++ b/docs/CLOUD.md @@ -340,9 +340,13 @@ SDK; the same function reads a compiled YAML spec, where a helper step such as (`.on(github.issues())`), the `--on` sources and the deploy target (every launched run lands in `--repo`, so GitHub is always required), the `cli:` of each `f.agent`/`f.llm` call in the default body (else the nearest `flows.json` -`cli`, else `claude`), and `tools.mcp`. Handler bodies are not scanned: hosted -dispatch runs the default body (flows #301), so only a handler's trigger is a -requirement. The deploy body carries the same list as `requirements` for +`cli`, else `claude`), and `tools.mcp`. Handler bodies are not statically +scanned for requirements. Hosted dispatch runs the one schema-2 extension +handler matching Cloud's normalized `{ event: { provider, eventType } }` +envelope; ticket deliveries, unmatched events, and direct runs keep the base +flow's default body. Ambiguous extension matches fail closed. A handler's +trigger is therefore a requirement, while its body must stay within the +extension manifest's declared permissions. The deploy body carries the same list as `requirements` for Cloud to cross-check, and a declared harness Cloud cannot run yet (`gemini`) refuses the deploy unless `--agents` overrides it. diff --git a/packages/sdk/src/authored-flow-executor.ts b/packages/sdk/src/authored-flow-executor.ts index bbd66d05e..f9d53249a 100644 --- a/packages/sdk/src/authored-flow-executor.ts +++ b/packages/sdk/src/authored-flow-executor.ts @@ -45,7 +45,7 @@ import { import { AuthoredFlowLifecycle } from './authored-flow-lifecycle.js'; import { JournalClient } from './journal-client.js'; import { createHookEvaluator } from './authored-hooks.js'; -import { probeFlowExtension, type LoadedFlowExtension } from './flow-extension-loader.js'; +import { extensionHandlerForHostedInput, probeFlowExtension, type LoadedFlowExtension } from './flow-extension-loader.js'; import type { CompletionReason as ProtocolCompletionReason, RunCompletionReason as ProtocolRunCompletionReason, @@ -175,6 +175,7 @@ export async function executeAuthoredFlow( ...(options.onWait !== undefined ? { onWait: options.onWait } : {}), }; const definition = getDefinition(handle); + const hostedHandler = extensionHandlerForHostedInput(input, options.extensions ?? []); const headerFields = Object.keys(definition.header).filter(key => key !== 'tools' && key !== 'budget' && key !== 'memory' && key !== 'version' && key !== 'hooks'); if (definition.header.tools && Object.keys(definition.header.tools).some(key => !['mcp', ...helperProviders.map(p => p.namespace)].includes(key))) headerFields.push('tools'); if (definition.header.tools?.relayfile !== undefined) headerFields.push('tools.relayfile'); @@ -568,7 +569,7 @@ export async function executeAuthoredFlow( let bodyFailed = false; let bodyFailure: unknown; try { - const bodyPromise = lifecycle.runBody(() => definition.body(context, input as Input)); + const bodyPromise = lifecycle.runBody(() => (hostedHandler?.body ?? definition.body)(context, input as Input)); await bodyPromise; } catch (error) { bodyFailed = true; diff --git a/packages/sdk/src/flow-extension-loader.ts b/packages/sdk/src/flow-extension-loader.ts index 14be52141..814c9a7b1 100644 --- a/packages/sdk/src/flow-extension-loader.ts +++ b/packages/sdk/src/flow-extension-loader.ts @@ -99,6 +99,54 @@ function subscriptionOf(handler: TriggerHandler): { provider: string; event: str return action === undefined ? { provider, event: type } : { provider, event: type, action }; } +type HostedEventIdentity = { + readonly provider: string; + readonly event: string; + readonly action?: string; +}; + +function hostedEventIdentity(input: unknown): HostedEventIdentity | undefined { + if (typeof input !== 'object' || input === null || Array.isArray(input)) return undefined; + const event = (input as { event?: unknown }).event; + if (typeof event !== 'object' || event === null || Array.isArray(event)) return undefined; + const { provider, eventType } = event as { provider?: unknown; eventType?: unknown }; + if (typeof provider !== 'string' || provider.length === 0 || typeof eventType !== 'string' || eventType.length === 0) return undefined; + const separator = eventType.indexOf('.'); + if (separator === -1) return { provider, event: eventType }; + const name = eventType.slice(0, separator); + const action = eventType.slice(separator + 1); + if (name.length === 0 || action.length === 0) return undefined; + return { provider, event: name, action }; +} + +/** + * Select the one extension handler authorized by Cloud's normalized event + * envelope. The base body remains the fallback for ticket deliveries and + * direct runs. Overlapping extension subscriptions fail closed: silently + * choosing lock order would suppress an installed handler while claiming the + * composition ran. + */ +export function extensionHandlerForHostedInput( + input: unknown, + extensions: readonly Pick[], +): TriggerHandler | undefined { + const identity = hostedEventIdentity(input); + if (identity === undefined) return undefined; + const matches = extensions.flatMap(extension => extension.handlers.flatMap(handler => { + const subscription = subscriptionOf(handler); + if (subscription === undefined || subscription.provider !== identity.provider || subscription.event !== identity.event) return []; + if (subscription.action !== undefined && subscription.action !== identity.action) return []; + return [{ extension: extension.name, handler }]; + })); + if (matches.length > 1) { + throw new PluginError( + 'plugin_event_ambiguous', + `Hosted event ${identity.provider}.${identity.event}${identity.action === undefined ? '' : `.${identity.action}`} matches multiple extension handlers (${matches.map(match => match.extension).join(', ')}).`, + ); + } + return matches[0]?.handler; +} + function assertDeclaredSubscription(name: string, manifest: FlowExtensionManifest, handler: TriggerHandler, index: number): void { const subscription = subscriptionOf(handler); if (handler.trigger.kind === 'schedule') unsupported(name, `handler ${index} (a schedule trigger)`); diff --git a/packages/sdk/src/plugin-manifest.ts b/packages/sdk/src/plugin-manifest.ts index fb4791d79..54dcd5be1 100644 --- a/packages/sdk/src/plugin-manifest.ts +++ b/packages/sdk/src/plugin-manifest.ts @@ -7,6 +7,7 @@ export const PLUGIN_FAILURE_KINDS = [ 'plugin_unsupported', 'plugin_credential_missing', 'plugin_server_unreachable', // schema 2 flow extensions (see flow-extension-manifest.ts, plugin-source.ts, plugin-github.ts) 'plugin_kind_invalid', 'plugin_incompatible', 'plugin_event_unroutable', 'plugin_source_invalid', + 'plugin_event_ambiguous', 'plugin_source_unresolved', 'plugin_fetch_failed', 'plugin_path_invalid', 'plugin_too_large', 'plugin_source_drift', 'plugin_lock_invalid', ] as const; diff --git a/packages/sdk/tests/authored-flow.test.ts b/packages/sdk/tests/authored-flow.test.ts index 32da08917..d43c6c299 100644 --- a/packages/sdk/tests/authored-flow.test.ts +++ b/packages/sdk/tests/authored-flow.test.ts @@ -1,9 +1,11 @@ import { rmSync } from 'node:fs'; import type { Server } from 'node:net'; -import { flow, type Ctx, type FlowHeader } from '@relayflows/surface'; +import { flow, github, type Ctx, type FlowHeader } from '@relayflows/surface'; +import { getFlowDefinition } from '@relayflows/surface/runtime'; import { afterAll, beforeAll, describe, expect, it } from 'vitest'; import { executeAuthoredFlow } from '../src/authored-flow-executor.js'; import { JournalClient } from '../src/journal-client.js'; +import type { LoadedFlowExtension } from '../src/flow-extension-loader.js'; import { kernelDialectError, sendOk, @@ -562,6 +564,53 @@ describe('authored flow journal executor', () => { } }); + it('runs the matching extension handler for a normalized hosted event instead of the base body', async () => { + const base = flow<{ event: { provider: string; eventType: string } }>('software-factory', async (f) => { + await f.run('printf base-body'); + f.done('success'); + }); + const extensionHandle = flow('babysitter', async (f) => f.done('declined')) + .on(github.pull_request('labeled'), async (f) => { + await f.run('printf extension-body'); + f.done('success'); + }); + const extension = { + name: 'babysitter', + handlers: getFlowDefinition(extensionHandle).handlers, + hooks: Object.freeze({}), + manifest: undefined, + } as unknown as LoadedFlowExtension; + const client = await connectedClient('authored-extension-handler-test'); + const before = startedSpecs.length; + try { + await executeAuthoredFlow(base, client, { + event: { provider: 'github', eventType: 'pull_request.labeled' }, + }, { extensions: [extension] }); + } finally { + client.close(); + } + expect(commandsSince(before)).toEqual(['printf extension-body', ':']); + }); + + it('fails closed when a hosted event matches more than one extension handler', async () => { + const handler = getFlowDefinition( + flow('one', async (f) => f.done('declined')) + .on(github.pull_request('labeled'), async (f) => f.done('success')), + ).handlers[0]!; + const extension = (name: string) => ({ + name, + handlers: [handler], + hooks: Object.freeze({}), + manifest: undefined, + }) as unknown as LoadedFlowExtension; + await expect(executeAuthoredFlow( + flow('software-factory', async (f) => f.done('success')), + new JournalClient('/unused'), + { event: { provider: 'github', eventType: 'pull_request.labeled' } }, + { extensions: [extension('one'), extension('two')] }, + )).rejects.toMatchObject({ code: 'plugin_event_ambiguous' }); + }); + async function connectedClient(name: string): Promise { const client = new JournalClient(path, { requestTimeoutMs: 2000 }); await client.connect(); From 989f748369d2d21553d3615ddaf254199236d507 Mon Sep 17 00:00:00 2001 From: Miya Date: Tue, 22 Sep 2026 09:00:14 +0200 Subject: [PATCH 2/3] test(sdk): cover ambiguous extension refusal Session-Id: 01a0c4a6-dd65-7ce1-a90e-de1b0b4e86c3 --- packages/sdk/tests/preflight.test.ts | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/packages/sdk/tests/preflight.test.ts b/packages/sdk/tests/preflight.test.ts index d7726699e..ddfab4c18 100644 --- a/packages/sdk/tests/preflight.test.ts +++ b/packages/sdk/tests/preflight.test.ts @@ -3,8 +3,10 @@ import { join } from 'node:path'; import { tmpdir } from 'node:os'; import { addPlugin } from '../src/cli/add.js'; import { runPluginCommand } from '../src/cli/plugin.js'; +import { extensionHandlerForHostedInput } from '../src/flow-extension-loader.js'; import { SHA_A, fakeGithub, type FakeEntry } from './fake-github.js'; import type { PreflightFailureKind } from '../src/failure-kinds.js'; +import { PluginError } from '../src/plugin-manifest.js'; import { preflightHelpers } from '../src/preflight.js'; import { describe, expect, it } from 'vitest'; import { @@ -693,6 +695,26 @@ describe('preflight: CLI resolution and refusal predicates', () => { expect(await addPlugin(ref, io, { cwd: root, extension: { fetch, versions: { sdk: '2.0.22', surface: '2.0.22' } } })).toBe(2); } finally { rmSync(root, { recursive: true, force: true }); } } + // `plugin_event_ambiguous`: two installed extensions claim the same + // normalized hosted event. The dispatcher must refuse instead of + // silently selecting lock order and suppressing one handler. + const handler = { + trigger: { + kind: 'webhook', name: 'github', + filter: { provider: 'github', type: 'pull_request', payload: { action: 'labeled' } }, + }, + body: async () => {}, + }; + try { + extensionHandlerForHostedInput( + { event: { provider: 'github', eventType: 'pull_request.labeled' } }, + [{ name: 'one', handlers: [handler] }, { name: 'two', handlers: [handler] }] as never, + ); + throw new Error('expected ambiguous extension dispatch to refuse'); + } catch (error) { + expect(error).toBeInstanceOf(PluginError); + refusalKinds.push((error as PluginError).code); + } // `plugin_lock_invalid`: a declaration with no lockfile entry behind it. const root = mkdtempSync(join(tmpdir(), 'plugin-lock-taxonomy-')); try { From 482d2562df158b8f49e6b924c94f1f8e1c9f5c16 Mon Sep 17 00:00:00 2001 From: Miya Date: Tue, 22 Sep 2026 09:35:18 +0200 Subject: [PATCH 3/3] fix(sdk): fail closed on hosted extension dispatch Session-Id: 01a0c4a6-dd65-7ce1-a90e-de1b0b4e86c3 --- docs/CLOUD.md | 17 ++- packages/sdk/src/authored-flow-executor.ts | 29 ++++- packages/sdk/src/flow-extension-loader.ts | 95 ++++++++++---- packages/sdk/tests/authored-flow.test.ts | 143 +++++++++++++++++---- packages/sdk/tests/preflight.test.ts | 9 +- 5 files changed, 230 insertions(+), 63 deletions(-) diff --git a/docs/CLOUD.md b/docs/CLOUD.md index d3056358e..5a74ef5b4 100644 --- a/docs/CLOUD.md +++ b/docs/CLOUD.md @@ -341,13 +341,16 @@ SDK; the same function reads a compiled YAML spec, where a helper step such as launched run lands in `--repo`, so GitHub is always required), the `cli:` of each `f.agent`/`f.llm` call in the default body (else the nearest `flows.json` `cli`, else `claude`), and `tools.mcp`. Handler bodies are not statically -scanned for requirements. Hosted dispatch runs the one schema-2 extension -handler matching Cloud's normalized `{ event: { provider, eventType } }` -envelope; ticket deliveries, unmatched events, and direct runs keep the base -flow's default body. Ambiguous extension matches fail closed. A handler's -trigger is therefore a requirement, while its body must stay within the -extension manifest's declared permissions. The deploy body carries the same list as `requirements` for -Cloud to cross-check, and a declared harness Cloud cannot run yet (`gemini`) +scanned for requirements. Authored input never selects an extension handler: +only server-authenticated integration delivery metadata passed out of band may +match one. Direct runs and authenticated deliveries that match no extension +handler keep the base flow's default body; overlapping or malformed matches +fail closed. This release also refuses a matching handler with +`plugin_unsupported` before either body starts because schema-2 entries are +ordinary JavaScript and their manifest permissions are not yet isolated by the +runtime (gate 8 / #442). A handler's trigger is therefore a requirement, but +the handler cannot execute until that boundary exists. The deploy body carries +the same list as `requirements` for Cloud to cross-check, and a declared harness Cloud cannot run yet (`gemini`) refuses the deploy unless `--agents` overrides it. Before `flows deploy`, `flows schedule` and `flows run --cloud` submit anything, diff --git a/packages/sdk/src/authored-flow-executor.ts b/packages/sdk/src/authored-flow-executor.ts index f9d53249a..25b979b6b 100644 --- a/packages/sdk/src/authored-flow-executor.ts +++ b/packages/sdk/src/authored-flow-executor.ts @@ -44,8 +44,14 @@ import { } from './authored-flow-operation.js'; import { AuthoredFlowLifecycle } from './authored-flow-lifecycle.js'; import { JournalClient } from './journal-client.js'; +import { PluginError } from './plugin-manifest.js'; import { createHookEvaluator } from './authored-hooks.js'; -import { extensionHandlerForHostedInput, probeFlowExtension, type LoadedFlowExtension } from './flow-extension-loader.js'; +import { + extensionHandlerForHostedDispatch, + probeFlowExtension, + type HostedExtensionDispatch, + type LoadedFlowExtension, +} from './flow-extension-loader.js'; import type { CompletionReason as ProtocolCompletionReason, RunCompletionReason as ProtocolRunCompletionReason, @@ -152,6 +158,11 @@ export interface ExecuteAuthoredFlowOptions { readonly rootRunId?: string; /** Installed flow-extension plugins, in lock order, so `f.hook` can AND-compose them. */ readonly extensions?: readonly LoadedFlowExtension[]; + /** + * Server-authenticated integration delivery authority. Never derive this + * from authored input: direct `/workflows/run` callers control that JSON. + */ + readonly extensionDispatch?: HostedExtensionDispatch; } export async function executeAuthoredFlow( @@ -175,7 +186,7 @@ export async function executeAuthoredFlow( ...(options.onWait !== undefined ? { onWait: options.onWait } : {}), }; const definition = getDefinition(handle); - const hostedHandler = extensionHandlerForHostedInput(input, options.extensions ?? []); + const hostedHandler = extensionHandlerForHostedDispatch(options.extensionDispatch, options.extensions ?? []); const headerFields = Object.keys(definition.header).filter(key => key !== 'tools' && key !== 'budget' && key !== 'memory' && key !== 'version' && key !== 'hooks'); if (definition.header.tools && Object.keys(definition.header.tools).some(key => !['mcp', ...helperProviders.map(p => p.namespace)].includes(key))) headerFields.push('tools'); if (definition.header.tools?.relayfile !== undefined) headerFields.push('tools.relayfile'); @@ -196,6 +207,18 @@ export async function executeAuthoredFlow( for (const extension of options.extensions ?? []) { if (extension.manifest !== undefined) await probeFlowExtension(extension.manifest); } + if (hostedHandler !== undefined) { + // A schema-2 entry is ordinary authored JavaScript. Passing it the base + // context would not constrain direct Node access, f.run, helpers, MCP or + // agent harnesses to manifest.permissions; those declarations are still + // explicitly unenforced by gate 8 / #442. Refuse before either body runs. + // A later isolated runtime may replace this gate only when it can prove + // the manifest boundary, not merely proxy selected Ctx properties. + throw new PluginError( + 'plugin_unsupported', + `${hostedHandler.extension.name}: hosted extension handlers require enforced manifest permission isolation (gate 8 / #442).`, + ); + } const budget = new AuthoredBudget(definition.header.budget); if (definition.header.memory?.agent === true) { @@ -569,7 +592,7 @@ export async function executeAuthoredFlow( let bodyFailed = false; let bodyFailure: unknown; try { - const bodyPromise = lifecycle.runBody(() => (hostedHandler?.body ?? definition.body)(context, input as Input)); + const bodyPromise = lifecycle.runBody(() => definition.body(context, input as Input)); await bodyPromise; } catch (error) { bodyFailed = true; diff --git a/packages/sdk/src/flow-extension-loader.ts b/packages/sdk/src/flow-extension-loader.ts index 814c9a7b1..be1d90110 100644 --- a/packages/sdk/src/flow-extension-loader.ts +++ b/packages/sdk/src/flow-extension-loader.ts @@ -99,52 +99,91 @@ function subscriptionOf(handler: TriggerHandler): { provider: string; event: str return action === undefined ? { provider, event: type } : { provider, event: type, action }; } -type HostedEventIdentity = { +/** + * Server-authenticated integration delivery metadata. This is executor + * authority, not authored input: a caller must derive it from its verified + * delivery record and pass it out of band. User-controlled `input.event` + * objects never become this value. + */ +const HOSTED_EXTENSION_DISPATCH_AUTHORITY = Symbol('hosted-extension-dispatch-authority'); + +export type HostedExtensionDispatch = { + readonly [HOSTED_EXTENSION_DISPATCH_AUTHORITY]: true; + readonly provenance: 'integration-watch'; readonly provider: string; - readonly event: string; - readonly action?: string; + readonly eventType: string; + readonly deliveryId: string; }; -function hostedEventIdentity(input: unknown): HostedEventIdentity | undefined { - if (typeof input !== 'object' || input === null || Array.isArray(input)) return undefined; - const event = (input as { event?: unknown }).event; - if (typeof event !== 'object' || event === null || Array.isArray(event)) return undefined; - const { provider, eventType } = event as { provider?: unknown; eventType?: unknown }; - if (typeof provider !== 'string' || provider.length === 0 || typeof eventType !== 'string' || eventType.length === 0) return undefined; - const separator = eventType.indexOf('.'); - if (separator === -1) return { provider, event: eventType }; - const name = eventType.slice(0, separator); - const action = eventType.slice(separator + 1); - if (name.length === 0 || action.length === 0) return undefined; - return { provider, event: name, action }; +type HostedEventIdentity = { readonly provider: string; readonly event: string; readonly action?: string }; +const DISPATCH_PROVIDER = /^[a-z0-9][a-z0-9-]{0,63}$/; +const DISPATCH_EVENT = /^[A-Za-z_][A-Za-z0-9_]*$/; +const DISPATCH_DELIVERY = /^[A-Za-z0-9_.:-]{1,200}$/; + +function hostedEventIdentity(dispatch: unknown): HostedEventIdentity { + if (typeof dispatch !== 'object' || dispatch === null || Array.isArray(dispatch)) { + throw new PluginError('plugin_event_unroutable', 'Hosted extension dispatch authority is malformed.'); + } + const { provenance, provider, eventType, deliveryId } = dispatch as Partial; + if ((dispatch as Partial)[HOSTED_EXTENSION_DISPATCH_AUTHORITY] !== true + || provenance !== 'integration-watch' + || typeof provider !== 'string' || !DISPATCH_PROVIDER.test(provider) + || typeof deliveryId !== 'string' || !DISPATCH_DELIVERY.test(deliveryId) + || typeof eventType !== 'string') { + throw new PluginError('plugin_event_unroutable', 'Hosted extension dispatch authority is malformed.'); + } + const parts = eventType.split('.'); + if ((parts.length !== 1 && parts.length !== 2) || parts.some(part => !DISPATCH_EVENT.test(part))) { + throw new PluginError('plugin_event_unroutable', `Hosted extension event ${JSON.stringify(eventType)} is malformed.`); + } + return parts.length === 1 + ? { provider, event: parts[0]! } + : { provider, event: parts[0]!, action: parts[1]! }; +} + +/** + * Brand metadata only after the host has authenticated the integration + * delivery. The symbol is deliberately not serializable, so copying a direct + * run's JSON into executor options cannot mint dispatch authority. + */ +export function hostedExtensionDispatchFromVerifiedDelivery( + delivery: Omit, +): HostedExtensionDispatch { + const dispatch = Object.freeze({ + [HOSTED_EXTENSION_DISPATCH_AUTHORITY]: true as const, + provenance: 'integration-watch' as const, + ...delivery, + }); + hostedEventIdentity(dispatch); + return dispatch; } /** - * Select the one extension handler authorized by Cloud's normalized event - * envelope. The base body remains the fallback for ticket deliveries and - * direct runs. Overlapping extension subscriptions fail closed: silently - * choosing lock order would suppress an installed handler while claiming the - * composition ran. + * Resolve a server-authenticated delivery to one extension handler. Authored + * input is deliberately absent from this API: direct runs may contain any + * JSON shape and cannot opt themselves into extension execution. Overlapping + * subscriptions fail closed, including a generic event handler overlapping + * an action-specific handler. */ -export function extensionHandlerForHostedInput( - input: unknown, +export function extensionHandlerForHostedDispatch( + dispatch: unknown, extensions: readonly Pick[], -): TriggerHandler | undefined { - const identity = hostedEventIdentity(input); - if (identity === undefined) return undefined; +): { readonly extension: Pick; readonly handler: TriggerHandler } | undefined { + if (dispatch === undefined) return undefined; + const identity = hostedEventIdentity(dispatch); const matches = extensions.flatMap(extension => extension.handlers.flatMap(handler => { const subscription = subscriptionOf(handler); if (subscription === undefined || subscription.provider !== identity.provider || subscription.event !== identity.event) return []; if (subscription.action !== undefined && subscription.action !== identity.action) return []; - return [{ extension: extension.name, handler }]; + return [{ extension, handler }]; })); if (matches.length > 1) { throw new PluginError( 'plugin_event_ambiguous', - `Hosted event ${identity.provider}.${identity.event}${identity.action === undefined ? '' : `.${identity.action}`} matches multiple extension handlers (${matches.map(match => match.extension).join(', ')}).`, + `Hosted event ${identity.provider}.${identity.event}${identity.action === undefined ? '' : `.${identity.action}`} matches multiple extension handlers (${matches.map(match => match.extension.name).join(', ')}).`, ); } - return matches[0]?.handler; + return matches[0]; } function assertDeclaredSubscription(name: string, manifest: FlowExtensionManifest, handler: TriggerHandler, index: number): void { diff --git a/packages/sdk/tests/authored-flow.test.ts b/packages/sdk/tests/authored-flow.test.ts index d43c6c299..7f1baca1a 100644 --- a/packages/sdk/tests/authored-flow.test.ts +++ b/packages/sdk/tests/authored-flow.test.ts @@ -4,8 +4,12 @@ import { flow, github, type Ctx, type FlowHeader } from '@relayflows/surface'; import { getFlowDefinition } from '@relayflows/surface/runtime'; import { afterAll, beforeAll, describe, expect, it } from 'vitest'; import { executeAuthoredFlow } from '../src/authored-flow-executor.js'; +import type { FlowExtensionManifest } from '../src/flow-extension-manifest.js'; import { JournalClient } from '../src/journal-client.js'; -import type { LoadedFlowExtension } from '../src/flow-extension-loader.js'; +import { + hostedExtensionDispatchFromVerifiedDelivery, + type LoadedFlowExtension, +} from '../src/flow-extension-loader.js'; import { kernelDialectError, sendOk, @@ -564,22 +568,19 @@ describe('authored flow journal executor', () => { } }); - it('runs the matching extension handler for a normalized hosted event instead of the base body', async () => { + it('does not let direct-run input impersonate an integration delivery', async () => { const base = flow<{ event: { provider: string; eventType: string } }>('software-factory', async (f) => { await f.run('printf base-body'); f.done('success'); }); + let extensionRuns = 0; const extensionHandle = flow('babysitter', async (f) => f.done('declined')) .on(github.pull_request('labeled'), async (f) => { + extensionRuns += 1; await f.run('printf extension-body'); f.done('success'); }); - const extension = { - name: 'babysitter', - handlers: getFlowDefinition(extensionHandle).handlers, - hooks: Object.freeze({}), - manifest: undefined, - } as unknown as LoadedFlowExtension; + const extension = loadedExtension('babysitter', extensionHandle); const client = await connectedClient('authored-extension-handler-test'); const before = startedSpecs.length; try { @@ -589,28 +590,89 @@ describe('authored flow journal executor', () => { } finally { client.close(); } - expect(commandsSince(before)).toEqual(['printf extension-body', ':']); + expect(commandsSince(before)).toEqual(['printf base-body', ':']); + expect(extensionRuns).toBe(0); + }); + + it('refuses an authenticated matching handler before base or extension authority can run', async () => { + let baseRuns = 0; + let extensionRuns = 0; + const base = flow('software-factory', async (f) => { + baseRuns += 1; + await f.run('printf base-body'); + f.done('success'); + }); + const extension = loadedExtension('babysitter', flow('babysitter', async (f) => f.done('declined')) + .on(github.pull_request('labeled'), async (f) => { + extensionRuns += 1; + await f.run('printf extension-body'); + f.done('success'); + })); + + await expect(executeAuthoredFlow(base, new JournalClient('/unused'), undefined, { + extensions: [extension], + extensionDispatch: integrationDispatch('pull_request.labeled'), + })).rejects.toMatchObject({ code: 'plugin_unsupported' }); + expect(baseRuns).toBe(0); + expect(extensionRuns).toBe(0); }); - it('fails closed when a hosted event matches more than one extension handler', async () => { - const handler = getFlowDefinition( - flow('one', async (f) => f.done('declined')) - .on(github.pull_request('labeled'), async (f) => f.done('success')), - ).handlers[0]!; - const extension = (name: string) => ({ - name, - handlers: [handler], - hooks: Object.freeze({}), - manifest: undefined, - }) as unknown as LoadedFlowExtension; + it('fails closed when one event matches identical extension subscriptions', async () => { + const extension = (name: string) => loadedExtension(name, + flow(name, async (f) => f.done('declined')) + .on(github.pull_request('labeled'), async (f) => f.done('success'))); await expect(executeAuthoredFlow( flow('software-factory', async (f) => f.done('success')), new JournalClient('/unused'), - { event: { provider: 'github', eventType: 'pull_request.labeled' } }, - { extensions: [extension('one'), extension('two')] }, + undefined, + { + extensions: [extension('one'), extension('two')], + extensionDispatch: integrationDispatch('pull_request.labeled'), + }, )).rejects.toMatchObject({ code: 'plugin_event_ambiguous' }); }); + it('fails closed when a generic subscription overlaps an action subscription', async () => { + const generic = loadedExtension('generic', flow('generic', async (f) => f.done('declined')) + .on(github.pull_request(), async (f) => f.done('success'))); + const action = loadedExtension('action', flow('action', async (f) => f.done('declined')) + .on(github.pull_request('labeled'), async (f) => f.done('success'))); + + await expect(executeAuthoredFlow( + flow('software-factory', async (f) => f.done('success')), + new JournalClient('/unused'), + undefined, + { + extensions: [generic, action], + extensionDispatch: integrationDispatch('pull_request.labeled'), + }, + )).rejects.toMatchObject({ code: 'plugin_event_ambiguous' }); + }); + + it.each([ + ['an extra event segment', { provenance: 'integration-watch', provider: 'github', eventType: 'pull_request.labeled.extra', deliveryId: 'delivery-1' }], + ['an invalid provider', { provenance: 'integration-watch', provider: 'GitHub', eventType: 'pull_request.labeled', deliveryId: 'delivery-1' }], + ['a missing delivery id', { provenance: 'integration-watch', provider: 'github', eventType: 'pull_request.labeled' }], + ['a caller-asserted provenance kind', { provenance: 'direct-run', provider: 'github', eventType: 'pull_request.labeled', deliveryId: 'delivery-1' }], + ])('refuses malformed hosted dispatch authority: %s', async (_case, extensionDispatch) => { + await expect(executeAuthoredFlow( + flow('software-factory', async (f) => f.done('success')), + new JournalClient('/unused'), + undefined, + { extensionDispatch: extensionDispatch as never }, + )).rejects.toMatchObject({ code: 'plugin_event_unroutable' }); + }); + + it('refuses a serialized copy of verified dispatch metadata', async () => { + const serialized = JSON.parse(JSON.stringify(integrationDispatch('pull_request.labeled'))); + await expect(executeAuthoredFlow( + flow('software-factory', async (f) => f.done('success')), + new JournalClient('/unused'), + undefined, + { extensionDispatch: serialized as never }, + )).rejects.toMatchObject({ code: 'plugin_event_unroutable' }); + }); + async function connectedClient(name: string): Promise { const client = new JournalClient(path, { requestTimeoutMs: 2000 }); await client.connect(); @@ -626,6 +688,43 @@ describe('authored flow journal executor', () => { } }); +function integrationDispatch(eventType: string) { + return hostedExtensionDispatchFromVerifiedDelivery({ provider: 'github', eventType, deliveryId: 'delivery-1' }); +} + +function loadedExtension(name: string, handle: ReturnType): LoadedFlowExtension { + const manifest: FlowExtensionManifest = { + schema: 2, + kind: 'flow-extension', + name, + version: '1.0.0', + compat: { surface: '*', sdk: '*', base: [{ name: 'software-factory', version: '*' }] }, + entry: `${name}.flow.ts`, + extends: { handlers: true, hooks: [] }, + triggers: [{ provider: 'github', event: 'pull_request', actions: ['labeled'] }], + permissions: { + integrations: ['github'], + harnesses: [], + mcp: [], + writes: ['github:pull_request'], + }, + preflight: { credentials: [], servers: [] }, + }; + return { + name, + version: '1.0.0', + ref: `github:AgentWorkforce/flows@${'a'.repeat(40)}#${name}`, + digest: 'b'.repeat(64), + directory: `/extensions/${name}`, + entryPath: `/extensions/${name}/${name}.flow.ts`, + manifest, + handle, + getDefinition: getFlowDefinition, + handlers: getFlowDefinition(handle).handlers, + hooks: Object.freeze({}), + }; +} + function outputFor(command: string): string { if (command.startsWith('emit:')) return command.slice('emit:'.length); return command === 'printf authored-journal-ok' ? 'authored-journal-ok' : ''; diff --git a/packages/sdk/tests/preflight.test.ts b/packages/sdk/tests/preflight.test.ts index ddfab4c18..4dc843515 100644 --- a/packages/sdk/tests/preflight.test.ts +++ b/packages/sdk/tests/preflight.test.ts @@ -3,7 +3,10 @@ import { join } from 'node:path'; import { tmpdir } from 'node:os'; import { addPlugin } from '../src/cli/add.js'; import { runPluginCommand } from '../src/cli/plugin.js'; -import { extensionHandlerForHostedInput } from '../src/flow-extension-loader.js'; +import { + extensionHandlerForHostedDispatch, + hostedExtensionDispatchFromVerifiedDelivery, +} from '../src/flow-extension-loader.js'; import { SHA_A, fakeGithub, type FakeEntry } from './fake-github.js'; import type { PreflightFailureKind } from '../src/failure-kinds.js'; import { PluginError } from '../src/plugin-manifest.js'; @@ -706,8 +709,8 @@ describe('preflight: CLI resolution and refusal predicates', () => { body: async () => {}, }; try { - extensionHandlerForHostedInput( - { event: { provider: 'github', eventType: 'pull_request.labeled' } }, + extensionHandlerForHostedDispatch( + hostedExtensionDispatchFromVerifiedDelivery({ provider: 'github', eventType: 'pull_request.labeled', deliveryId: 'delivery-1' }), [{ name: 'one', handlers: [handler] }, { name: 'two', handlers: [handler] }] as never, ); throw new Error('expected ambiguous extension dispatch to refuse');