diff --git a/README.md b/README.md index cccf073af..1b31f12d3 100644 --- a/README.md +++ b/README.md @@ -41,12 +41,32 @@ can run autonomously over days and weeks. Every agent session is observable and # Get Started -Installation: -``` +Install the CLI, then the authoring package in your own project: +```sh npm install -g relayflows +mkdir my-flow && cd my-flow && npm install @relayflows/surface ``` -Give your agent a skill to write a flow: +Write a flow — save this as `hello.flow.ts`: +```ts +import { flow } from "@relayflows/surface"; + +export default flow("hello", async (f) => { + await f.run('echo "hello from a relayflow"'); + f.done("success"); +}); ``` + +Run it: +```sh +flows run hello.flow.ts --input '{}' +``` + +That's the whole loop — `flows run` spins up the local kernel itself on first use, no separate daemon step. You should see a completed run report. + +`f.run` and `f.agent` both actually dispatch today. `f.agent` runs a real coding-agent CLI the same way a declarative `type: agent` step does — it needs a `flows.json` in your project declaring which CLI to use (see `docs/SURFACE.md` §5 and `packages/sdk/src/cli/check.ts`'s `readProjectConfig`); without one, `flows run` refuses with a clear `agent_cli_unresolved` diagnostic rather than hanging. `f.llm`, `f.human`, `f.dispatch`, and `f.cloud` are still `docs/SURFACE.md`'s design surface, not yet runnable — see [`examples/`](examples/) for what the full shape looks like, and each example's own README for exactly what runs today versus what's still landing. + +Give your agent a skill to write a flow: +```sh npx skills add https://github.com/agentworkforce/skills --skill writing-relayflows ``` diff --git a/packages/sdk/src/authored-flow-error.ts b/packages/sdk/src/authored-flow-error.ts index 901e6c759..7aeaed8eb 100644 --- a/packages/sdk/src/authored-flow-error.ts +++ b/packages/sdk/src/authored-flow-error.ts @@ -4,6 +4,8 @@ import type { } from './protocol.js'; export type AuthoredFlowExecutionErrorCode = + | 'agent_cli_unresolved' + | 'agent_parked' | 'duplicate_completion' | 'journal_protocol_violation' | 'missing_completion' @@ -15,6 +17,7 @@ export type AuthoredFlowExecutionErrorCode = | 'unsupported_header' | 'unsettled_derived_work' | 'unsupported_promise_lifecycle' + | 'unsupported_workspace_permission' | 'unawaited_step' | 'unsupported_verb'; diff --git a/packages/sdk/src/authored-flow-executor.ts b/packages/sdk/src/authored-flow-executor.ts index bdc8b11df..4b7dd53f6 100644 --- a/packages/sdk/src/authored-flow-executor.ts +++ b/packages/sdk/src/authored-flow-executor.ts @@ -1,6 +1,7 @@ import { COMPLETION_REASONS, RUN_COMPLETION_REASONS, + type AgentOptions, type AgentResult, type CloudHelper, type CompletionReason as SurfaceCompletionReason, @@ -9,8 +10,13 @@ import { type Step, } from '@relayflows/surface'; import type { FlowHandle } from '@relayflows/surface/runtime'; +import { join } from 'node:path'; import { compileSpec, toKernelSpec } from './compile.js'; import { getAuthoredFlowDefinition } from './authored-flow.js'; +import type { GetFlowDefinition } from './authored-flow-loader.js'; +import { checkAuthoredFlow } from './cli/check.js'; +import { classifyOutcome, type RunLifecycleOptions } from './cli/run.js'; +import type { PreflightDiagnostic } from './preflight.js'; import { AuthoredFlowExecutionError, type AuthoredFlowExecutionErrorCode, @@ -27,7 +33,7 @@ import type { RunCompletionReason as ProtocolRunCompletionReason, RunOutcome, } from './protocol.js'; -import { SPEC_SCHEMA_VERSION } from './spec.js'; +import { SPEC_SCHEMA_VERSION, type FlowSpec } from './spec.js'; type Assert = T; type Equal = [A] extends [B] @@ -49,6 +55,15 @@ type EveryRunCompletionReasonIsAcceptedByDone = Assert< export { AuthoredFlowExecutionError, type AuthoredFlowExecutionErrorCode }; +/** + * Matches the `"path/glob: readonly"` / `"path/glob: readwrite"` shorthand + * shown in docs/SURFACE.md and the examples — the only shape a workspace + * string could plausibly declare a permission in. No parser anywhere in this + * package turns that annotation into a real restriction, so `lowerAgent` + * refuses rather than silently accepting and ignoring it. + */ +const WORKSPACE_PERMISSION_ANNOTATION = /:\s*(readonly|readwrite)\s*$/i; + export interface AuthoredFlowJournalStep { readonly id: string; readonly runId: string; @@ -79,12 +94,46 @@ type JournalStepUsesStepCompletionReason = Assert< * `JournalClient`; values are read back from `step.completed` journal entries. * Unsupported headers, verbs, gates, or completion lowering fail closed. */ +export interface ExecuteAuthoredFlowOptions { + /** + * Defaults to this package's own static import — correct for every + * existing (internal, single-instance) caller. A flow loaded from an + * external path via `loadAuthoredFlow` must pass ITS resolved + * `getDefinition` instead; see authored-flow-loader.ts's comment on why. + */ + readonly getDefinition?: GetFlowDefinition; + /** + * The flow file's own path — passed straight to `checkAuthoredFlow` + * (cli/check.ts) so `f.agent` resolves a CLI the same way a declarative + * `type: agent` step does: nearest `flows.json`, real auth/model probing, + * canonicalized path. `checkAuthoredFlow` always does `dirname()` on this, + * matching `checkFlow`'s real `flows check ` contract — pass a FILE + * path, not a directory, or the search starts one level too high. Defaults + * to a synthetic `flow.ts` under `process.cwd()` for exactly this reason: + * `process.cwd()` itself is a directory, and `dirname(process.cwd())` + * would search cwd's PARENT. + */ + readonly flowPath?: string; + /** Passed straight through to classifyOutcome (cli/run.ts) for f.agent's wait. */ + readonly signal?: RunLifecycleOptions['signal']; + readonly onWait?: RunLifecycleOptions['onWait']; +} + export async function executeAuthoredFlow( handle: FlowHandle, journal: JournalClient, input?: Input, + options: ExecuteAuthoredFlowOptions = {}, ): Promise { - const definition = getAuthoredFlowDefinition(handle); + const getDefinition = options.getDefinition ?? getAuthoredFlowDefinition; + const flowPath = options.flowPath ?? join(process.cwd(), 'flow.ts'); + // Named separately from `options` because `lowerAgent` below has its own, + // differently-typed `options: AgentOptions` parameter that shadows this one. + const waitOptions: RunLifecycleOptions = { + ...(options.signal !== undefined ? { signal: options.signal } : {}), + ...(options.onWait !== undefined ? { onWait: options.onWait } : {}), + }; + const definition = getDefinition(handle); const headerFields = Object.keys(definition.header); if (headerFields.length > 0) { throw new AuthoredFlowExecutionError( @@ -112,6 +161,97 @@ export async function executeAuthoredFlow( return readSuccessfulOutput(journal, outcome, id, journalSteps); }; + // `name` (the first f.agent argument, e.g. "fixer") is not wired to the + // kernel's `agent` field: that field selects a NAMED declaration from + // `FlowSpec.agents`, and this executor refuses every non-empty header + // (see the top of this function) — an authored flow has no way to declare + // one today. `name` is kept only for step-id readability; CLI selection + // goes through the project's flows.json default below, same as it does + // for a bare `type: agent` YAML step with no explicit `cli`. + const lowerAgent = async ( + id: string, + options: AgentOptions, + ): Promise => { + if (options.workspace !== undefined && WORKSPACE_PERMISSION_ANNOTATION.test(options.workspace)) { + throw new AuthoredFlowExecutionError( + 'unsupported_workspace_permission', + `flow "${definition.name}" step "${id}": workspace "${options.workspace}" declares a ` + + 'permission annotation ("...: readonly" / "...: readwrite"), but nothing enforces it — ' + + 'no parser anywhere in this package turns that annotation into a real restriction ' + + '(kernel/DAEMON-LIFECYCLE.md\'s permission model is untouched by f.agent). ' + + 'Silently accepting and ignoring it would let a flow believe a restriction is in effect ' + + "when it is not. Declare a bare surface name (no trailing \": readonly\"/\": readwrite\") " + + 'if you do not need enforcement, or use the declarative spec\'s `permissions` field, which is real.', + ); + } + const authoring: FlowSpec = { + version: SPEC_SCHEMA_VERSION, + name: `${definition.name}/${id}`, + steps: [{ + id, + type: 'agent', + instruction: options.task, + ...(options.workspace === undefined ? {} : { + surfaces: { workspace: [{ surface: options.workspace }] }, + }), + }], + }; + // The kernel never resolves a `cli` on its own — every declarative + // `flows run`/`flows check` binds it first via this exact function + // (cli/check.ts), searching for the nearest flows.json from `flowPath` + // and real-probing auth/model readiness. An authored agent step gets + // nothing for free just because it was declared in TS instead of YAML. + const { report, flow: resolved } = checkAuthoredFlow(authoring, flowPath); + if (!report.ok || resolved === undefined) { + const refusal = report.diagnostics.find( + (diagnostic): diagnostic is PreflightDiagnostic & { severity: 'refusal' } => + diagnostic.severity === 'refusal', + ); + throw new AuthoredFlowExecutionError( + 'agent_cli_unresolved', + refusal?.message + ?? `flow "${definition.name}" step "${id}": no CLI could be resolved for f.agent ` + + `(searched for flows.json from "${flowPath}")`, + ); + } + const spec = toKernelSpec(resolved); + const outcome = await journal.runStart(spec); + // Reuse the declarative CLI's own wait/classification (cli/run.ts) rather + // than a hand-rolled poll: `step.completed` and the run's own terminal + // state are appended as two SEPARATE actions (kernel/relayflowd-core/src/machine.rs + // completion_actions vs complete_run_actions), so a naive read right + // after runStart can race a real, valid completion — and a genuinely + // long-running agent has no reason to be bounded by anything other than + // its own worker's lease, which classifyOutcome already follows + // (renewing as the lease renews, per docs/SURFACE.md §5's WAITING + // [worker_lease] contract), never an unrelated fixed deadline. + const execution = await classifyOutcome(journal, 'run', outcome, report, '', waitOptions); + if (execution.exitCode === 3) { + const parked = execution.report.parkedStep; + throw new AuthoredFlowExecutionError( + 'agent_parked', + execution.report.diagnostics.at(-1)?.message + ?? `flow "${definition.name}" step "${id}" parked` + + (parked !== undefined ? ` (${parked.type})` : '') + + ': no worker is attached to run it.', + undefined, + outcome.run_id, + ); + } + if (execution.exitCode !== 0) { + const reason = execution.report.completionReason; + throw new AuthoredFlowExecutionError( + 'step_failed', + execution.report.diagnostics.at(-1)?.message + ?? `flow "${definition.name}" step "${id}" did not complete successfully ` + + `(status: ${execution.report.status ?? 'unknown'})`, + isSurfaceCompletionReason(reason) ? reason : undefined, + outcome.run_id, + ); + } + return readSuccessfulAgentOutput(journal, outcome.run_id, id, journalSteps); + }; + const context: Ctx = { run(command) { assertOperationAllowed('run', definition.name, requestedCompletion); @@ -134,13 +274,15 @@ export async function executeAuthoredFlow( lifecycle, )); }, - agent() { + agent(name, options) { assertOperationAllowed('agent', definition.name, requestedCompletion); + void name; // step-id readability only — see the comment on lowerAgent. const id = `agent-${nextStep++}`; - return trackStep(authoredSteps, unsupportedStep( + return trackStep(authoredSteps, new AuthoredFlowOperation( id, 'agent', () => assertOperationAllowed('agent', definition.name, requestedCompletion), + () => lowerAgent(id, options), lifecycle, )); }, @@ -286,42 +428,90 @@ function unsupportedCloud(assertOpen: () => void): CloudHelper { }) as CloudHelper; } -async function readSuccessfulOutput( +/** + * Shared by every `f.*` verb that lowers to one kernel step run in isolation: + * find its `step.completed` entry, record it, and refuse anything but a + * clean success before handing the raw `output` back for verb-specific + * extraction (a plain string for `f.run`, an `AgentResult` for `f.agent`). + * + * Callers are responsible for having already established that the RUN + * reached a terminal, successful state before calling this — `f.run`'s + * caller relies on `runStart`'s own immediate response (the kernel drives a + * deterministic step to completion inline, no race); `f.agent`'s caller + * relies on `classifyOutcome` (cli/run.ts) having already polled to a true + * terminal state. Given that, a single read from the start of this run's + * (small, single-step) journal is enough — no polling here, and no run + * outcome ever needs re-checking. + */ +async function readCompletedStepOutput( journal: JournalClient, - outcome: RunOutcome, + runId: string, stepId: string, journalSteps: AuthoredFlowJournalStep[], -): Promise { - const entries = (await journal.journalRead(outcome.run_id, 1)).entries; +): Promise { + const entries = (await journal.journalRead(runId, 1)).entries; const completed = entries.find((entry) => isStepCompleted(entry, stepId)); if (!isStepCompleted(completed, stepId)) { - throw protocolViolation(outcome.run_id, `journal has no step.completed for "${stepId}"`); + throw protocolViolation(runId, `journal has no step.completed for "${stepId}"`); } const reason = completed.payload.completionReason; - journalSteps.push(Object.freeze({ id: stepId, runId: outcome.run_id, completionReason: reason })); + journalSteps.push(Object.freeze({ id: stepId, runId, completionReason: reason })); if (reason !== 'success') { throw new AuthoredFlowExecutionError( 'step_failed', `journal step "${stepId}" completed with ${reason}`, reason, - outcome.run_id, - ); - } - if (outcome.status !== 'completed' || outcome.completion_reason !== 'success') { - throw protocolViolation( - outcome.run_id, - `successful step entry conflicts with run outcome ${outcome.status}/${String(outcome.completion_reason)}`, + runId, ); } + return completed.payload.output; +} - const output = completed.payload.output; +async function readSuccessfulOutput( + journal: JournalClient, + outcome: RunOutcome, + stepId: string, + journalSteps: AuthoredFlowJournalStep[], +): Promise { + const output = await readCompletedStepOutput(journal, outcome.run_id, stepId, journalSteps); if (!isRecord(output) || typeof output['stdout_tail'] !== 'string') { throw protocolViolation(outcome.run_id, `step "${stepId}" has no string stdout_tail`); } return output['stdout_tail']; } +/** + * The kernel journals an agent step's raw `CliResult` shape + * (`exit_code`/`stdout_tail`/`stderr_tail`) — same as a deterministic step — + * UNLESS the CLI's stdout parsed as JSON, in which case `output` is that + * parsed object directly (worker.ts: `parseJsonOutput(result.stdout_tail) ?? + * result`). Neither shape carries a real `artifacts` list today: nothing in + * the kernel or worker enumerates files an agent wrote, and `f.agent`'s + * `AgentOptions` has no `output` schema parameter for a CLI to target either + * (unlike the declarative `AgentStepSpec.output` field). `artifacts` is + * therefore always empty here — honest about what data exists, not a + * placeholder for something not yet wired. + */ +async function readSuccessfulAgentOutput( + journal: JournalClient, + runId: string, + stepId: string, + journalSteps: AuthoredFlowJournalStep[], +): Promise { + const output = await readCompletedStepOutput(journal, runId, stepId, journalSteps); + if (!isRecord(output)) { + throw protocolViolation(runId, `step "${stepId}" produced a non-object output`); + } + if (typeof output['stdout_tail'] === 'string') { + return { summary: output['stdout_tail'], artifacts: [] }; + } + // The CLI's stdout parsed as JSON, so `output` is that value, not a + // CliResult. Fall back to a stable, inspectable summary rather than + // refusing a run whose agent step genuinely succeeded. + return { summary: JSON.stringify(output), artifacts: [] }; +} + interface StepCompletedEntry { entry_type: 'step.completed'; step_id: string; diff --git a/packages/sdk/src/authored-flow-loader.ts b/packages/sdk/src/authored-flow-loader.ts index 41f08f3ad..dde721aee 100644 --- a/packages/sdk/src/authored-flow-loader.ts +++ b/packages/sdk/src/authored-flow-loader.ts @@ -1,7 +1,12 @@ import { accessSync, constants } from 'node:fs'; +import { createRequire } from 'node:module'; import { resolve } from 'node:path'; import { pathToFileURL } from 'node:url'; -import { getAuthoredFlowDefinition, type FlowHandle } from './authored-flow.js'; +import { + getAuthoredFlowDefinition, + type AuthoredFlowDefinition, + type FlowHandle, +} from './authored-flow.js'; export class AuthoredFlowLoadError extends Error { constructor(message: string) { @@ -10,8 +15,23 @@ export class AuthoredFlowLoadError extends Error { } } +/** Same signature as `getAuthoredFlowDefinition`, resolved from wherever a flow was loaded. */ +export type GetFlowDefinition = (handle: FlowHandle) => AuthoredFlowDefinition; + +export interface LoadedAuthoredFlow { + readonly handle: FlowHandle; + /** + * Bound to the SAME `@relayflows/surface` module instance the flow file + * itself imported `flow` from — see the comment on + * {@link resolveGetFlowDefinition}. Callers that go on to execute the flow + * (not just validate it) must keep using this, not the SDK's own static + * `getAuthoredFlowDefinition` import. + */ + readonly getDefinition: GetFlowDefinition; +} + /** Import and validate a direct-run module without executing its authored body. */ -export async function loadAuthoredFlow(path: string): Promise { +export async function loadAuthoredFlow(path: string): Promise { const absolutePath = resolve(path); try { accessSync(absolutePath, constants.R_OK); @@ -28,17 +48,70 @@ export async function loadAuthoredFlow(path: string): Promise { ); } + const getDefinition = await resolveGetFlowDefinition(absolutePath, path); const handle = authoredModule['default'] as FlowHandle; try { - getAuthoredFlowDefinition(handle); + getDefinition(handle); } catch (error) { throw new AuthoredFlowLoadError( `Flow "${path}" must default-export flow(...): ${errorMessage(error)}`, ); } - return handle; + return { handle, getDefinition }; +} + +/** + * `getFlowDefinition` recognizes a handle by object identity in a `WeakMap` + * scoped to whichever copy of `@relayflows/surface` created it — deliberately: + * that is what makes a handle unforgeable (surface/tests/flow.test.ts + * "refuses malformed and forged handles at the runtime boundary" locks this + * in, including against a well-known-symbol forgery, so this must not be + * "fixed" by switching that map to a symbol-tagged property instead). + * + * An author's `.flow.ts` almost never lives inside this monorepo. It + * resolves `@relayflows/surface` from ITS OWN `node_modules` — a physically + * different module instance than the one this SDK package statically + * imports for itself, so the SDK's own copy's `WeakMap` never has the + * entry the flow file's copy wrote. The fix is not a new identity + * mechanism; it is asking the SAME question Node itself would ask: resolve + * `@relayflows/surface/runtime` from the flow file's own location, exactly + * as the flow file's `import { flow } from "@relayflows/surface"` + * already did, and use THAT copy's `getFlowDefinition`. Since the flow + * module already imported successfully (by the time this runs), a + * compatible `@relayflows/surface` is provably resolvable from this same + * anchor. + */ +async function resolveGetFlowDefinition( + absolutePath: string, + displayPath: string, +): Promise { + const require = createRequire(pathToFileURL(absolutePath)); + let resolvedRuntimePath: string; + try { + resolvedRuntimePath = require.resolve('@relayflows/surface/runtime'); + } catch (error) { + throw new AuthoredFlowLoadError( + `Flow "${displayPath}" imports @relayflows/surface, but @relayflows/surface/runtime ` + + `could not be resolved from the same location: ${errorMessage(error)}`, + ); + } + const runtimeModule = await import(pathToFileURL(resolvedRuntimePath).href) as { + getFlowDefinition?: unknown; + }; + if (typeof runtimeModule.getFlowDefinition !== 'function') { + throw new AuthoredFlowLoadError( + `Flow "${displayPath}": the @relayflows/surface/runtime resolved from its location ` + + 'does not export getFlowDefinition — check its @relayflows/surface version.', + ); + } + return runtimeModule.getFlowDefinition as GetFlowDefinition; } +// Exported for callers (internal SDK tests, and any co-located flow that is +// provably the same `@relayflows/surface` instance as this package) that +// have no need to resolve a separate copy. +export { getAuthoredFlowDefinition }; + function errorMessage(error: unknown): string { return error instanceof Error ? error.message : 'unknown authored-flow error'; } diff --git a/packages/sdk/src/cli/direct-run.ts b/packages/sdk/src/cli/direct-run.ts index 056ec4919..a8bfc1bbd 100644 --- a/packages/sdk/src/cli/direct-run.ts +++ b/packages/sdk/src/cli/direct-run.ts @@ -44,8 +44,13 @@ export async function runDirectFlow( if (connected !== undefined) return connected; try { - const handle = await loadAuthoredFlow(path); - const result = await executeAuthoredFlow(handle, client, input); + const { handle, getDefinition } = await loadAuthoredFlow(path); + const result = await executeAuthoredFlow(handle, client, input, { + getDefinition, + flowPath: path, + ...(options.signal !== undefined ? { signal: options.signal } : {}), + ...(options.onWait !== undefined ? { onWait: options.onWait } : {}), + }); const terminal = result.journalSteps.at(-1); if (terminal === undefined) { return protocolFailure('run', base, socketPath, new Error( @@ -65,8 +70,21 @@ export async function runDirectFlow( }, }; } catch (error) { + // `agent_cli_unresolved` and `unsupported_workspace_permission` are + // preflight-shaped refusals, not protocol failures — `flows check` + // returns exit 2 for the equivalent declarative-spec failures, and this + // path should match it. Known gap, not solved here: if a `f.run` before + // the failing `f.agent` already journaled real work, this still reports + // as a clean refusal — true upfront preflight would need to know every + // `f.agent` call an imperative TS body will make before running any of + // it, which isn't knowable without running the body (see the comment on + // ExecuteAuthoredFlowOptions and this project's own examples/README for + // the same limitation already documented elsewhere). if (error instanceof AuthoredFlowLoadError - || (error instanceof AuthoredFlowExecutionError && error.code === 'unsupported_header')) { + || (error instanceof AuthoredFlowExecutionError + && (error.code === 'unsupported_header' + || error.code === 'agent_cli_unresolved' + || error.code === 'unsupported_workspace_permission'))) { return { exitCode: 2, report: { @@ -78,6 +96,23 @@ export async function runDirectFlow( }, }; } + if (error instanceof AuthoredFlowExecutionError && error.code === 'agent_parked') { + return { + exitCode: 3, + report: { + ...base, + ok: false, + runId: error.runId, + socketPath, + status: 'parked', + diagnostics: [...base.diagnostics, { + severity: 'parked', + kind: 'run_parked', + message: error.message, + }], + }, + }; + } const runId = error instanceof AuthoredFlowExecutionError ? error.runId : undefined; return protocolFailure('run', base, socketPath, error, runId); } finally { diff --git a/packages/sdk/tests/authored-flow.test.ts b/packages/sdk/tests/authored-flow.test.ts index 985ee93a6..0347e5882 100644 --- a/packages/sdk/tests/authored-flow.test.ts +++ b/packages/sdk/tests/authored-flow.test.ts @@ -146,6 +146,29 @@ describe('authored flow journal executor', () => { }), disconnectedJournal)).rejects.toMatchObject({ code: 'unsupported_gate' }); }); + it('refuses a workspace permission annotation f.agent cannot enforce, before contacting the journal', async () => { + const disconnectedJournal = new JournalClient('/journal-must-not-be-contacted'); + + for (const workspace of ['src/**: readonly', 'src/**: readwrite', 'src/**:readonly']) { + await expect(executeAuthoredFlow(flow('workspace-permission-not-enforced', async (f) => { + await f.agent('worker', { task: 'must not dispatch', workspace }); + f.done('success'); + }), disconnectedJournal)).rejects.toMatchObject({ + code: 'unsupported_workspace_permission', + }); + } + + // A bare surface name (no permission annotation) is unaffected — this + // suite's other f.agent cases already exercise the resolved path; this + // one only needs to prove the annotation check does not over-match. + await expect(executeAuthoredFlow(flow('workspace-bare-surface', async (f) => { + await f.agent('worker', { task: 'x', workspace: 'repo' }); + f.done('success'); + }), disconnectedJournal)).rejects.not.toMatchObject({ + code: 'unsupported_workspace_permission', + }); + }); + it('rejects invalid raw headers before the executor can contact the journal', async () => { const disconnectedJournal = new JournalClient('/journal-must-not-be-contacted'); @@ -283,14 +306,14 @@ describe('authored flow journal executor', () => { }), code: 'unsupported_verb', }, - { - handle: flow('manual-agent-chain', async (f) => { - f.agent('worker', { task: 'unsupported' }).then(undefined, () => undefined); - await new Promise((resolve) => setTimeout(resolve, 100)); - f.done('success'); - }), - code: 'unsupported_verb', - }, + // No f.agent case here: it now really dispatches (authored-flow-executor.ts + // lowers it to a real kernel AgentStepSpec through checkAuthoredFlow's + // preflight), so against this mock journal server — no project, no + // flows.json, no CLI to resolve — it fails at CLI resolution before + // the manual-chain detection this test exercises ever gets a chance + // to run. That's a different behavior than what this test is for; the + // `run`/`llm` cases above already cover manual-chain detection + // generalizing across verbs. ]; try { @@ -344,14 +367,11 @@ describe('authored flow journal executor', () => { }), code: 'unsupported_verb', }, - { - handle: flow('consumed-agent-rejection', async (f) => { - f.agent('worker', { task: 'unsupported' }).then(undefined, () => 'consumed'); - await new Promise((resolve) => setTimeout(resolve, 100)); - f.done('success'); - }), - code: 'unsupported_verb', - }, + // No f.agent case here — same reason as the identical array in + // "refuses manually chained work even when it settles before the body + // returns" above: f.agent now really dispatches, so against this mock + // journal server it fails at CLI resolution, not at the rejection + // this test is about. ]; try { diff --git a/packages/sdk/tests/live-kernel.test.ts b/packages/sdk/tests/live-kernel.test.ts index b633c29e9..54aa37b00 100644 --- a/packages/sdk/tests/live-kernel.test.ts +++ b/packages/sdk/tests/live-kernel.test.ts @@ -16,11 +16,13 @@ import { dirname, join, resolve } from 'node:path'; import { spawn, spawnSync, type ChildProcess } from 'node:child_process'; import { fileURLToPath } from 'node:url'; import { afterEach, beforeAll, describe, expect, it } from 'vitest'; +import { flow } from '@relayflows/surface'; import { compileYaml, toKernelSpec } from '../src/compile.js'; import { checkFlow } from '../src/cli/check.js'; import { JournalClient } from '../src/journal-client.js'; import type { StepDispatchEvent } from '../src/protocol.js'; import { AgentWorker } from '../src/worker.js'; +import { executeAuthoredFlow } from '../src/authored-flow-executor.js'; import { resolveSpecCliPaths } from '../src/cli/hn-monitor.js'; import { emitDueTicks, type TickCursor } from '../src/tick-source.js'; @@ -313,6 +315,142 @@ steps: await worker.close(); }); + it('f.agent lowers to a real agent step and dispatches through a live worker', async () => { + // The authored-TS-flow half of the same round trip the previous test + // proves for a declarative YAML spec: executeAuthoredFlow's f.agent + // lowering (authored-flow-executor.ts) must submit a real kernel + // AgentStepSpec, get it dispatched to a real attached worker, and map + // the worker's real completion back into a surface AgentResult. + const directory = temporaryDirectory('flows-live-authored-agent-'); + const dataDir = join(directory, 'data'); + const cli = join(directory, 'agent-cli'); + writeFileSync(cli, `#!/usr/bin/env node +// checkAuthoredFlow's preflight probes " auth status" for real before +// ever dispatching a step — this stub must answer it like a genuine adapter +// would, the same as the declarative-YAML sibling test's CLI is spared from +// needing to because that test calls client.runStart directly, bypassing +// preflight entirely. +if (process.argv[2] === 'auth' && process.argv[3] === 'status') process.exit(0); +if (process.argv[2] !== '--relayflows-adapter-v1') process.exit(9); +process.stdout.write('relayflows-agent-cli-v1\\n'); +let input = ''; +process.stdin.setEncoding('utf8'); +process.stdin.on('data', chunk => { input += chunk; }); +process.stdin.on('end', () => { + if (input.trim() === '') process.exit(0); + const request = JSON.parse(input); + process.stdout.write('relayflows-agent-cli-v1-execute\\n'); + process.stdout.write('handled: ' + request.instruction); +}); +`); + chmodSync(cli, 0o755); + // f.agent has no way to declare a CLI itself (AgentOptions is + // {task, workspace} only) — it resolves one exactly the way a bare + // `type: agent` YAML step with no explicit `cli` does: the nearest + // flows.json's project default, found searching upward from the flow's + // own path (checkAuthoredFlow, cli/check.ts). + writeFileSync(join(directory, 'flows.json'), JSON.stringify({ cli })); + await startDaemon(dataDir); + + const client = await connectClient(dataDir); + await client.hello('live-sdk-authored-agent-worker'); + const worker = new AgentWorker(client, { + workerId: 'live-sdk-authored-agent-worker', + pins: { + workspace: [{ surface: 'repo', revision_id: 'rev-a' }], + streams: [], + }, + }); + await worker.attach(); + + const runClient = await connectClient(dataDir); + await runClient.hello('live-sdk-authored-agent-run'); + let capturedResult: { summary: string; artifacts: string[] } | undefined; + const handle = flow('authored-agent', async (f) => { + capturedResult = await f.agent('worker', { + task: 'Perform the declared work.', + }); + f.done('success'); + }); + + let result: Awaited>; + try { + result = await executeAuthoredFlow(handle, runClient, undefined, { + flowPath: join(directory, 'authored-agent.flow.ts'), + }); + } finally { + // In a `finally`, not after: if executeAuthoredFlow throws mid-dispatch, + // the worker must still drain whatever it already started before this + // test tears down — worker.close()'s own contract (worker.ts) is to + // guarantee that. runClient isn't closed here; connectClient already + // registered it for the shared afterEach teardown above. + await worker.close(); + } + + expect(result.completionReason).toBe('success'); + expect(capturedResult?.summary).toBe('handled: Perform the declared work.'); + expect(capturedResult?.artifacts).toEqual([]); + }); + + it("f.agent's default flowPath anchors on cwd, not cwd's parent", async () => { + // checkAuthoredFlow (cli/check.ts) always does dirname() on the path it's + // given, matching flows check's real contract: a FILE path in, its + // directory searched. executeAuthoredFlow's default flowPath used to be + // bare `process.cwd()` — itself a directory — so dirname() searched cwd's + // PARENT, one level too high, missing a flows.json genuinely sitting in + // cwd. The fix is the synthetic `join(cwd, 'flow.ts')` default; this + // proves it by putting flows.json only in cwd, never in its parent. + const directory = temporaryDirectory('flows-live-defpath-'); + const dataDir = join(directory, 'data'); + const cli = join(directory, 'agent-cli'); + writeFileSync(cli, `#!/usr/bin/env node +if (process.argv[2] === 'auth' && process.argv[3] === 'status') process.exit(0); +if (process.argv[2] !== '--relayflows-adapter-v1') process.exit(9); +process.stdout.write('relayflows-agent-cli-v1\\n'); +let input = ''; +process.stdin.setEncoding('utf8'); +process.stdin.on('data', chunk => { input += chunk; }); +process.stdin.on('end', () => { + if (input.trim() === '') process.exit(0); + JSON.parse(input); + process.stdout.write('relayflows-agent-cli-v1-execute\\ndefault-flowpath-ok'); +}); +`); + chmodSync(cli, 0o755); + writeFileSync(join(directory, 'flows.json'), JSON.stringify({ cli })); + await startDaemon(dataDir); + + const client = await connectClient(dataDir); + await client.hello('live-default-flowpath-worker'); + const worker = new AgentWorker(client, { + workerId: 'live-default-flowpath-worker', + pins: { workspace: [{ surface: 'repo', revision_id: 'rev-a' }], streams: [] }, + }); + await worker.attach(); + + const runClient = await connectClient(dataDir); + await runClient.hello('live-default-flowpath-run'); + let capturedResult: { summary: string; artifacts: string[] } | undefined; + const handle = flow('default-flowpath', async (f) => { + capturedResult = await f.agent('worker', { task: 'x' }); + f.done('success'); + }); + + const previousCwd = process.cwd(); + process.chdir(directory); + let result: Awaited>; + try { + // No `flowPath` option — this is the exact default under test. + result = await executeAuthoredFlow(handle, runClient); + } finally { + process.chdir(previousCwd); + await worker.close(); + } + + expect(result.completionReason).toBe('success'); + expect(capturedResult?.summary).toBe('default-flowpath-ok'); + }); + it('can always get a parked run to a late-attaching worker', async () => { // The contract that cost the most time to establish, so it is pinned here. // diff --git a/packages/surface/package.json b/packages/surface/package.json index dee27925e..cff67463a 100644 --- a/packages/surface/package.json +++ b/packages/surface/package.json @@ -8,11 +8,13 @@ "exports": { ".": { "types": "./dist/index.d.ts", - "import": "./dist/index.js" + "import": "./dist/index.js", + "require": "./dist/index.js" }, "./runtime": { "types": "./dist/runtime.d.ts", - "import": "./dist/runtime.js" + "import": "./dist/runtime.js", + "require": "./dist/runtime.js" } }, "files": [ @@ -28,6 +30,9 @@ "test": "bun run build && tsc -p tsconfig.test.json && vitest run" }, "license": "Apache-2.0", + "engines": { + "node": ">=20.19.0 || >=22.12.0" + }, "devDependencies": { "typescript": "^5.6.0", "vitest": "^2.1.0"