diff --git a/docs/SURFACE.md b/docs/SURFACE.md
index 460f1d27f..001790350 100644
--- a/docs/SURFACE.md
+++ b/docs/SURFACE.md
@@ -490,6 +490,28 @@ The herdr model: first-party helpers are just plugins that ship in the box; the
- **Preflight is part of the contract:** a plugin declares what must be provable before a run using it starts (credentials present, server reachable, scope grantable). A plugin that can't state its preflight doesn't load. Covenant 2 extends to the ecosystem by construction.
- Receipts, budget attribution, and identity scoping apply to plugin verbs exactly as to first-party ones — they come from the compile target, so a plugin can't opt out.
+### Initial plugin manifest and runtime
+
+`flows add helper-datadog` installs `@flows/helper-datadog` in the nearest
+`flows.json` project. See `testdata/plugins/helper-datadog/flows-plugin.json`
+for the manifest and `packages/sdk/src/plugin-manifest.ts` for validation.
+Successful installation records the package in `flows.json.plugins`; optional
+`flows-plugin.d.ts` augments `@relayflows/surface`'s `Ctx` and is added to the
+project's `tsconfig.json` include list (currently plain JSON configs).
+
+This first slice supports `lowersTo: "effect"`. A plugin supplies an ESM
+`src/index.js` exporting `execute(namespace, method, input, { idempotencyKey })`.
+The SDK snapshots arguments, validates their JSON Schema, and executes the
+provider through the existing journal-backed agent effect protocol. Providers
+must honor the supplied kernel effect key. Credentials and HTTP(S) HEAD probes
+run before authored flow bodies, including `flows check`.
+
+Follow-ups: other primitive targets, trigger/gate dispatch (currently refused
+with `plugin_unsupported`), manifest-driven type generation, JSONC tsconfigs,
+plugin code bundling/pinning, declarative-flow plugin preflight, and restart
+recovery of an interrupted plugin effect. Plugin effects currently inherit the
+internal authored executor's child-run lifecycle, not a resumable authored root.
+
## 4. Build: the immutable bundle
`flows build` seals a flow into a content-addressed, immutable bundle: canonical spec JSON, compiled TS with pinned deps, helper/plugin lockfile, assets, preflight declaration, identity signature — `flow@sha256:…`, pushed to a bucket/registry. `flows deploy` points a trigger at a digest; `flows run flow@sha256:…` executes from the bucket on any cell, no checkout. Preflight runs at build time for everything build-provable and again at deploy time for environment facts (credentials, workers, MCP servers). The working tree is for authoring; **production only ever runs digests.**
diff --git a/docs/evidence/spec-J-plugin-registry.md b/docs/evidence/spec-J-plugin-registry.md
new file mode 100644
index 000000000..6e05e95f9
--- /dev/null
+++ b/docs/evidence/spec-J-plugin-registry.md
@@ -0,0 +1,139 @@
+# Plugin registry minimal slice evidence
+
+Ships npm installation, effect-only manifests/runtime, authored preflight, and
+optional module augmentation. Follow-ups are listed in SURFACE §3.
+No kernel source changes. No new skipped tests.
+
+Dependency setup: `npm ci --ignore-scripts` in packages/sdk, followed by copying
+this worktree's built surface `dist` into its installed surface package. Bun was
+added to PATH and the kernel was built from this worktree for the final SDK run.
+An earlier SDK run used an older shared daemon and lacked Bun; final results below
+supersede that environment run. The last receipt metadata edit was covered by the
+focused rerun and both typechecks after the full suite.
+
+## Focused checks
+
+From packages/sdk:
+```
+RELAYFLOWD_BIN=/Users/khaliqgant/.relayflows-toolchain/target/2930744996/debug/relayflowd ./node_modules/.bin/vitest run tests/preflight.test.ts tests/plugin-add.test.ts tests/plugin-loader.test.ts
+```
+Captured output:
+```
+
+ RUN v2.1.9 /Users/khaliqgant/flows-spec-J-plugin/packages/sdk
+
+ ✓ tests/preflight.test.ts (27 tests) 35ms
+ ✓ tests/plugin-loader.test.ts (5 tests) 146ms
+ ✓ tests/plugin-add.test.ts (7 tests) 772ms
+ ✓ installs a real offline npm fixture and includes declarations 369ms
+ ✓ typechecks the augmented verb and rejects unknown namespaces 392ms
+
+ Test Files 3 passed (3)
+ Tests 39 passed (39)
+ Start at 19:12:07
+ Duration 1.31s (transform 281ms, setup 0ms, collect 960ms, tests 953ms, environment 0ms, prepare 98ms)
+
+```
+
+## Types
+
+From packages/sdk: `npm run typecheck && npm run typecheck:tests` (exit 0).
+Captured output:
+```
+
+> @relayflows/sdk@2.0.8 typecheck
+> tsc --noEmit && tsc -p tsconfig.type-tests.json
+
+
+> @relayflows/sdk@2.0.8 typecheck:tests
+> tsc -p tsconfig.tests.json
+
+```
+
+## Kernel
+
+From kernel:
+```
+PATH=/Users/khaliqgant/.rustup/toolchains/stable-aarch64-apple-darwin/bin:$PATH sh ../ops/cargo.sh test --workspace
+```
+Exit 0. Captured result lines (full local log: /private/tmp/spec-J-kernel.log):
+```
+test result: ok. 40 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.56s
+test result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s
+test result: ok. 3 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.03s
+test result: ok. 40 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 37.52s
+test result: ok. 6 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.07s
+test result: ok. 3 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.02s
+test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.04s
+test result: ok. 2 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.08s
+test result: ok. 3 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 3.55s
+test result: ok. 3 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.07s
+test result: ok. 5 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.03s
+test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.03s
+test result: ok. 4 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.07s
+test result: ok. 3 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.58s
+test result: ok. 3 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.03s
+test result: ok. 2 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.02s
+test result: ok. 4 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.19s
+test result: ok. 3 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.03s
+test result: ok. 3 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.08s
+test result: ok. 60 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.59s
+test result: ok. 4 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.01s
+test result: ok. 9 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.03s
+test result: ok. 28 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.12s
+test result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s
+test result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s
+test result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s
+```
+
+## Full SDK suite
+
+From packages/sdk:
+```
+PATH=/Users/khaliqgant/.bun/bin:$PATH RELAYFLOWD_BIN=/Users/khaliqgant/.relayflows-toolchain/target/2930744996/debug/relayflowd ./node_modules/.bin/vitest run
+```
+Exit 1: the real Claude analyzer readiness probe is unavailable. This is an
+unpassed acceptance test, not a green full suite. Existing opt-in adapter tests
+remain skipped; no skip flags were enabled. Captured final output excerpt
+(full local log: /private/tmp/spec-J-sdk-final.log):
+```
+ ✓ built flows CLI against live relayflowd > preflights before journaling and names an unreachable socket 1972ms
+ ✓ built flows CLI against live relayflowd > starts exactly one daemon when two runs race for one empty data dir 967ms
+ ✓ surface resume after a real daemon kill > resumes a three-step run with each successful completion exactly once 1845ms
+
+⎯⎯⎯⎯⎯⎯⎯ Failed Tests 1 ⎯⎯⎯⎯⎯⎯⎯
+
+ FAIL tests/live-kernel.test.ts > built flows CLI against live relayflowd > hn-monitor analyze-story reaches done through the real Claude analyzer CLI
+Error: LIVE_ANALYZER_UNAVAILABLE: "/Users/khaliqgant/flows-spec-J-plugin/testdata/preflight/analyze-story-claude-cli auth status" exited 1: analyze-story-claude-cli: "claude -p --model claude-haiku-4-5-20251001" exited 1: — failing because gate-2 acceptance requires the real analyzer to execute. Set RELAYFLOWS_ALLOW_ANALYZER_SKIP=1 only if this run is not gate evidence.
+ ❯ tests/live-kernel.test.ts:1223:15
+ 1221| const notice = `LIVE_ANALYZER_UNAVAILABLE: ${readiness.detail}`;
+ 1222| if (process.env['RELAYFLOWS_ALLOW_ANALYZER_SKIP'] !== '1') {
+ 1223| throw new Error(
+ | ^
+ 1224| `${notice} — failing because gate-2 acceptance requires the …
+ 1225| + 'Set RELAYFLOWS_ALLOW_ANALYZER_SKIP=1 only if this run is …
+
+⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯[1/1]⎯
+
+ Test Files 1 failed | 69 passed | 1 skipped (71)
+ Tests 1 failed | 1220 passed | 3 skipped (1224)
+ Start at 19:11:02
+ Duration 57.03s (transform 1.25s, setup 0ms, collect 8.68s, tests 214.59s, environment 6ms, prepare 1.97s)
+
+```
+
+## CLI smoke
+
+Scratch project used a fixture-only npm shim that forwards to real npm's offline
+local-package install, rather than contacting the public npm registry.
+```
+Command: node packages/sdk/dist/cli.js add helper-datadog (scratch project, npm stand-in installs local offline fixture)
+Added @flows/helper-datadog
+exit=0
+{
+ "plugins": [
+ "@flows/helper-datadog"
+ ]
+}
+
+```
diff --git a/packages/sdk/src/authored-flow-executor.ts b/packages/sdk/src/authored-flow-executor.ts
index cd06d6ca1..4910dd229 100644
--- a/packages/sdk/src/authored-flow-executor.ts
+++ b/packages/sdk/src/authored-flow-executor.ts
@@ -1,3 +1,5 @@
+import { pluginHelpers } from './plugin-loader.js';
+import { runPluginEffect } from './authored-plugin-effect.js';
import { randomUUID } from 'node:crypto';
import { dirname } from 'node:path';
import { assertSlackCredentials, runSlackEffect } from './authored-slack-effect.js';
@@ -306,6 +308,17 @@ export async function executeAuthoredFlow(
),
};
+ Object.assign(context, pluginHelpers(checkedMcp.plugins ?? [], (plugin, verb, args) => {
+ const label = `${verb.namespace}.${verb.method}`;
+ assertOperationAllowed(label, definition.name, requestedCompletion);
+ const id = `plugin-${nextStep++}`;
+ return trackStep(authoredSteps, new AuthoredFlowOperation(
+ id, label, () => assertOperationAllowed(label, definition.name, requestedCompletion),
+ () => runPluginEffect(journal, definition.name, id, plugin, verb, args, journalSteps, budget),
+ lifecycle,
+ ));
+ }));
+
let bodyFailed = false;
let bodyFailure: unknown;
try {
diff --git a/packages/sdk/src/authored-plugin-effect.ts b/packages/sdk/src/authored-plugin-effect.ts
new file mode 100644
index 000000000..186f1ed0e
--- /dev/null
+++ b/packages/sdk/src/authored-plugin-effect.ts
@@ -0,0 +1,152 @@
+import { createHash, randomUUID } from 'node:crypto';
+import { compileSpec, toKernelSpec } from './compile.js';
+import { invokePlugin, type LoadedPlugin } from './plugin-loader.js';
+import type { PluginVerb } from './plugin-manifest.js';
+import type { AuthoredBudget } from './authored-budget.js';
+import { SPEC_SCHEMA_VERSION } from './spec.js';
+import type { JournalClient } from './journal-client.js';
+import type { StepDispatchEvent } from './protocol.js';
+import { AuthoredFlowExecutionError } from './authored-flow-error.js';
+import type { AuthoredFlowJournalStep } from './authored-flow-executor.js';
+import { readCompletedStepOutput } from './authored-step-output.js';
+import { withWorkerLease } from './worker-lease.js';
+
+export class PluginStepError extends AuthoredFlowExecutionError {
+ constructor(readonly diagnostic: string, runId: string) {
+ super('step_failed', diagnostic, 'worker_error', runId);
+ }
+}
+
+/** Helpers lower to an existing agent effect, never a fourth kernel primitive.
+ * The Plugin receipt lives in step.completed.output: { type, input, output, ... }.
+ * A private stream routes the step to this short-lived SDK worker exclusively.
+ */
+export async function runPluginEffect(
+ journal: JournalClient, flowName: string, id: string, plugin: LoadedPlugin, verb: PluginVerb,
+ args: unknown, journalSteps: AuthoredFlowJournalStep[], budget: AuthoredBudget,
+): Promise {
+ const server = verb.namespace;
+ const tool = verb.method;
+ const idempotencyKey = `plugin:${server}:${tool}:${createHash('sha256').update(JSON.stringify(args)).digest('hex')}`;
+ const surfacePath = `/plugins/${pathPart(server)}/${pathPart(tool)}`;
+ const stream = `plugin-worker-${randomUUID()}`;
+ const instruction = JSON.stringify({ type: 'effect', server, tool, input: args });
+ const peer = journal.createPeer();
+ let diagnostic: string | undefined;
+ let settled!: () => void;
+ let failed!: (error: unknown) => void;
+ const completed = new Promise((resolve, reject) => { settled = resolve; failed = reject; });
+ void completed.catch(() => undefined);
+ let timer: ReturnType | undefined;
+ let work: Promise | undefined;
+ let dispatchExpired = false;
+ const dispatch = (event: StepDispatchEvent): void => {
+ if (dispatchExpired) return;
+ const dispatched = event.spec as { instruction?: string; surfaces?: { streams?: { stream: string }[] } };
+ if (event.step_id !== id || event.step_type !== 'agent' || work !== undefined
+ || dispatched?.instruction !== instruction
+ || !dispatched.surfaces?.streams?.some(pin => pin.stream === stream)) {
+ failed(new Error('Plugin worker received an unexpected dispatch'));
+ return;
+ }
+ clearTimeout(timer);
+ timer = undefined;
+ work = execute(event);
+ void work.then(settled, failed);
+ };
+ async function execute(event: StepDispatchEvent): Promise {
+ const receipt = { type: 'effect' as const, plugin: plugin.manifest.name, version: plugin.manifest.version,
+ namespace: server, method: tool, input: args, idempotencyKey: event.idempotency_key };
+ let output: unknown;
+ let confirmed = false;
+ try {
+ // Renew the worker lease while the Plugin tool is in flight. Without this,
+ // a tool that runs longer than the initial lease loses ownership and
+ // the write-back path fails — see worker-lease.ts for the renewal contract.
+ confirmed = await withWorkerLease(peer, event, async () => {
+ return peer.performEffect({
+ runId: event.run_id, stepId: id, attempt: event.attempt, idempotencyKey: event.idempotency_key,
+ surfacePath, revisionBefore: 'pending', revisionAfter: idempotencyKey,
+ }, async () => {
+ try { output = await invokePlugin(plugin, verb, args, event.idempotency_key); }
+ catch { diagnostic = 'plugin_execution_failed'; throw new PluginExecutionFailure(); }
+ });
+ });
+ // A confirmed election without its receipt is an interrupted writeback,
+ // not a successful result we may invent or a call we may safely repeat.
+ if (!confirmed) throw new Error('plugin_result_unavailable');
+ } catch (error) {
+ if (error instanceof PluginExecutionFailure) diagnostic = 'plugin_execution_failed';
+ else if (error instanceof Error && error.message === 'plugin_result_unavailable') diagnostic = error.message;
+ else throw error; // Journal failures remain fail-closed, never provider errors.
+ }
+ await peer.stepComplete(event.run_id, id, event.attempt, event.idempotency_key,
+ diagnostic === undefined ? 'success' : 'worker_error', {
+ output: { ...receipt, output: output ?? null, ...(diagnostic ? { diagnostic } : {}) },
+ started_pins: event.pins, end_pins: event.pins,
+ ...(diagnostic ? { trajectory_tail: { ...receipt, diagnostic } } : {}),
+ effects: confirmed ? [{ surface_path: surfacePath, idempotency_key: event.idempotency_key }] : [],
+ });
+ }
+ let childRunId: string | undefined;
+ async function cancelChildRun(): Promise {
+ const runId = childRunId;
+ childRunId = undefined;
+ if (runId !== undefined) {
+ try { await journal.runCancel(runId); } catch { /* fail-open on cleanup */ }
+ }
+ }
+ const deadlineError = new Error('Plugin worker dispatch deadline exceeded');
+ let finishDeadline!: () => void;
+ const deadline = new Promise((resolve, reject) => {
+ finishDeadline = resolve;
+ // One deadline covers connection, initialization, run creation and dispatch.
+ timer = setTimeout(() => {
+ dispatchExpired = true;
+ failed(deadlineError);
+ reject(deadlineError);
+ }, 30_000);
+ });
+ try {
+ await Promise.race([peer.connect(), deadline]);
+ await Promise.race([peer.hello('flows-plugin'), deadline]);
+ peer.on('step.dispatch', dispatch);
+ await Promise.race([peer.workerAttach(stream, ['agent'], { workspace: [], streams: [{ stream, read_offset: 0 }] }, 1), deadline]);
+ const spec = toKernelSpec(compileSpec({ version: SPEC_SCHEMA_VERSION, name: `${flowName}/${id}`,
+ steps: [{ id, type: 'agent', instruction,
+ surfaces: { streams: [{ stream }], external: [surfacePath] }, maxIterations: 1 }],
+ }));
+ return await Promise.race([budget.execute(journal, spec, async outcome => {
+ childRunId = outcome.run_id;
+ // run.start may answer after the deadline and the outer cleanup.
+ if (dispatchExpired) await cancelChildRun();
+ await completed;
+ if (diagnostic !== undefined) {
+ try { await readCompletedStepOutput(journal, outcome.run_id, id, journalSteps); }
+ catch (error) {
+ if (!(error instanceof AuthoredFlowExecutionError) || error.code !== 'step_failed') throw error;
+ throw new PluginStepError(diagnostic, outcome.run_id);
+ }
+ }
+ const receipt = await readCompletedStepOutput(journal, outcome.run_id, id, journalSteps) as { output: unknown };
+ return receipt.output;
+ }), deadline]);
+ } finally {
+ clearTimeout(timer);
+ finishDeadline();
+ peer.off('step.dispatch', dispatch);
+ // If the dispatch deadline fired we started a child run that will never
+ // be worked -- cancel it so its journal doesn't stay indeterminate.
+ // Best-effort: the parent flow already surfaced the deadline via the
+ // rejected `completed` promise, so a cancel error here must not mask it.
+ if (dispatchExpired) await cancelChildRun();
+ peer.close();
+ await work?.catch(() => undefined);
+ }
+}
+
+function pathPart(value: string): string {
+ return encodeURIComponent(value).replaceAll('.', '%2E');
+}
+
+class PluginExecutionFailure extends Error {}
diff --git a/packages/sdk/src/cli.ts b/packages/sdk/src/cli.ts
index c6437bda9..b4e245332 100644
--- a/packages/sdk/src/cli.ts
+++ b/packages/sdk/src/cli.ts
@@ -1,4 +1,5 @@
#!/usr/bin/env node
+import { addPlugin } from './cli/add.js';
import { watchCheck } from './cli-watch.js';
import { checkHelperBody } from './cli/check-helper-body.js';
@@ -44,6 +45,7 @@ export interface CliIo {
type CliExitCode = 0 | 1 | 2 | 3;
type ParsedArgs =
+ | { command: 'add'; value: string }
| ReplayArgs
| BuildArgs
| DeployArgs
@@ -61,6 +63,7 @@ type ParsedArgs =
const DEFAULT_DATA_DIR = '.relayflowd';
const USAGE = [
'Usage:',
+ 'flows add ',
'flows build [--out ] ',
'flows build --verify ',
'flows deploy @sha256: --to ',
@@ -108,6 +111,8 @@ export async function runCli(
return 2;
}
+ if (parsed.command === 'add') return addPlugin(parsed.value, io);
+
if (parsed.command === 'serve-webhook') return runServeWebhook(parsed, io);
if (parsed.command === 'cloud-run') return runCloudCli(parsed, io);
@@ -409,6 +414,7 @@ function emitWait(
function parseArgs(args: readonly string[]): ParsedArgs | undefined {
const command = args[0];
+ if (command === 'add') return args.length === 2 ? { command: 'add', value: args[1]! } : undefined;
if (command === 'replay') return parseReplayArgs(args.slice(1));
if (command === 'build') return parseBuildArgs(args.slice(1));
if (command === 'deploy') return parseDeployArgs(args.slice(1));
diff --git a/packages/sdk/src/cli/add.ts b/packages/sdk/src/cli/add.ts
new file mode 100644
index 000000000..46912359e
--- /dev/null
+++ b/packages/sdk/src/cli/add.ts
@@ -0,0 +1,45 @@
+import { execFileSync } from 'node:child_process';
+import { existsSync, readFileSync, writeFileSync } from 'node:fs';
+import { join } from 'node:path';
+import type { CliIo } from '../cli.js';
+import { findPluginProject, probePlugin, readPlugin } from '../plugin-loader.js';
+import { PluginError, pluginPackageName } from '../plugin-manifest.js';
+
+export async function addPlugin(name: string, io: CliIo, options: {
+ cwd?: string;
+ install?: (packageName: string, root: string) => void;
+} = {}): Promise<0 | 2> {
+ try {
+ const packageName = pluginPackageName(name);
+ const root = findPluginProject(options.cwd ?? process.cwd());
+ if (!root) throw new PluginError('plugin_manifest_invalid', 'flows add requires a project with flows.json.');
+ const configPath = join(root, 'flows.json');
+ const config = JSON.parse(readFileSync(configPath, 'utf8'));
+ if (!config || Array.isArray(config) || typeof config !== 'object' || (config.plugins !== undefined && (!Array.isArray(config.plugins) || !config.plugins.every((p: unknown) => typeof p === 'string')))) throw new PluginError('plugin_manifest_invalid', 'Invalid flows.json plugins list.');
+ const tsPath = join(root, 'tsconfig.json');
+ const tsconfig = existsSync(tsPath) ? { config: JSON.parse(readFileSync(tsPath, 'utf8')) } : { config: {} };
+ if (!tsconfig.config || typeof tsconfig.config !== 'object' || Array.isArray(tsconfig.config) || (tsconfig.config.include !== undefined && (!Array.isArray(tsconfig.config.include) || !tsconfig.config.include.every((p: unknown) => typeof p === 'string')))) throw new PluginError('plugin_manifest_invalid', 'Invalid tsconfig.json include list.');
+ try {
+ (options.install ?? ((pkg, cwd) => { execFileSync('npm', ['install', '--save', pkg], { cwd, encoding: 'utf8', stdio: 'pipe' }); }))(packageName, root);
+ } catch (error) {
+ const message = String((error as { stderr?: unknown }).stderr ?? error);
+ throw new PluginError(/E404|404 Not Found/.test(message) ? 'plugin_unknown' : 'plugin_install_failed', `Could not install ${packageName}.`);
+ }
+ const plugin = readPlugin(join(root, 'node_modules', packageName), packageName);
+ await probePlugin(plugin);
+ if (!existsSync(join(plugin.directory, 'src/index.js'))) throw new PluginError('plugin_manifest_invalid', 'Plugin requires src/index.js.');
+ const declaration = `node_modules/${packageName}/flows-plugin.d.ts`;
+ if (existsSync(join(root, declaration))) {
+ tsconfig.config.include = [...new Set([...(tsconfig.config.include ?? ['**/*']), declaration])];
+ writeFileSync(tsPath, `${JSON.stringify(tsconfig.config, null, 2)}\n`);
+ }
+ config.plugins = [...new Set([...(config.plugins ?? []), packageName])];
+ writeFileSync(configPath, `${JSON.stringify(config, null, 2)}\n`);
+ io.stdout(`Added ${packageName}`);
+ return 0;
+ } catch (error) {
+ const refusal = error instanceof PluginError ? error : new PluginError('plugin_manifest_invalid', (error as Error).message);
+ io.stderr(`REFUSED [${refusal.code}] ${refusal.message}`);
+ return 2;
+ }
+}
diff --git a/packages/sdk/src/cli/check-typescript.ts b/packages/sdk/src/cli/check-typescript.ts
index 65d30deb8..cf35cba3d 100644
--- a/packages/sdk/src/cli/check-typescript.ts
+++ b/packages/sdk/src/cli/check-typescript.ts
@@ -1,3 +1,4 @@
+import type { LoadedPlugin } from '../plugin-loader.js';
import { dirname, resolve } from 'node:path';
import type { AuthoredFlowDefinition } from '../authored-flow.js';
import { loadAuthoredFlow } from '../authored-flow-loader.js';
@@ -6,6 +7,7 @@ import { SPEC_SCHEMA_VERSION, type McpServerConfig } from '../spec.js';
import { inputFailureReport, readProjectConfig, type CheckReport } from './check.js';
export interface CheckedMcp {
+ plugins?: readonly LoadedPlugin[];
report: CheckReport;
servers: Readonly>;
inventory: Readonly>;
@@ -38,10 +40,12 @@ export async function checkMcpHeader(
const config = readProjectConfig(dirname(resolve(path)));
const result = await preflight({ version: SPEC_SCHEMA_VERSION, name: definition.name,
steps: [{ id: 'header', type: 'deterministic', command: ':' }] }, {
+ pluginSearchStart: dirname(resolve(path)),
mcpServers: header.tools?.mcp ?? [], mcp: config.mcp,
probes: { command: () => true, cli: () => { throw new Error('no CLI declared'); }, executor: () => false },
});
return {
+ plugins: result.plugins,
servers: config.mcp ?? empty.servers,
inventory: result.mcpTools ?? empty.inventory,
report: { ...result, path, projectConfigPath: config.path, gates: [],
diff --git a/packages/sdk/src/cli/check.ts b/packages/sdk/src/cli/check.ts
index 2c294d09b..9c1f8d9eb 100644
--- a/packages/sdk/src/cli/check.ts
+++ b/packages/sdk/src/cli/check.ts
@@ -199,14 +199,17 @@ export function readProjectConfig(start: string): ProjectConfig {
} catch {
throw new CheckFailure('config_invalid', `Project config "${configPath}" is not valid JSON.`);
}
- if (!isObject(value) || Object.keys(value).some((key) => !['cli', 'executors', 'models', 'mcp', 'deploy'].includes(key))) {
- throw new CheckFailure('config_invalid', `Project config "${configPath}" expects only cli, executors, models, mcp, and deploy.`);
+ if (!isObject(value) || Object.keys(value).some((key) => !['cli', 'executors', 'models', 'mcp', 'deploy', 'plugins'].includes(key))) {
+ throw new CheckFailure('config_invalid', `Project config "${configPath}" expects only cli, executors, models, mcp, deploy, and plugins.`);
}
if (value['deploy'] !== undefined && (!isObject(value['deploy'])
|| Object.keys(value['deploy']).some(key => key !== 'bucket')
|| !isNonEmptyString(value['deploy']['bucket']))) {
throw new CheckFailure('config_invalid', `Project config "${configPath}" has invalid deploy.bucket.`);
}
+ if (value['plugins'] !== undefined && (!Array.isArray(value['plugins']) || !value['plugins'].every(isNonEmptyString))) {
+ throw new CheckFailure('config_invalid', 'Project plugins must be package names.');
+ }
if (value['cli'] !== undefined && !isNonEmptyString(value['cli'])) {
throw new CheckFailure('config_invalid', `Project config "${configPath}" has an invalid cli.`);
}
diff --git a/packages/sdk/src/failure-kinds.ts b/packages/sdk/src/failure-kinds.ts
index 7fbf59584..d60167127 100644
--- a/packages/sdk/src/failure-kinds.ts
+++ b/packages/sdk/src/failure-kinds.ts
@@ -1,7 +1,9 @@
+import { PLUGIN_FAILURE_KINDS } from './plugin-manifest.js';
const SHARED_SPEC_FAILURE_KINDS = ['invalid_spec'] as const;
/** Environment refusal kinds produced after spec validation succeeds. */
const PREFLIGHT_ENVIRONMENT_FAILURE_KINDS = [
+ ...PLUGIN_FAILURE_KINDS,
'helper_slack.credential_missing',
'helper_slack.mount_required',
'mcp_undeclared_server',
diff --git a/packages/sdk/src/plugin-loader.ts b/packages/sdk/src/plugin-loader.ts
new file mode 100644
index 000000000..8911590ea
--- /dev/null
+++ b/packages/sdk/src/plugin-loader.ts
@@ -0,0 +1,88 @@
+import { existsSync, readFileSync, readdirSync } from 'node:fs';
+import { dirname, join, resolve } from 'node:path';
+import { pathToFileURL } from 'node:url';
+import { Ajv } from 'ajv';
+import type { Step } from '@relayflows/surface';
+import { snapshotJsonValue } from './json-value.js';
+import { assertSupportedPlugin, PluginError, pluginPackageName, validatePluginManifest, type PluginManifest, type PluginVerb } from './plugin-manifest.js';
+
+export interface LoadedPlugin { readonly directory: string; readonly manifest: PluginManifest }
+export function findPluginProject(start: string): string | undefined {
+ let current = resolve(start);
+ for (;;) {
+ if (existsSync(join(current, 'flows.json'))) return current;
+ const parent = dirname(current);
+ if (parent === current) return undefined;
+ current = parent;
+ }
+}
+export function readPlugin(directory: string, packageName: string): LoadedPlugin {
+ const path = join(directory, 'flows-plugin.json');
+ if (!existsSync(path)) throw new PluginError('plugin_manifest_missing', `${packageName} has no flows-plugin.json.`);
+ let input: unknown;
+ try { input = JSON.parse(readFileSync(path, 'utf8')); }
+ catch { throw new PluginError('plugin_manifest_invalid', `${packageName} manifest is unreadable or invalid JSON.`); }
+ const manifest = validatePluginManifest(input, packageName);
+ assertSupportedPlugin(manifest);
+ return Object.freeze({ directory, manifest });
+}
+export async function probePlugin(plugin: LoadedPlugin, env: NodeJS.ProcessEnv = process.env): Promise {
+ for (const credential of plugin.manifest.preflight.credentials) {
+ if (!env[credential]?.trim()) throw new PluginError('plugin_credential_missing', `${plugin.manifest.name} requires ${credential}.`);
+ }
+ for (const server of plugin.manifest.preflight.servers) {
+ try {
+ const response = await fetch(server, { method: 'HEAD', signal: AbortSignal.timeout(5000) });
+ await response.body?.cancel();
+ if (!response.ok) throw new Error('unsuccessful response');
+ } catch { throw new PluginError('plugin_server_unreachable', `${plugin.manifest.name} cannot reach ${server}.`); }
+ }
+}
+export async function loadPlugins(start: string): Promise {
+ const root = findPluginProject(start);
+ if (root === undefined) return Object.freeze([]);
+ let config: { plugins?: unknown };
+ try { config = JSON.parse(readFileSync(join(root, 'flows.json'), 'utf8')); }
+ catch { throw new PluginError('plugin_manifest_invalid', 'Invalid flows.json.'); }
+ if (config === null || typeof config !== 'object' || (config.plugins !== undefined && (!Array.isArray(config.plugins) || !config.plugins.every(p => typeof p === 'string')))) {
+ throw new PluginError('plugin_manifest_invalid', 'flows.json plugins must be package names.');
+ }
+ const scope = join(root, 'node_modules/@flows');
+ const names = new Set((config.plugins as string[] | undefined)?.map(pluginPackageName));
+ if (existsSync(scope)) for (const name of readdirSync(scope).sort()) {
+ if (name.startsWith('helper-') && !names.has(`@flows/${name}`)) {
+ throw new PluginError('plugin_unlisted', `@flows/${name} is installed but not declared in flows.json plugins. Run flows add ${name}.`);
+ }
+ }
+ const plugins = [...names].sort().map(name => readPlugin(join(root, 'node_modules', name), name));
+ const namespaces = new Set();
+ for (const plugin of plugins) {
+ for (const namespace of new Set(plugin.manifest.verbs.map(v => v.namespace))) {
+ if (namespaces.has(namespace)) throw new PluginError('plugin_manifest_invalid', `Duplicate plugin namespace ${namespace}.`);
+ namespaces.add(namespace);
+ }
+ await probePlugin(plugin);
+ if (!existsSync(join(plugin.directory, 'src/index.js'))) throw new PluginError('plugin_manifest_invalid', `${plugin.manifest.name} requires src/index.js.`);
+ }
+ return Object.freeze(plugins);
+}
+export function pluginHelpers(plugins: readonly LoadedPlugin[], invoke: (plugin: LoadedPlugin, verb: PluginVerb, args: unknown) => Step): Record {
+ const helpers: Record Step>> = Object.create(null);
+ const ajv = new Ajv({ strict: false });
+ for (const plugin of plugins) for (const verb of plugin.manifest.verbs) {
+ const validate = ajv.compile(verb.args);
+ const namespace = helpers[verb.namespace] ??= Object.create(null);
+ namespace[verb.method] = args => {
+ const snapshot = snapshotJsonValue(args, `f.${verb.namespace}.${verb.method} arguments`);
+ if (!validate(snapshot)) throw new PluginError('plugin_manifest_invalid', `Invalid arguments for ${verb.namespace}.${verb.method}: ${ajv.errorsText(validate.errors)}`);
+ return invoke(plugin, verb, snapshot);
+ };
+ }
+ for (const helper of Object.values(helpers)) Object.freeze(helper);
+ return Object.freeze(helpers);
+}
+export async function invokePlugin(plugin: LoadedPlugin, verb: PluginVerb, args: unknown, idempotencyKey: string): Promise {
+ const module = await import(pathToFileURL(join(plugin.directory, 'src/index.js')).href);
+ if (typeof module.execute !== 'function') throw new Error('plugin_runtime_invalid: expected execute export');
+ return snapshotJsonValue(await module.execute(verb.namespace, verb.method, args, Object.freeze({ idempotencyKey })), 'plugin output');
+}
diff --git a/packages/sdk/src/plugin-manifest.ts b/packages/sdk/src/plugin-manifest.ts
new file mode 100644
index 000000000..ba133444a
--- /dev/null
+++ b/packages/sdk/src/plugin-manifest.ts
@@ -0,0 +1,76 @@
+import { Ajv } from 'ajv';
+import { snapshotJsonValue } from './json-value.js';
+
+export const PLUGIN_FAILURE_KINDS = [
+ 'plugin_unknown', 'plugin_unlisted', 'plugin_install_failed', 'plugin_manifest_missing',
+ 'plugin_manifest_invalid', 'plugin_preflight_missing', 'plugin_verb_unknown_primitive',
+ 'plugin_unsupported', 'plugin_credential_missing', 'plugin_server_unreachable',
+] as const;
+export type PluginFailureKind = typeof PLUGIN_FAILURE_KINDS[number];
+export class PluginError extends Error {
+ constructor(readonly code: PluginFailureKind, message: string) { super(message); }
+}
+export interface PluginVerb {
+ namespace: string;
+ method: string;
+ lowersTo: 'run' | 'llm' | 'agent' | 'effect' | 'wait';
+ args: Record;
+}
+export interface PluginManifest {
+ name: string;
+ version: string;
+ verbs: readonly PluginVerb[];
+ triggers: readonly unknown[];
+ gates: readonly unknown[];
+ preflight: { credentials: readonly string[]; servers: readonly string[] };
+}
+const identifier = /^[A-Za-z_$][A-Za-z0-9_$]*$/;
+const reserved = new Set(['run', 'llm', 'agent', 'human', 'dispatch', 'done', 'cloud', 'memory', 'mcp', 'slack', '__proto__', 'constructor', 'prototype', 'then']);
+const object = (v: unknown): v is Record => typeof v === 'object' && v !== null && !Array.isArray(v);
+const strings = (v: unknown): v is string[] => Array.isArray(v) && v.every(s => typeof s === 'string' && s.trim().length > 0);
+export function pluginPackageName(name: string): string {
+ const bare = name.replace(/^@flows\//, '');
+ if (!/^helper-[a-z0-9]+(?:-[a-z0-9]+)*$/.test(bare)) {
+ throw new PluginError('plugin_manifest_invalid', 'Expected helper- or @flows/helper-.');
+ }
+ return `@flows/${bare}`;
+}
+export function validatePluginManifest(input: unknown, packageName?: string): PluginManifest {
+ let v: unknown;
+ try { v = snapshotJsonValue(input, 'plugin manifest'); }
+ catch { throw new PluginError('plugin_manifest_invalid', 'Plugin manifest must be JSON data.'); }
+ const invalid = (message: string): never => { throw new PluginError('plugin_manifest_invalid', message); };
+ if (!object(v)) return invalid('Expected a plugin manifest object.');
+ if (!Object.hasOwn(v, 'preflight')) throw new PluginError('plugin_preflight_missing', 'Plugin must declare preflight.');
+ if (typeof v.name !== 'string' || typeof v.version !== 'string' || !v.version.trim()) return invalid('Plugin name and version are required.');
+ const resolved = pluginPackageName(v.name);
+ if (packageName !== undefined && resolved !== packageName) return invalid('Plugin name does not match its package.');
+ if (!Array.isArray(v.verbs) || !Array.isArray(v.triggers) || !Array.isArray(v.gates)) return invalid('Expected verbs, triggers, and gates arrays.');
+ if (!object(v.preflight) || !strings(v.preflight.credentials) || !strings(v.preflight.servers)) return invalid('Preflight requires credentials and servers arrays.');
+ if (!v.preflight.credentials.every(s => /^[A-Za-z_][A-Za-z0-9_]*$/.test(s))) return invalid('Invalid credential environment variable.');
+ for (const server of v.preflight.servers) {
+ try { if (!['http:', 'https:'].includes(new URL(server).protocol)) return invalid('Servers must be HTTP(S) URLs.'); }
+ catch { return invalid('Servers must be HTTP(S) URLs.'); }
+ }
+ const seen = new Set();
+ const ajv = new Ajv({ strict: false });
+ for (const verb of v.verbs) {
+ if (!object(verb) || typeof verb.namespace !== 'string' || typeof verb.method !== 'string'
+ || !identifier.test(verb.namespace) || !identifier.test(verb.method)
+ || reserved.has(verb.namespace) || ['__proto__', 'constructor', 'prototype', 'then'].includes(verb.method)) return invalid('Invalid or reserved plugin verb.');
+ if (!['run', 'llm', 'agent', 'effect', 'wait'].includes(String(verb.lowersTo))) {
+ throw new PluginError('plugin_verb_unknown_primitive', `Unknown primitive for ${verb.namespace}.${verb.method}.`);
+ }
+ if (!object(verb.args)) return invalid('Verb args must be a JSON Schema.');
+ try { ajv.compile(verb.args); } catch { return invalid('Invalid verb argument JSON Schema.'); }
+ const key = `${verb.namespace}.${verb.method}`;
+ if (seen.has(key)) return invalid(`Duplicate verb ${key}.`);
+ seen.add(key);
+ }
+ return v as unknown as PluginManifest;
+}
+export function assertSupportedPlugin(manifest: PluginManifest): void {
+ if (manifest.triggers.length || manifest.gates.length || manifest.verbs.some(v => v.lowersTo !== 'effect')) {
+ throw new PluginError('plugin_unsupported', 'This slice supports effect verbs; trigger, gate, run, llm, agent and wait dispatch are follow-ups.');
+ }
+}
diff --git a/packages/sdk/src/preflight.ts b/packages/sdk/src/preflight.ts
index 77b9564a3..7400e6b21 100644
--- a/packages/sdk/src/preflight.ts
+++ b/packages/sdk/src/preflight.ts
@@ -1,3 +1,5 @@
+import { loadPlugins, type LoadedPlugin } from './plugin-loader.js';
+import { PluginError } from './plugin-manifest.js';
import type { FlowSpec, StepSpec, TriggerSpec, McpServerConfig } from './spec.js';
import { McpError, openMcpSession, type McpDiagnostic } from './mcp-client.js';
import { BudgetSyntaxError } from './budget.js';
@@ -64,6 +66,7 @@ export interface PreflightProbes {
}
export interface PreflightOptions {
+ pluginSearchStart?: string;
/** Validated tools.mcp header and nearest flows.json connections. */
mcpServers?: readonly string[];
mcp?: Readonly>;
@@ -103,6 +106,7 @@ export interface PreflightWarning {
export type PreflightDiagnostic = PreflightRefusal | PreflightWarning;
export interface PreflightResult {
+ plugins?: readonly LoadedPlugin[];
mcpTools?: Readonly>;
ok: boolean;
gates: StepGateInspection[];
@@ -133,7 +137,14 @@ export function preflight(flow: FlowSpec, options: PreflightOptions): PreflightR
export function preflight(flow: unknown, options: PreflightOptions): PreflightResult | Promise {
const result = preflightSync(flow, options);
if (options.mcpServers === undefined) return result;
- return probeMcp(result, options);
+ return probeMcp(result, options).then(async checked => {
+ if (!checked.ok || options.pluginSearchStart === undefined) return checked;
+ try { return { ...checked, plugins: await loadPlugins(options.pluginSearchStart) }; }
+ catch (error) {
+ if (!(error instanceof PluginError)) throw error;
+ return { ...checked, ok: false, diagnostics: [...checked.diagnostics, { severity: 'refusal' as const, kind: error.code, message: error.message }] };
+ }
+ });
}
async function probeMcp(result: PreflightResult, options: PreflightOptions): Promise {
diff --git a/packages/sdk/tests/authored-plugin-effect.test.ts b/packages/sdk/tests/authored-plugin-effect.test.ts
new file mode 100644
index 000000000..621f869e7
--- /dev/null
+++ b/packages/sdk/tests/authored-plugin-effect.test.ts
@@ -0,0 +1,113 @@
+import { EventEmitter } from 'node:events';
+import { resolve } from 'node:path';
+import { afterEach, expect, it, vi } from 'vitest';
+import { AuthoredBudget } from '../src/authored-budget.js';
+import { runPluginEffect } from '../src/authored-plugin-effect.js';
+import type { JournalClient } from '../src/journal-client.js';
+import { invokePlugin, readPlugin } from '../src/plugin-loader.js';
+
+vi.mock('../src/plugin-loader.js', async importOriginal => ({
+ ...await importOriginal(), invokePlugin: vi.fn(),
+}));
+afterEach(() => { vi.useRealTimers(); vi.restoreAllMocks(); vi.clearAllMocks(); });
+
+function fixture() {
+ vi.useFakeTimers();
+ const peer = Object.assign(new EventEmitter(), {
+ connect: vi.fn(async () => {}), hello: vi.fn(async () => {}), workerAttach: vi.fn(async () => {}),
+ close: vi.fn(),
+ performEffect: vi.fn(async (_effect, perform: () => Promise) => { await perform(); return true; }),
+ stepComplete: vi.fn(async () => {}),
+ stepHeartbeat: vi.fn(async () => ({ lease_deadline_ms: Date.now() + 60_000 })),
+ });
+ const started = Promise.withResolvers();
+ const journal = {
+ createPeer: () => peer,
+ runStart: vi.fn(async (spec: any) => { started.resolve(spec); return { run_id: 'plugin-run' }; }),
+ runCancel: vi.fn(async () => {}),
+ journalRead: vi.fn(async () => ({ entries: [{ entry_type: 'step.completed', step_id: 'plugin-1',
+ payload: { completionReason: 'success', output: { type: 'effect', output: { ok: true } } } }] })),
+ };
+ const plugin = readPlugin(resolve('../../testdata/plugins/helper-datadog'), '@flows/helper-datadog');
+ const run = () => runPluginEffect(journal as unknown as JournalClient, 'test', 'plugin-1', plugin,
+ plugin.manifest.verbs[0]!, {}, [], new AuthoredBudget(undefined));
+ return { peer, journal, started, run };
+}
+
+it.each(['connect', 'hello', 'workerAttach'] as const)('bounds a stalled %s handshake and stops setup after timeout', async stage => {
+ const f = fixture();
+ const stalled = Promise.withResolvers();
+ f.peer[stage].mockReturnValue(stalled.promise);
+ const rejected = expect(f.run()).rejects.toThrow('dispatch deadline exceeded');
+ await vi.advanceTimersByTimeAsync(30_000);
+ await rejected;
+ expect(f.peer[stage]).toHaveBeenCalledTimes(1);
+ expect(f.peer.close).toHaveBeenCalledTimes(1);
+ stalled.resolve();
+ await vi.advanceTimersByTimeAsync(0);
+ expect(f.journal.runStart).not.toHaveBeenCalled();
+ expect(f.peer.listenerCount('step.dispatch')).toBe(0);
+ expect(invokePlugin).not.toHaveBeenCalled();
+ expect(vi.getTimerCount()).toBe(0);
+});
+
+it('includes initialization in the dispatch deadline and cancels the child run', async () => {
+ const f = fixture();
+ const initialized = Promise.withResolvers();
+ f.peer.hello.mockReturnValue(initialized.promise);
+ const rejected = expect(f.run()).rejects.toThrow('dispatch deadline exceeded');
+ await vi.advanceTimersByTimeAsync(20_000);
+ initialized.resolve();
+ await f.started.promise;
+ await vi.advanceTimersByTimeAsync(9_999);
+ expect(f.peer.close).not.toHaveBeenCalled();
+ await vi.advanceTimersByTimeAsync(1);
+ await rejected;
+ expect(f.journal.runCancel).toHaveBeenCalledTimes(1);
+ expect(f.journal.runCancel).toHaveBeenCalledWith('plugin-run');
+ expect(f.peer.close).toHaveBeenCalledTimes(1);
+});
+
+it('bounds run creation and cancels a child whose run.start response arrives after timeout', async () => {
+ const f = fixture();
+ const outcome = Promise.withResolvers<{ run_id: string }>();
+ f.journal.runStart.mockImplementation(async spec => { f.started.resolve(spec); return outcome.promise; });
+ const rejected = expect(f.run()).rejects.toThrow('dispatch deadline exceeded');
+ await f.started.promise;
+ await vi.advanceTimersByTimeAsync(30_000);
+ await rejected;
+ expect(f.peer.close).toHaveBeenCalledTimes(1);
+ expect(f.journal.runCancel).not.toHaveBeenCalled();
+ outcome.resolve({ run_id: 'plugin-run' });
+ await vi.advanceTimersByTimeAsync(0);
+ expect(f.journal.runCancel).toHaveBeenCalledTimes(1);
+ expect(f.journal.runCancel).toHaveBeenCalledWith('plugin-run');
+ expect(invokePlugin).not.toHaveBeenCalled();
+});
+
+it('clears the deadline on dispatch so a provider can complete after 30 seconds', async () => {
+ const f = fixture();
+ const provider = Promise.withResolvers();
+ vi.mocked(invokePlugin).mockReturnValue(provider.promise);
+ const running = f.run();
+ const spec = await f.started.promise;
+ await vi.advanceTimersByTimeAsync(29_000);
+ f.peer.emit('step.dispatch', {
+ run_id: 'plugin-run', step_id: 'plugin-1', step_type: 'agent', spec: spec.steps[0],
+ attempt: 1, idempotency_key: 'kernel-key', pins: {},
+ lease_deadline_ms: Date.now() + 45_000, lease_id: 'lease-plugin-1',
+ });
+ try {
+ await vi.advanceTimersByTimeAsync(2_000);
+ expect(invokePlugin).toHaveBeenCalledTimes(1);
+ expect(f.peer.close).not.toHaveBeenCalled();
+ provider.resolve({ ok: true });
+ await expect(running).resolves.toEqual({ ok: true });
+ expect(f.peer.stepComplete).toHaveBeenCalledTimes(1);
+ expect(f.journal.runCancel).not.toHaveBeenCalled();
+ expect(vi.getTimerCount()).toBe(0);
+ } finally {
+ provider.resolve({ ok: true });
+ await running.catch(() => undefined);
+ }
+});
diff --git a/packages/sdk/tests/plugin-add.test.ts b/packages/sdk/tests/plugin-add.test.ts
new file mode 100644
index 000000000..f0e9fb6aa
--- /dev/null
+++ b/packages/sdk/tests/plugin-add.test.ts
@@ -0,0 +1,62 @@
+import { execFileSync } from 'node:child_process';
+import { cpSync, mkdtempSync, mkdirSync, symlinkSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join, resolve } from 'node:path';
+import { afterEach, expect, it, vi } from 'vitest';
+import { addPlugin } from '../src/cli/add.js';
+import { runCli } from '../src/cli.js';
+const dirs: string[] = [];
+const fixtureRoot = resolve('../../testdata/plugins');
+afterEach(() => { dirs.splice(0).forEach(p => rmSync(p, { recursive: true, force: true })); vi.unstubAllEnvs(); });
+function project() {
+ const cwd = mkdtempSync(join(tmpdir(), 'plugin-add-')); dirs.push(cwd);
+ writeFileSync(join(cwd, 'flows.json'), '{}');
+ writeFileSync(join(cwd, 'package.json'), '{"private":true}');
+ const messages: string[] = [];
+ const io = { stdout: (s: string) => messages.push(s), stderr: (s: string) => messages.push(s) };
+ const install = (name: string, root: string) => cpSync(join(fixtureRoot, name.replace('@flows/', '')), join(root, 'node_modules', name), { recursive: true });
+ return { cwd, io, install, messages };
+}
+it('installs a real offline npm fixture and includes declarations', async () => {
+ vi.stubEnv('DATADOG_API_KEY', 'test'); const p = project();
+ const install = (name: string, root: string) => {
+ expect(name).toBe('@flows/helper-datadog');
+ execFileSync('npm', ['install', '--save', '--ignore-scripts', '--offline', join(fixtureRoot, 'helper-datadog')], { cwd: root, stdio: 'pipe' });
+ };
+ expect(await addPlugin('helper-datadog', p.io, { cwd: p.cwd, install })).toBe(0);
+ expect(JSON.parse(readFileSync(join(p.cwd, 'flows.json'), 'utf8')).plugins).toEqual(['@flows/helper-datadog']);
+ expect(JSON.parse(readFileSync(join(p.cwd, 'tsconfig.json'), 'utf8')).include).toContain('node_modules/@flows/helper-datadog/flows-plugin.d.ts');
+});
+it.each([['helper-datadog', 'plugin_credential_missing'], ['helper-broken', 'plugin_manifest_missing'], ['helper-no-preflight', 'plugin_preflight_missing']])('refuses %s with %s', async (name, code) => {
+ vi.stubEnv('DATADOG_API_KEY', ''); const p = project();
+ expect(await addPlugin(name, p.io, p)).toBe(2);
+ expect(p.messages.join('\n')).toContain(code);
+ expect(JSON.parse(readFileSync(join(p.cwd, 'flows.json'), 'utf8'))).toEqual({});
+});
+it('classifies npm failures', async () => {
+ for (const [stderr, code] of [['npm error code E404', 'plugin_unknown'], ['offline', 'plugin_install_failed']]) {
+ const p = project();
+ expect(await addPlugin('helper-nonexistent-npm-pkg', p.io, { cwd: p.cwd, install: () => { throw { stderr }; } })).toBe(2);
+ expect(p.messages.join('\n')).toContain(code);
+ }
+});
+it('wires add into CLI dispatch', async () => {
+ const p = project(); expect(await runCli(['add', '../bad'], p.io)).toBe(2);
+ expect(p.messages.join('\n')).toContain('plugin_manifest_invalid');
+});
+
+it('typechecks the augmented verb and rejects unknown namespaces', async () => {
+ vi.stubEnv('DATADOG_API_KEY', 'test'); const p = project();
+ expect(await addPlugin('helper-datadog', p.io, p)).toBe(0);
+ mkdirSync(join(p.cwd, 'node_modules/@relayflows'), { recursive: true });
+ symlinkSync(resolve('node_modules/@relayflows/surface'), join(p.cwd, 'node_modules/@relayflows/surface'));
+ const config = JSON.parse(readFileSync(join(p.cwd, 'tsconfig.json'), 'utf8'));
+ config.compilerOptions = { strict: true, target: 'ES2022', module: 'ESNext', moduleResolution: 'Bundler', skipLibCheck: true, noEmit: true };
+ writeFileSync(join(p.cwd, 'tsconfig.json'), JSON.stringify(config));
+ const source = "import { flow } from '@relayflows/surface'; export default flow('x', async f => { await f.datadog.query({metric:'test'}); f.done('success'); });";
+ writeFileSync(join(p.cwd, 'flow.ts'), source);
+ const tsc = resolve('node_modules/typescript/bin/tsc');
+ execFileSync(process.execPath, [tsc, '-p', join(p.cwd, 'tsconfig.json')], { stdio: 'pipe' });
+ writeFileSync(join(p.cwd, 'flow.ts'), source.replace('f.datadog.query', 'f.notarealplugin.foo'));
+ expect(() => execFileSync(process.execPath, [tsc, '-p', join(p.cwd, 'tsconfig.json')], { stdio: 'pipe' })).toThrow();
+});
diff --git a/packages/sdk/tests/plugin-loader.test.ts b/packages/sdk/tests/plugin-loader.test.ts
new file mode 100644
index 000000000..daffe4765
--- /dev/null
+++ b/packages/sdk/tests/plugin-loader.test.ts
@@ -0,0 +1,89 @@
+import { spawnSync, spawn, type ChildProcess } from 'node:child_process';
+import { once } from 'node:events';
+import { cpSync, mkdtempSync, readFileSync, realpathSync, rmSync, writeFileSync } from 'node:fs';
+import { homedir, tmpdir } from 'node:os';
+import { join, resolve } from 'node:path';
+import { flow } from '@relayflows/surface';
+import { afterEach, expect, it, vi } from 'vitest';
+import { loadPlugins, pluginHelpers, probePlugin, readPlugin } from '../src/plugin-loader.js';
+import { validatePluginManifest } from '../src/plugin-manifest.js';
+import { executeAuthoredFlow } from '../src/authored-flow-executor.js';
+import { JournalClient } from '../src/journal-client.js';
+import { socketPathFor } from '../src/daemon-connection.js';
+const fixture = resolve('../../testdata/plugins/helper-datadog');
+const manifest = JSON.parse(readFileSync(join(fixture, 'flows-plugin.json'), 'utf8'));
+const dirs: string[] = []; const children: ChildProcess[] = []; const clients: JournalClient[] = [];
+afterEach(async () => {
+ clients.splice(0).forEach(c => c.close());
+ for (const child of children.splice(0)) if (child.exitCode === null && child.signalCode === null) { const exit = once(child, 'exit'); child.kill(); await exit; }
+ dirs.splice(0).forEach(p => rmSync(p, { recursive: true, force: true })); vi.unstubAllEnvs(); vi.restoreAllMocks();
+});
+function project() {
+ const root = mkdtempSync(join(tmpdir(), 'plugin-load-')); dirs.push(root);
+ writeFileSync(join(root, 'flows.json'), JSON.stringify({ plugins: ['@flows/helper-datadog'] }));
+ cpSync(fixture, join(root, 'node_modules/@flows/helper-datadog'), { recursive: true }); return root;
+}
+it.each([{}, { plugins: [] }])('rejects an installed helper without an explicit declaration: %j', async config => {
+ const root = project();
+ writeFileSync(join(root, 'flows.json'), JSON.stringify(config));
+ vi.stubEnv('DATADOG_API_KEY', '');
+ await expect(loadPlugins(root)).rejects.toMatchObject({ code: 'plugin_unlisted' });
+});
+it('rejects an extra installed helper before probing the declared plugins', async () => {
+ const root = project();
+ cpSync(fixture, join(root, 'node_modules/@flows/helper-extra'), { recursive: true });
+ vi.stubEnv('DATADOG_API_KEY', '');
+ await expect(loadPlugins(root)).rejects.toMatchObject({ code: 'plugin_unlisted', message: expect.stringContaining('@flows/helper-extra') });
+});
+it('loads explicitly declared helpers using either supported package name spelling', async () => {
+ const root = project();
+ vi.stubEnv('DATADOG_API_KEY', 'test');
+ for (const name of ['helper-datadog', '@flows/helper-datadog']) {
+ writeFileSync(join(root, 'flows.json'), JSON.stringify({ plugins: [name] }));
+ expect((await loadPlugins(root)).map(plugin => plugin.manifest.name)).toEqual([manifest.name]);
+ }
+});
+it('freezes manifest and refuses unknown primitives and missing preflight', () => {
+ expect(Object.isFrozen(validatePluginManifest(manifest).verbs[0]!.args)).toBe(true);
+ expect(() => validatePluginManifest({ ...manifest, verbs: [{ ...manifest.verbs[0], lowersTo: 'new-word' }] })).toThrow('Unknown primitive');
+ const { preflight, ...rest } = manifest; expect(() => validatePluginManifest(rest)).toThrow('declare preflight');
+ expect(() => validatePluginManifest({ ...manifest, verbs: [{ ...manifest.verbs[0], namespace: 'run' }] })).toThrow('reserved');
+});
+it('checks credentials and server reachability', async () => {
+ const plugin = readPlugin(fixture, '@flows/helper-datadog');
+ await expect(probePlugin(plugin, {})).rejects.toMatchObject({ code: 'plugin_credential_missing' });
+ vi.spyOn(globalThis, 'fetch').mockRejectedValue(new Error('offline'));
+ const withServer = { ...plugin, manifest: validatePluginManifest({ ...manifest, preflight: { credentials: [], servers: ['https://example.invalid/health'] } }) };
+ await expect(probePlugin(withServer)).rejects.toMatchObject({ code: 'plugin_server_unreachable' });
+});
+it('snapshots and validates inputs before invoking', async () => {
+ vi.stubEnv('DATADOG_API_KEY', 'test'); const plugins = await loadPlugins(project()); const invoke = vi.fn();
+ const helpers = pluginHelpers(plugins, invoke) as any; const args = { metric: 'cpu' };
+ helpers.datadog.query(args); args.metric = 'mutated';
+ expect(invoke.mock.calls[0]![2]).toEqual({ metric: 'cpu' }); expect(Object.isFrozen(invoke.mock.calls[0]![2])).toBe(true);
+ expect(() => helpers.datadog.query({ metric: 1 })).toThrow('Invalid arguments');
+});
+it('refuses before evaluating the body or contacting the journal', async () => {
+ vi.stubEnv('DATADOG_API_KEY', ''); const body = vi.fn(async f => f.done('success'));
+ await expect(executeAuthoredFlow(flow('refusal', body), new JournalClient('/no-socket'), undefined, { flowPath: join(project(), 'flow.ts') })).rejects.toMatchObject({ report: { diagnostics: expect.arrayContaining([expect.objectContaining({ kind: 'plugin_credential_missing' })]) } });
+ expect(body).not.toHaveBeenCalled();
+});
+it('journals plugin effect input and receipt through the daemon', async () => {
+ vi.stubEnv('DATADOG_API_KEY', 'test'); const root = project(); const dataDir = join(root, 'data');
+ const worktreeKey = spawnSync('cksum', { input: realpathSync(resolve('../..')), encoding: 'utf8' }).stdout.trim().split(' ')[0]!;
+ const target = process.env.CARGO_TARGET_DIR ?? join(process.env.RELAYFLOWS_TOOLCHAIN_HOME ?? join(homedir(), '.relayflows-toolchain'), 'target', worktreeKey);
+ const binary = process.env.RELAYFLOWD_BIN ?? join(target, 'debug/relayflowd');
+ const child = spawn(binary!, ['--data-dir', dataDir, 'serve'], { stdio: 'ignore' }); children.push(child);
+ let client!: JournalClient;
+ for (let n = 0; ; n++) {
+ client = new JournalClient(socketPathFor(dataDir), { connectTimeoutMs: 100 });
+ try { await client.connect(); await client.hello('plugin-test'); clients.push(client); break; }
+ catch (error) { client.close(); if (n === 99) throw error; await new Promise(r => setTimeout(r, 20)); }
+ }
+ let receipt: unknown;
+ const result = await executeAuthoredFlow(flow('plugin', async f => { receipt = await (f as any).datadog.query({ metric: 'cpu' }); f.done('success'); }), client, undefined, { flowPath: join(root, 'flow.ts') });
+ expect(receipt).toMatchObject({ metric: 'cpu', value: 42 }); const step = result.journalSteps[0]!;
+ const entries = (await client.journalRead(step.runId, 1)).entries as any[];
+ expect(entries.filter(e => e.entry_type === 'effect.confirmed')).toHaveLength(1);
+ expect(entries.find(e => e.entry_type === 'step.completed')).toMatchObject({ payload: { completionReason: 'success', output: { type: 'effect', input: { metric: 'cpu' }, output: { value: 42 } } } });
+}, 15000);
diff --git a/packages/sdk/tests/preflight.test.ts b/packages/sdk/tests/preflight.test.ts
index cdb888d4f..601b948d9 100644
--- a/packages/sdk/tests/preflight.test.ts
+++ b/packages/sdk/tests/preflight.test.ts
@@ -1,3 +1,8 @@
+import { mkdtempSync, mkdirSync, writeFileSync, rmSync } from 'node:fs';
+import { join } from 'node:path';
+import { tmpdir } from 'node:os';
+import { addPlugin } from '../src/cli/add.js';
+import type { PreflightFailureKind } from '../src/failure-kinds.js';
import { preflightHelpers } from '../src/preflight.js';
import { describe, expect, it } from 'vitest';
import {
@@ -395,6 +400,50 @@ describe('preflight: CLI resolution and refusal predicates', () => {
expect(memoryRefusal).toMatchObject({ kind: 'memory_unreachable', severity: 'refusal' });
refusalKinds.push(memoryRefusal!.kind);
expect(JSON.stringify(memoryRefusal)).not.toContain('raw secret');
+ // Plugin refusals exercise the public loader boundary, not synthetic diagnostics.
+ const basePlugin = { name: 'helper-test', version: '0.1.0', verbs: [], triggers: [], gates: [], preflight: { credentials: [], servers: [] } };
+ const cases = [
+ undefined,
+ {},
+ { ...basePlugin, version: null },
+ { ...basePlugin, verbs: [{ namespace: 'test', method: 'call', lowersTo: 'new-word', args: {} }] },
+ { ...basePlugin, triggers: [{}] },
+ { ...basePlugin, preflight: { credentials: ['FLOWS_J_TEST_MISSING_CREDENTIAL'], servers: [] } },
+ { ...basePlugin, preflight: { credentials: [], servers: ['http://127.0.0.1:1'] } },
+ ];
+ for (const manifest of cases) {
+ const root = mkdtempSync(join(tmpdir(), 'plugin-taxonomy-'));
+ try {
+ writeFileSync(join(root, 'flows.json'), JSON.stringify({ plugins: ['@flows/helper-test'] }));
+ const directory = join(root, 'node_modules/@flows/helper-test');
+ mkdirSync(directory, { recursive: true });
+ if (manifest !== undefined) writeFileSync(join(directory, 'flows-plugin.json'), JSON.stringify(manifest));
+ const result = await preflight(flow({ id: 'a', type: 'deterministic', command: 'x' }), {
+ probes: probes(), mcpServers: [], pluginSearchStart: root,
+ });
+ refusalKinds.push(...result.diagnostics.filter(d => d.severity === 'refusal').map(d => d.kind as PreflightFailureKind));
+ for (const stderr of ['E404', 'offline']) {
+ await addPlugin('helper-test', { stdout() {}, stderr(line) {
+ refusalKinds.push(line.match(/\[([^\]]+)\]/)![1] as PreflightFailureKind);
+ } }, { cwd: root, install() { throw { stderr }; } });
+ }
+ } finally { rmSync(root, { recursive: true, force: true }); }
+ }
+ // `plugin_unlisted` fires when a @flows/helper-* package is present in
+ // node_modules but is missing from the declared plugins list — the loader
+ // refuses to auto-load undeclared packages (plugin-loader.ts). Exercise it
+ // with an installed but unlisted helper directory.
+ {
+ const root = mkdtempSync(join(tmpdir(), 'plugin-unlisted-'));
+ try {
+ writeFileSync(join(root, 'flows.json'), JSON.stringify({ plugins: [] }));
+ mkdirSync(join(root, 'node_modules/@flows/helper-unlisted'), { recursive: true });
+ const result = await preflight(flow({ id: 'a', type: 'deterministic', command: 'x' }), {
+ probes: probes(), mcpServers: [], pluginSearchStart: root,
+ });
+ refusalKinds.push(...result.diagnostics.filter(d => d.severity === 'refusal').map(d => d.kind as PreflightFailureKind));
+ } finally { rmSync(root, { recursive: true, force: true }); }
+ }
expect(new Set(refusalKinds)).toEqual(new Set(PREFLIGHT_FAILURE_KINDS));
expect(JSON.stringify(scenarios)).not.toContain('raw secret');
});
diff --git a/packages/surface/src/index.ts b/packages/surface/src/index.ts
index 7af220819..dac12590c 100644
--- a/packages/surface/src/index.ts
+++ b/packages/surface/src/index.ts
@@ -25,3 +25,4 @@ export { flowRunWritebackIdempotency, type SlackHelper, type SlackReceipt } from
export type { Helpers } from "./helpers/index.js";
export type { MemoryHelper, MemoryFinding, MemoryRecallOptions, HistoryEntry, TrajectoryEntry } from "./memory.js";
export { webhook, type TriggerSource, type WebhookFilter, type WebhookValue } from "./triggers.js";
+export type { PluginMethod, PluginPrimitive } from './plugin-contract.js';
diff --git a/packages/surface/src/plugin-contract.ts b/packages/surface/src/plugin-contract.ts
new file mode 100644
index 000000000..8f1fc5ad7
--- /dev/null
+++ b/packages/surface/src/plugin-contract.ts
@@ -0,0 +1,5 @@
+import type { Step } from './step.js';
+
+/** Plugins augment Ctx in @relayflows/surface; no catch-all index signature. */
+export type PluginMethod = (args: Args) => Step