fix(sdk): fail closed on hosted extension dispatch

docs/CLOUD.md

@@ -340,10 +340,17 @@ 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
-Cloud to cross-check, and a declared harness Cloud cannot run yet (`gemini`)
+`cli`, else `claude`), and `tools.mcp`. Handler bodies are not statically
+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,

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 { 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<Input = undefined>(
@@ -175,6 +186,7 @@ export async function executeAuthoredFlow<Input = undefined>(
     ...(options.onWait !== undefined ? { onWait: options.onWait } : {}),
   };
   const definition = getDefinition<Input>(handle);
+  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');
@@ -195,6 +207,18 @@ export async function executeAuthoredFlow<Input = undefined>(
   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) {

packages/sdk/src/flow-extension-loader.ts

@@ -99,6 +99,93 @@ function subscriptionOf(handler: TriggerHandler): { provider: string; event: str
   return action === undefined ? { provider, event: type } : { provider, event: type, action };
 }
 
+/**
+ * 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 eventType: string;
+  readonly deliveryId: string;
+};
+
+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<HostedExtensionDispatch>;
+  if ((dispatch as Partial<HostedExtensionDispatch>)[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, typeof HOSTED_EXTENSION_DISPATCH_AUTHORITY | 'provenance'>,
+): HostedExtensionDispatch {
+  const dispatch = Object.freeze({
+    [HOSTED_EXTENSION_DISPATCH_AUTHORITY]: true as const,
+    provenance: 'integration-watch' as const,
+    ...delivery,
+  });
+  hostedEventIdentity(dispatch);
+  return dispatch;
+}
+
+/**
+ * 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 extensionHandlerForHostedDispatch(
+  dispatch: unknown,
+  extensions: readonly Pick<LoadedFlowExtension, 'name' | 'handlers'>[],
+): { readonly extension: Pick<LoadedFlowExtension, 'name' | 'handlers'>; 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, 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.name).join(', ')}).`,
+    );
+  }
+  return matches[0];
+}
+
 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)`);

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;

packages/sdk/tests/authored-flow.test.ts

@@ -1,9 +1,15 @@
 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 type { FlowExtensionManifest } from '../src/flow-extension-manifest.js';
 import { JournalClient } from '../src/journal-client.js';
+import {
+  hostedExtensionDispatchFromVerifiedDelivery,
+  type LoadedFlowExtension,
+} from '../src/flow-extension-loader.js';
 import {
   kernelDialectError,
   sendOk,
@@ -562,6 +568,111 @@ describe('authored flow journal executor', () => {
     }
   });
 
+  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 = loadedExtension('babysitter', extensionHandle);
+    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 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 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'),
+      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<JournalClient> {
     const client = new JournalClient(path, { requestTimeoutMs: 2000 });
     await client.connect();
@@ -577,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<typeof flow>): 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' : '';

packages/sdk/tests/preflight.test.ts

@@ -3,8 +3,13 @@ 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 {
+  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';
 import { preflightHelpers } from '../src/preflight.js';
 import { describe, expect, it } from 'vitest';
 import {
@@ -693,6 +698,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 {
+        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');
+      } 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 {