From 41906cd74e59bfd40f7b87f6dd0cca52ce99bacc Mon Sep 17 00:00:00 2001 From: Miya Date: Fri, 11 Sep 2026 19:13:00 +0200 Subject: [PATCH 1/3] =?UTF-8?q?feat(sdk,surface):=20plugin=20registry=20?= =?UTF-8?q?=E2=80=94=20flows=20add=20+=20plugin=20contract=20(#305)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Session-Id: 01a09168-b666-7ae2-9f29-0ea10e48b894 Session-Id: 01a091dd-02a2-7820-8006-4430d2a5c76e Session-Id: efeda5df-9b7c-48d4-b2ce-957f5bef0a82 --- docs/SURFACE.md | 22 +++ docs/evidence/spec-J-plugin-registry.md | 139 ++++++++++++++++++ packages/sdk/src/authored-flow-executor.ts | 13 ++ packages/sdk/src/authored-plugin-effect.ts | 137 +++++++++++++++++ packages/sdk/src/cli.ts | 6 + packages/sdk/src/cli/add.ts | 45 ++++++ packages/sdk/src/cli/check-typescript.ts | 4 + packages/sdk/src/cli/check.ts | 7 +- packages/sdk/src/failure-kinds.ts | 2 + packages/sdk/src/plugin-loader.ts | 84 +++++++++++ packages/sdk/src/plugin-manifest.ts | 76 ++++++++++ packages/sdk/src/preflight.ts | 13 +- packages/sdk/tests/plugin-add.test.ts | 62 ++++++++ packages/sdk/tests/plugin-loader.test.ts | 69 +++++++++ packages/sdk/tests/preflight.test.ts | 34 +++++ packages/surface/src/index.ts | 1 + packages/surface/src/plugin-contract.ts | 5 + testdata/plugins/helper-broken/package.json | 1 + .../plugins/helper-datadog/flows-plugin.d.ts | 6 + .../plugins/helper-datadog/flows-plugin.json | 6 + testdata/plugins/helper-datadog/package.json | 1 + testdata/plugins/helper-datadog/src/index.js | 4 + .../helper-no-preflight/flows-plugin.json | 1 + .../plugins/helper-no-preflight/package.json | 1 + 24 files changed, 736 insertions(+), 3 deletions(-) create mode 100644 docs/evidence/spec-J-plugin-registry.md create mode 100644 packages/sdk/src/authored-plugin-effect.ts create mode 100644 packages/sdk/src/cli/add.ts create mode 100644 packages/sdk/src/plugin-loader.ts create mode 100644 packages/sdk/src/plugin-manifest.ts create mode 100644 packages/sdk/tests/plugin-add.test.ts create mode 100644 packages/sdk/tests/plugin-loader.test.ts create mode 100644 packages/surface/src/plugin-contract.ts create mode 100644 testdata/plugins/helper-broken/package.json create mode 100644 testdata/plugins/helper-datadog/flows-plugin.d.ts create mode 100644 testdata/plugins/helper-datadog/flows-plugin.json create mode 100644 testdata/plugins/helper-datadog/package.json create mode 100644 testdata/plugins/helper-datadog/src/index.js create mode 100644 testdata/plugins/helper-no-preflight/flows-plugin.json create mode 100644 testdata/plugins/helper-no-preflight/package.json 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..fff3fc48d --- /dev/null +++ b/packages/sdk/src/authored-plugin-effect.ts @@ -0,0 +1,137 @@ +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; + try { + await peer.connect(); + await peer.hello('flows-plugin'); + peer.on('step.dispatch', dispatch); + await peer.workerAttach(stream, ['agent'], { workspace: [], streams: [{ stream, read_offset: 0 }] }, 1); + 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 budget.execute(journal, spec, async outcome => { + timer = setTimeout(() => { + dispatchExpired = true; + failed(new Error('Plugin worker dispatch deadline exceeded')); + }, 30_000); + childRunId = outcome.run_id; + 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; + }); + } finally { + clearTimeout(timer); + 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 && childRunId !== undefined) { + try { await journal.runCancel(childRunId); } catch { /* fail-open on cleanup */ } + } + 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..5ec85a666 --- /dev/null +++ b/packages/sdk/src/plugin-loader.ts @@ -0,0 +1,84 @@ +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.add(pluginPackageName(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..11d60be71 --- /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_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/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..c1e93451f --- /dev/null +++ b/packages/sdk/tests/plugin-loader.test.ts @@ -0,0 +1,69 @@ +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('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..2eb3a13bc 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,35 @@ 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 }); } + } 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; +export type PluginPrimitive = 'run' | 'llm' | 'agent' | 'effect' | 'wait'; diff --git a/testdata/plugins/helper-broken/package.json b/testdata/plugins/helper-broken/package.json new file mode 100644 index 000000000..ca9d06bd8 --- /dev/null +++ b/testdata/plugins/helper-broken/package.json @@ -0,0 +1 @@ +{"name":"@flows/helper-broken","version":"0.1.0"} diff --git a/testdata/plugins/helper-datadog/flows-plugin.d.ts b/testdata/plugins/helper-datadog/flows-plugin.d.ts new file mode 100644 index 000000000..e2aedb232 --- /dev/null +++ b/testdata/plugins/helper-datadog/flows-plugin.d.ts @@ -0,0 +1,6 @@ +import type { Step } from '@relayflows/surface'; +declare module '@relayflows/surface' { + interface Ctx { + datadog: { query(args: { metric: string }): Step }; + } +} diff --git a/testdata/plugins/helper-datadog/flows-plugin.json b/testdata/plugins/helper-datadog/flows-plugin.json new file mode 100644 index 000000000..17a8b7874 --- /dev/null +++ b/testdata/plugins/helper-datadog/flows-plugin.json @@ -0,0 +1,6 @@ +{ + "name": "helper-datadog", "version": "0.1.0", + "verbs": [{ "namespace": "datadog", "method": "query", "lowersTo": "effect", "args": { "type": "object", "properties": { "metric": { "type": "string" } }, "required": ["metric"], "additionalProperties": false } }], + "triggers": [], "gates": [], + "preflight": { "credentials": ["DATADOG_API_KEY"], "servers": [] } +} diff --git a/testdata/plugins/helper-datadog/package.json b/testdata/plugins/helper-datadog/package.json new file mode 100644 index 000000000..298d662d1 --- /dev/null +++ b/testdata/plugins/helper-datadog/package.json @@ -0,0 +1 @@ +{"name":"@flows/helper-datadog","version":"0.1.0","type":"module","main":"src/index.js"} diff --git a/testdata/plugins/helper-datadog/src/index.js b/testdata/plugins/helper-datadog/src/index.js new file mode 100644 index 000000000..37c850504 --- /dev/null +++ b/testdata/plugins/helper-datadog/src/index.js @@ -0,0 +1,4 @@ +// Offline fixture: real providers must honor the kernel-issued idempotency key. +export async function execute(namespace, method, input, { idempotencyKey }) { + return { namespace, method, metric: input.metric, value: 42, idempotencyKey }; +} diff --git a/testdata/plugins/helper-no-preflight/flows-plugin.json b/testdata/plugins/helper-no-preflight/flows-plugin.json new file mode 100644 index 000000000..eceec9273 --- /dev/null +++ b/testdata/plugins/helper-no-preflight/flows-plugin.json @@ -0,0 +1 @@ +{"name":"helper-no-preflight","version":"0.1.0","verbs":[],"triggers":[],"gates":[]} diff --git a/testdata/plugins/helper-no-preflight/package.json b/testdata/plugins/helper-no-preflight/package.json new file mode 100644 index 000000000..7d3cb62ad --- /dev/null +++ b/testdata/plugins/helper-no-preflight/package.json @@ -0,0 +1 @@ +{"name":"@flows/helper-no-preflight","version":"0.1.0"} From f4cfc540930e18b89832b92039488af93a716b8d Mon Sep 17 00:00:00 2001 From: Miya Date: Fri, 11 Sep 2026 21:10:09 +0200 Subject: [PATCH 2/3] fix(sdk): require declared plugins and bound worker initialization Session-Id: 01a091dd-02a2-7820-8006-4430d2a5c76e Session-Id: efeda5df-9b7c-48d4-b2ce-957f5bef0a82 --- packages/sdk/src/authored-plugin-effect.ts | 39 ++++-- packages/sdk/src/plugin-loader.ts | 6 +- packages/sdk/src/plugin-manifest.ts | 2 +- .../sdk/tests/authored-plugin-effect.test.ts | 113 ++++++++++++++++++ packages/sdk/tests/plugin-loader.test.ts | 20 ++++ 5 files changed, 166 insertions(+), 14 deletions(-) create mode 100644 packages/sdk/tests/authored-plugin-effect.test.ts diff --git a/packages/sdk/src/authored-plugin-effect.ts b/packages/sdk/src/authored-plugin-effect.ts index fff3fc48d..186f1ed0e 100644 --- a/packages/sdk/src/authored-plugin-effect.ts +++ b/packages/sdk/src/authored-plugin-effect.ts @@ -89,21 +89,37 @@ export async function runPluginEffect( }); } 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 peer.connect(); - await peer.hello('flows-plugin'); + await Promise.race([peer.connect(), deadline]); + await Promise.race([peer.hello('flows-plugin'), deadline]); peer.on('step.dispatch', dispatch); - await peer.workerAttach(stream, ['agent'], { workspace: [], streams: [{ stream, read_offset: 0 }] }, 1); + 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 budget.execute(journal, spec, async outcome => { - timer = setTimeout(() => { - dispatchExpired = true; - failed(new Error('Plugin worker dispatch deadline exceeded')); - }, 30_000); + 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); } @@ -114,17 +130,16 @@ export async function runPluginEffect( } 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 && childRunId !== undefined) { - try { await journal.runCancel(childRunId); } catch { /* fail-open on cleanup */ } - } + if (dispatchExpired) await cancelChildRun(); peer.close(); await work?.catch(() => undefined); } diff --git a/packages/sdk/src/plugin-loader.ts b/packages/sdk/src/plugin-loader.ts index 5ec85a666..8911590ea 100644 --- a/packages/sdk/src/plugin-loader.ts +++ b/packages/sdk/src/plugin-loader.ts @@ -49,7 +49,11 @@ export async function loadPlugins(start: string): Promise((config.plugins as string[] | undefined)?.map(pluginPackageName)); - if (existsSync(scope)) for (const name of readdirSync(scope).sort()) if (name.startsWith('helper-')) names.add(pluginPackageName(name)); + 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) { diff --git a/packages/sdk/src/plugin-manifest.ts b/packages/sdk/src/plugin-manifest.ts index 11d60be71..ba133444a 100644 --- a/packages/sdk/src/plugin-manifest.ts +++ b/packages/sdk/src/plugin-manifest.ts @@ -2,7 +2,7 @@ import { Ajv } from 'ajv'; import { snapshotJsonValue } from './json-value.js'; export const PLUGIN_FAILURE_KINDS = [ - 'plugin_unknown', 'plugin_install_failed', 'plugin_manifest_missing', + '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; 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-loader.test.ts b/packages/sdk/tests/plugin-loader.test.ts index c1e93451f..daffe4765 100644 --- a/packages/sdk/tests/plugin-loader.test.ts +++ b/packages/sdk/tests/plugin-loader.test.ts @@ -23,6 +23,26 @@ function project() { 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'); From 56566471e322a5bcad223f38d9f301568ffab90f Mon Sep 17 00:00:00 2001 From: kjgbot Date: Fri, 11 Sep 2026 22:24:15 +0200 Subject: [PATCH 3/3] test(preflight): exercise plugin_unlisted so refusal-set parity holds @flows/helper-* packages installed in node_modules but absent from flows.json plugins now emit plugin_unlisted (plugin-loader.ts). The existing PREFLIGHT_FAILURE_KINDS walker did not have a scenario for this refusal, so PR#336's linux-x64-artifact failed on set parity. Adds a scoped fixture creating an unlisted helper directory and asserts the loader refuses it. Session-Id: efeda5df-9b7c-48d4-b2ce-957f5bef0a82 --- packages/sdk/tests/preflight.test.ts | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/packages/sdk/tests/preflight.test.ts b/packages/sdk/tests/preflight.test.ts index 2eb3a13bc..601b948d9 100644 --- a/packages/sdk/tests/preflight.test.ts +++ b/packages/sdk/tests/preflight.test.ts @@ -429,6 +429,21 @@ describe('preflight: CLI resolution and refusal predicates', () => { } } 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'); });