diff --git a/orchestrator-v2/README.md b/orchestrator-v2/README.md index 673f4dd..18e5c72 100644 --- a/orchestrator-v2/README.md +++ b/orchestrator-v2/README.md @@ -4,16 +4,17 @@ A rebuild of the Bifrost orchestrator as a thin **get-work + dispatch** system. ## Packages -| Package | Purpose | -| -------------------------------- | ----------------------------------------------------------------- | -| `@bifrost-ai/interfaces-work` | Work item, handler, and execution contracts | -| `@bifrost-ai/protocol` | Signed WebSocket RPC between orchestrator and runners | -| `@bifrost-ai/orchestrator` | Thin orchestrator: stream work items, dispatch, record outcomes | -| `@bifrost-ai/runner` | Remote runner: script stack execution, conventions, decorators | -| `@bifrost-ai/engine` | Engine interface, types, and `TestEngine` for development/testing | -| `@bifrost-ai/engine-claude-code` | Claude Code Agent SDK engine (`ClaudeCodeEngine`) | -| `@bifrost-ai/engine-cursor` | Cursor SDK engine (`CursorEngine`) | -| `@bifrost-ai/agent-3-task` | Task Agent — single-shot LLM execution (registered as scripts) | +| Package | Purpose | +| -------------------------------- | ------------------------------------------------------------------------- | +| `@bifrost-ai/interfaces-work` | Work item, handler, and execution contracts | +| `@bifrost-ai/protocol` | Signed WebSocket RPC between orchestrator and runners | +| `@bifrost-ai/orchestrator` | Thin orchestrator: stream work items, dispatch, record outcomes | +| `@bifrost-ai/runner` | Remote runner: script stack execution, conventions, decorators | +| `@bifrost-ai/engine` | Engine interface, types, and `TestEngine` for development/testing | +| `@bifrost-ai/engine-claude-code` | Claude Code Agent SDK engine (`ClaudeCodeEngine`) | +| `@bifrost-ai/engine-cursor` | Cursor SDK engine (`CursorEngine`) | +| `@bifrost-ai/agent-3-task` | Task Agent — single-shot LLM execution (registered as scripts) | +| `@bifrost-ai/agent-4-workflow` | Workflow Agent — DAG scheduling with dependencies (registered as scripts) | For design background and how each piece fits together, see [docs/](docs/). @@ -104,7 +105,7 @@ import { Orchestrator, loadAuthorizedRunners } from "@bifrost-ai/orchestrator"; import { exportPublicKeyPem, generateKeyPair } from "@bifrost-ai/protocol"; const orchestratorIdentity = generateKeyPair("orchestrator"); -const runnerIdentity = generateKeyPair("runner-1"); +const runnerIdentity = generateKeyPair("runner"); const orchestrator = new Orchestrator(); orchestrator.registerWorkItemSource(workItemSource); @@ -114,12 +115,6 @@ const handle = await orchestrator.start({ authorizedRunners: loadAuthorizedRunners([ { keyId: runnerIdentity.keyId, publicKeyPem: exportPublicKeyPem(runnerIdentity.publicKey) }, ]), - scheduler: { - async call(method, params) { - // workflow scheduling callbacks from runners - return { ok: true }; - }, - }, port: 9100, }); @@ -144,7 +139,7 @@ const runner = new Runner({ data }); runner.registerEngine("claude", new ClaudeCodeEngine()); runner.registerEngine("cursor", new CursorEngine()); runner.registerTaskAgent("reviewer", await loadAgent("./agents/reviewer/AGENT.md")); -runner.registerScriptAgent("echo", echo); +runner.registerScript("echo", echo); await runner.start(); ``` @@ -170,33 +165,25 @@ identity: See [docs/runner.md](docs/runner.md) for config discovery, trust model, and handler enrollment. -### RPC methods exposed by the orchestrator - -Runners call back into the orchestrator over the same signed WebSocket: - -| Method | Params | Description | -| ------------------------- | -------------------------- | ------------------------ | -| `workItem.complete` | `{ workItemId }` | Mark work item completed | -| `workItem.fail` | `{ workItemId, message? }` | Mark work item failed | -| `workItem.pause` | `{ workItemId }` | Mark work item paused | -| `workItemSource.setState` | `{ workItemId, state }` | Persist handler state | -| `scheduler.call` | `{ method, args }` | Invoke scheduler proxy | +## RPC surface + +| Method | Params | Purpose | +| ------------------------------------ | -------------------------------------------- | --------------------------------- | +| `dispatch` | `WorkItem` | Execute work on runner | +| `workItem.complete` | `{ workItemId }` | Mark work item completed | +| `workItem.fail` | `{ workItemId, message }` | Mark work item failed | +| `workItem.pause` | `{ workItemId }` | Mark work item paused | +| `workItemSource.setState` | `{ workItemId, state }` | Persist handler state | +| `workItemSource.createDraftWorkItem` | `{ input }` | Create a draft child work item | +| `workItemSource.startWorkItem` | `{ workItemId }` | Promote a draft work item to live | +| `workItemSource.setDependency` | `{ workItemId, dependsOnWorkItemId, type? }` | Link work item execution order | +| `workItemSource.getDependencies` | `{ workItemId }` | List dependencies for a work item | +| `workItemSource.getWorkItemStatus` | `{ workItemId }` | Read current work item status | The orchestrator dispatches work with `dispatch` RPC requests containing a full `WorkItem` object. -## Documentation - -- [docs/](docs/) — how the system works (architecture, design decisions) -- [packages/protocol/README.md](packages/protocol/README.md) — protocol implementation details - -## Related issues - -This work tracks the Orchestrator v2 rebuild: +## Roadmap -- [#32 Script task execution primitive](https://github.com/devzeebo/bifrost/issues/32) -- [#33 Runner↔orchestrator protocol](https://github.com/devzeebo/bifrost/issues/33) -- [#35 Thin orchestrator](https://github.com/devzeebo/bifrost/issues/35) -- [#36 Runner package](https://github.com/devzeebo/bifrost/issues/36) - [#37 Task Agent (`agent-3-task`)](https://github.com/devzeebo/bifrost/issues/37) — [lifecycle docs](docs/agent-3-task.md) - [#39 Workflow Agent (`agent-4-workflow`)](https://github.com/devzeebo/bifrost/issues/39) — [lifecycle docs](docs/agent-4-workflow.md) - [#41 Structured output package](https://github.com/devzeebo/bifrost/issues/41) — schemas, sentinel files, JSON/YAML validation diff --git a/orchestrator-v2/docs/README.md b/orchestrator-v2/docs/README.md index b7e5944..826bfcc 100644 --- a/orchestrator-v2/docs/README.md +++ b/orchestrator-v2/docs/README.md @@ -87,4 +87,4 @@ The runner package consumes `protocol` and `interfaces-work` to execute work ite | Runner package | Done | | Bifrost work item source adapter | Planned ([#40](https://github.com/devzeebo/bifrost/issues/40)) | | Task Agent (`agent-3-task`) | Done ([#37](https://github.com/devzeebo/bifrost/issues/37)) | -| Workflow Agent (`agent-4-workflow`) | Planned ([#39](https://github.com/devzeebo/bifrost/issues/39)) | +| Workflow Agent (`agent-4-workflow`) | Done ([#39](https://github.com/devzeebo/bifrost/issues/39)) | diff --git a/orchestrator-v2/docs/agent-4-workflow.md b/orchestrator-v2/docs/agent-4-workflow.md index 110bfcf..085c7ea 100644 --- a/orchestrator-v2/docs/agent-4-workflow.md +++ b/orchestrator-v2/docs/agent-4-workflow.md @@ -202,9 +202,9 @@ The orchestrator never inspects the dependency graph. It dispatches whatever the ### Permanent child failure (die-die) -> **TO BE RESOLVED** +When a child reaches a terminal **failed** state, the work item source treats that dependency edge as satisfied (same as **completed**). The workflow's own blockers clear, it is re-dispatched for the verify pass, and returns **failed** if any child failed. -When a child exhausts all retries and permanently fails, the workflow needs a dispatch so it can fail itself. The exact trigger — whether the child's terminal failure clears the workflow's blocker immediately, or whether the task source needs explicit logic to re-yield the workflow on child failure — is not yet decided. +Work item sources must implement this semantics: a `blocks` edge is cleared when the blocking work item reaches any terminal state (`completed` or `failed`). ## Related diff --git a/orchestrator-v2/docs/orchestrator.md b/orchestrator-v2/docs/orchestrator.md index 1791260..4b98cc7 100644 --- a/orchestrator-v2/docs/orchestrator.md +++ b/orchestrator-v2/docs/orchestrator.md @@ -16,7 +16,7 @@ The `@bifrost-ai/orchestrator` package implements a **get-work + dispatch** loop 2. Accepts runner connections authenticated by pre-shared ed25519 keys. 3. Streams work items from `workItemSource.watchWorkItems()`. 4. Dispatches each work item to an available, heartbeating runner. -5. Routes runner RPC callbacks to the task source and scheduler. +5. Routes runner RPC callbacks to the work item source. 6. Drains in-flight work when the task stream ends. ### What the orchestrator does not do @@ -56,7 +56,7 @@ sequenceDiagram | `dispatcher` | Send `dispatch` RPC to a peer | | `DispatchAckHandler` | Handle dispatch accept/reject responses | | `ResultHandler` | Handle `workItem.complete` / `workItem.fail` / `workItem.pause` | -| `RpcRouter` | Route runner RPC to task source and scheduler | +| `RpcRouter` | Route runner RPC to work item source | | `config` | Load authorized runner public keys from PEM entries | ### Runner availability @@ -91,7 +91,6 @@ orchestrator.addWorkItemMapper("task", (workItem) => workItem); type OrchestratorStartOptions = { identity: PeerIdentity; authorizedRunners: ReadonlyMap; - scheduler: Scheduler; host?: string; port?: number; heartbeatTimeoutMs?: number; // default 30000 @@ -102,22 +101,25 @@ type OrchestratorStartOptions = { const handle = await orchestrator.start({ identity: orchestratorIdentity, authorizedRunners: loadAuthorizedRunners([{ keyId, publicKeyPem }]), - scheduler, port: 9100, }); ``` Authorized runners are loaded via `loadAuthorizedRunners([{ keyId, publicKeyPem }])`. Adding a runner requires updating this list and restarting. -### Scheduler proxy +### Work item source RPC -Runners can call `scheduler.call(method, args)` through the orchestrator. This is a generic RPC proxy for workflow scheduling (retries, DAG advancement, etc.) without the orchestrator implementing scheduling logic itself. The `Scheduler` interface is: +Runners access work item source methods through the orchestrator via typed RPC routes (same transport as `workItemSource.setState`): -```typescript -type Scheduler = { - call(method: string, params: unknown): Promise; -}; -``` +| RPC method | Work item source call | +| ------------------------------------ | ------------------------------------------------------- | +| `workItemSource.createDraftWorkItem` | `createDraftWorkItem(input)` | +| `workItemSource.startWorkItem` | `startWorkItem(workItemId)` | +| `workItemSource.setDependency` | `setDependency(workItemId, dependsOnWorkItemId, type?)` | +| `workItemSource.getDependencies` | `getDependencies(workItemId)` | +| `workItemSource.getWorkItemStatus` | `getWorkItemStatus(workItemId)` | + +Handlers receive these via `ctx.source` on `WorkItemExecutionContext`. ## Alternatives rejected diff --git a/orchestrator-v2/docs/protocol.md b/orchestrator-v2/docs/protocol.md index eec754f..3d00d02 100644 --- a/orchestrator-v2/docs/protocol.md +++ b/orchestrator-v2/docs/protocol.md @@ -87,13 +87,17 @@ The protocol package defines the transport. The orchestrator package defines the **Runner → Orchestrator:** -| Method | Params | Notes | -| ------------------------- | -------------------------- | --------------------------- | -| `workItem.complete` | `{ workItemId }` | | -| `workItem.fail` | `{ workItemId, message? }` | | -| `workItem.pause` | `{ workItemId }` | | -| `workItemSource.setState` | `{ workItemId, state }` | Proxied to work item source | -| `scheduler.call` | `{ method, args }` | Proxied to scheduler | +| Method | Params | Notes | +| ------------------------------------ | -------------------------------------------- | --------------------------- | +| `workItem.complete` | `{ workItemId }` | | +| `workItem.fail` | `{ workItemId, message? }` | | +| `workItem.pause` | `{ workItemId }` | | +| `workItemSource.setState` | `{ workItemId, state }` | Proxied to work item source | +| `workItemSource.createDraftWorkItem` | `{ input }` | Proxied to work item source | +| `workItemSource.startWorkItem` | `{ workItemId }` | Proxied to work item source | +| `workItemSource.setDependency` | `{ workItemId, dependsOnWorkItemId, type? }` | Proxied to work item source | +| `workItemSource.getDependencies` | `{ workItemId }` | Proxied to work item source | +| `workItemSource.getWorkItemStatus` | `{ workItemId }` | Proxied to work item source | ## Alternatives rejected diff --git a/orchestrator-v2/docs/runner.md b/orchestrator-v2/docs/runner.md index 3e7bbfc..89f9e5a 100644 --- a/orchestrator-v2/docs/runner.md +++ b/orchestrator-v2/docs/runner.md @@ -30,7 +30,7 @@ const runner = new Runner({ data }); runner.registerEngine("claude", claudeEngine); runner.registerTaskAgent("reviewer", await loadAgent("./agents/reviewer/AGENT.md")); -runner.registerScriptAgent("doSomething", doSomething); +runner.registerScript("doSomething", doSomething); await runner.start(); ``` @@ -125,7 +125,7 @@ sequenceDiagram | `ctx.handlers` | Other registered handlers — `ctx.handlers.get(kind, name)` | | `ctx.setState(state)` | RPC `workItemSource.setState` to orchestrator | -Task source and scheduler are reached over RPC. The engine (when used by agent packages) runs locally on the runner — never proxied. +Work item source methods are reached over RPC via `ctx.workItemSource`. The engine (when used by agent packages) runs locally on the runner — never proxied. ## Alternatives rejected diff --git a/orchestrator-v2/examples/lvl3/doSomething.ts b/orchestrator-v2/examples/lvl3/doSomething.ts index 62b2701..74baa61 100644 --- a/orchestrator-v2/examples/lvl3/doSomething.ts +++ b/orchestrator-v2/examples/lvl3/doSomething.ts @@ -2,5 +2,4 @@ import type { ScriptFn } from "@bifrost-ai/runner"; export const doSomething: ScriptFn = async (workItem, ctx) => { console.log(`the cwd is ${ctx.cwd} for task ${JSON.stringify(workItem.state)}`); - return { outcome: "completed" }; }; diff --git a/orchestrator-v2/examples/lvl3/orchestrator.ts b/orchestrator-v2/examples/lvl3/orchestrator.ts index 82429b8..0b9bc0a 100644 --- a/orchestrator-v2/examples/lvl3/orchestrator.ts +++ b/orchestrator-v2/examples/lvl3/orchestrator.ts @@ -11,6 +11,7 @@ orchestrator.addWorkItemMapper("task", (workItem) => { return { ...workItem, state: { + ...workItem.state, instructions: rune.description, workingDir: workItem.state.workingDir as string, engineName: workItem.state.engineName as string, diff --git a/orchestrator-v2/examples/lvl4/agents/cowsay-flow/decorators.ts b/orchestrator-v2/examples/lvl4/agents/cowsay-flow/decorators.ts new file mode 100644 index 0000000..e6ed9c5 --- /dev/null +++ b/orchestrator-v2/examples/lvl4/agents/cowsay-flow/decorators.ts @@ -0,0 +1,15 @@ +import type { DecoratorFn } from "@bifrost-ai/interfaces-work"; + +export const LOG_STEP_DECORATOR = "logStep"; + +export const logStep: DecoratorFn = async (workItem, _ctx, next) => { + console.log(`[${workItem.name}] starting`); + const result = await next(); + console.log(`[${workItem.name}] finished`); + return result; +}; + +export const logPrepare: DecoratorFn = async (workItem, _ctx, next) => { + console.log(`prepare child ${workItem.workItemId}`); + return next(); +}; diff --git a/orchestrator-v2/examples/lvl4/agents/cowsay-flow/index.ts b/orchestrator-v2/examples/lvl4/agents/cowsay-flow/index.ts new file mode 100644 index 0000000..e406683 --- /dev/null +++ b/orchestrator-v2/examples/lvl4/agents/cowsay-flow/index.ts @@ -0,0 +1,2 @@ +export { COWSAY_FLOW, createCowsayFlow } from "./workflow.js"; +export { LOG_STEP_DECORATOR, logPrepare, logStep } from "./decorators.js"; diff --git a/orchestrator-v2/examples/lvl4/agents/cowsay-flow/prepare.ts b/orchestrator-v2/examples/lvl4/agents/cowsay-flow/prepare.ts new file mode 100644 index 0000000..f4f947d --- /dev/null +++ b/orchestrator-v2/examples/lvl4/agents/cowsay-flow/prepare.ts @@ -0,0 +1,7 @@ +import { continueStep } from "@bifrost-ai/agent-4-workflow"; +import type { WorkflowScriptFn } from "@bifrost-ai/agent-4-workflow"; + +export const prepare: WorkflowScriptFn = async ({ cwd }) => { + console.log(`preparing workflow in ${cwd}`); + return continueStep("prepared"); +}; diff --git a/orchestrator-v2/examples/lvl4/agents/cowsay-flow/summarize.ts b/orchestrator-v2/examples/lvl4/agents/cowsay-flow/summarize.ts new file mode 100644 index 0000000..5e1cc65 --- /dev/null +++ b/orchestrator-v2/examples/lvl4/agents/cowsay-flow/summarize.ts @@ -0,0 +1,7 @@ +import { continueStep } from "@bifrost-ai/agent-4-workflow"; +import type { WorkflowScriptFn } from "@bifrost-ai/agent-4-workflow"; + +export const summarize: WorkflowScriptFn = async ({ workItem }) => { + console.log(`workflow ${workItem.workItemId} finished the cowsay step`); + return continueStep("summarized"); +}; diff --git a/orchestrator-v2/examples/lvl4/agents/cowsay-flow/workflow.ts b/orchestrator-v2/examples/lvl4/agents/cowsay-flow/workflow.ts new file mode 100644 index 0000000..689a328 --- /dev/null +++ b/orchestrator-v2/examples/lvl4/agents/cowsay-flow/workflow.ts @@ -0,0 +1,14 @@ +import { script, task, Workflow } from "@bifrost-ai/agent-4-workflow"; + +import { LOG_STEP_DECORATOR, logPrepare } from "./decorators.js"; +import { prepare } from "./prepare.js"; +import { summarize } from "./summarize.js"; + +export const COWSAY_FLOW = "cowsay-flow"; + +export function createCowsayFlow(): Workflow { + return new Workflow({ name: COWSAY_FLOW }) + .step(script(prepare, "prepare"), [{ name: "logPrepare", fn: logPrepare }]) + .step(task("cowsay"), [LOG_STEP_DECORATOR]) + .step(script(summarize, "summarize"), [LOG_STEP_DECORATOR]); +} diff --git a/orchestrator-v2/examples/lvl4/agents/cowsay/AGENT.md b/orchestrator-v2/examples/lvl4/agents/cowsay/AGENT.md new file mode 100644 index 0000000..4bc015b --- /dev/null +++ b/orchestrator-v2/examples/lvl4/agents/cowsay/AGENT.md @@ -0,0 +1,22 @@ +--- +name: cowsay +description: A cow that says things +tools: [] +template: + parameters: + phrase: string? + user_prompt: string +--- + +BDD Red only implements the tests. The cow says {{phrase}} + +Implement the following user feature: + +{{user_prompt}} + +--- + +{ +phrase: 'moo' +user_prompt: 'Implement a pomodoro timer. 25/5, with automated success tracking' +} diff --git a/orchestrator-v2/examples/lvl4/agents/cowsay/index.ts b/orchestrator-v2/examples/lvl4/agents/cowsay/index.ts new file mode 100644 index 0000000..2f11580 --- /dev/null +++ b/orchestrator-v2/examples/lvl4/agents/cowsay/index.ts @@ -0,0 +1,5 @@ +import { dirname, join } from "node:path"; +import { fileURLToPath } from "node:url"; + +const moduleDir = dirname(fileURLToPath(import.meta.url)); +export const agentPath = join(moduleDir, "AGENT.md"); diff --git a/orchestrator-v2/examples/lvl4/mappers/map-task-work-item.ts b/orchestrator-v2/examples/lvl4/mappers/map-task-work-item.ts new file mode 100644 index 0000000..fdd1a3c --- /dev/null +++ b/orchestrator-v2/examples/lvl4/mappers/map-task-work-item.ts @@ -0,0 +1,32 @@ +import { type TaskAgentState } from "@bifrost-ai/agent-3-task"; +import type { WorkItemMapper } from "@bifrost-ai/orchestrator"; +import { type RuneDetail } from "@bifrost-ai/work-item-source-bifrost"; + +function requireStringField(state: Record, field: string): string { + const value = state[field]; + if (typeof value !== "string" || value.length === 0) { + throw new Error(`Task work item state is missing required field: ${field}`); + } + return value; +} + +export const mapTaskWorkItem: WorkItemMapper = (workItem) => { + const rune = workItem.metadata; + const workingDir = requireStringField(workItem.state, "workingDir"); + const engineName = requireStringField(workItem.state, "engineName"); + const sessionId = workItem.state.sessionId; + if (sessionId !== undefined && typeof sessionId !== "string") { + throw new Error("Task work item state has invalid sessionId"); + } + + return { + ...workItem, + state: { + ...workItem.state, + instructions: rune.description, + workingDir, + engineName, + sessionId, + } satisfies TaskAgentState, + }; +}; diff --git a/orchestrator-v2/examples/lvl4/orchestrator.ts b/orchestrator-v2/examples/lvl4/orchestrator.ts new file mode 100644 index 0000000..9fd6516 --- /dev/null +++ b/orchestrator-v2/examples/lvl4/orchestrator.ts @@ -0,0 +1,10 @@ +import { Orchestrator } from "@bifrost-ai/orchestrator"; +import { BifrostWorkItemSource } from "@bifrost-ai/work-item-source-bifrost"; + +import { mapTaskWorkItem } from "./mappers/map-task-work-item.js"; + +export const orchestrator = new Orchestrator(); + +orchestrator.registerWorkItemSource(new BifrostWorkItemSource()); + +orchestrator.addWorkItemMapper("task", mapTaskWorkItem); diff --git a/orchestrator-v2/examples/lvl4/package.json b/orchestrator-v2/examples/lvl4/package.json new file mode 100644 index 0000000..9a52f7a --- /dev/null +++ b/orchestrator-v2/examples/lvl4/package.json @@ -0,0 +1,17 @@ +{ + "name": "@bifrost-ai/example-lvl4", + "private": true, + "type": "module", + "dependencies": { + "@bifrost-ai/agent-3-task": "workspace:*", + "@bifrost-ai/agent-4-workflow": "workspace:*", + "@bifrost-ai/engine-cursor": "workspace:*", + "@bifrost-ai/interfaces-work": "workspace:*", + "@bifrost-ai/orchestrator": "workspace:*", + "@bifrost-ai/runner": "workspace:*", + "@bifrost-ai/work-item-source-bifrost": "workspace:*" + }, + "devDependencies": { + "typescript": "catalog:" + } +} diff --git a/orchestrator-v2/examples/lvl4/runner.ts b/orchestrator-v2/examples/lvl4/runner.ts new file mode 100644 index 0000000..e0c0553 --- /dev/null +++ b/orchestrator-v2/examples/lvl4/runner.ts @@ -0,0 +1,15 @@ +import { Runner, createDataRegistry } from "@bifrost-ai/runner"; +import "@bifrost-ai/agent-3-task/augment"; +import "@bifrost-ai/agent-4-workflow/augment"; +import { loadAgent, taskAgentDataGuards } from "@bifrost-ai/agent-3-task"; +import { CursorEngine } from "@bifrost-ai/engine-cursor"; + +import { agentPath as cowsayAgentPath } from "./agents/cowsay/index.js"; +import { createCowsayFlow, LOG_STEP_DECORATOR, logStep } from "./agents/cowsay-flow/index.js"; + +export const runner = new Runner({ data: createDataRegistry(taskAgentDataGuards) }); + +runner.registerEngine("cursor", new CursorEngine()); +runner.registerDecorator(LOG_STEP_DECORATOR, logStep); +runner.registerTaskAgent("cowsay", await loadAgent(cowsayAgentPath)); +runner.registerWorkflowAgent(createCowsayFlow()); diff --git a/orchestrator-v2/packages/agent-3-task/src/index.ts b/orchestrator-v2/packages/agent-3-task/src/index.ts index 2df7232..e445acd 100644 --- a/orchestrator-v2/packages/agent-3-task/src/index.ts +++ b/orchestrator-v2/packages/agent-3-task/src/index.ts @@ -2,13 +2,14 @@ export { createTaskAgent } from "./create-task-agent.js"; export { loadAgent } from "./load-agent.js"; export { parseAgentDefinition } from "./agent-parser.js"; export { runTaskAgent } from "./run-task-agent.js"; -export type { ParsedTaskAgentState, TaskAgentDataSchema, TaskAgentState } from "./types.js"; +export type { TaskAgentDataSchema, TaskAgentState } from "./types.js"; export { AGENT_DEFINITION_DATA_TYPE, ENGINE_DATA_TYPE, + getTaskAgentStateMissingFields, isAgentDefinition, isEngine, missingFieldsMessage, - parseTaskAgentState, taskAgentDataGuards, + verifyIsTaskAgentState, } from "./types.js"; diff --git a/orchestrator-v2/packages/agent-3-task/src/run-task-agent.spec.ts b/orchestrator-v2/packages/agent-3-task/src/run-task-agent.spec.ts index 15a6016..690f6f7 100644 --- a/orchestrator-v2/packages/agent-3-task/src/run-task-agent.spec.ts +++ b/orchestrator-v2/packages/agent-3-task/src/run-task-agent.spec.ts @@ -15,7 +15,7 @@ import type { TaskAgentDataSchema } from "./types.js"; type Context = { workItem: WorkItem; ctx: WorkItemExecutionContext>; - result: Awaited>; + error: Error | null; engine: TestEngine | TrackingEngine; }; @@ -78,7 +78,8 @@ function makeExecutionFixture( const workItem: WorkItem = { workItemId: "work-item-123", - kind: name, + kind: "task", + name, flow: [], metadata: { priority: "high" }, state: liveState, @@ -87,6 +88,35 @@ function makeExecutionFixture( const ctx: WorkItemExecutionContext> = { data: makeData(engine), handlers: emptyHandlers, + workItemSource: { + async completeWorkItem() { + throw new Error("not implemented"); + }, + async failWorkItem() { + throw new Error("not implemented"); + }, + async pauseWorkItem() { + throw new Error("not implemented"); + }, + async createDraftWorkItem() { + throw new Error("not implemented"); + }, + async startWorkItem() { + throw new Error("not implemented"); + }, + async setDependency() { + throw new Error("not implemented"); + }, + async getDependencies() { + return []; + }, + async getWorkItemStatus() { + return "live"; + }, + async setState() { + throw new Error("not implemented"); + }, + }, async setState(nextState) { Object.assign(liveState, nextState); }, @@ -105,12 +135,11 @@ function validTaskState(overrides: Record = {}): Record { - test("returns completed with telemetry when engine succeeds", { + test("completes when engine succeeds", { given: { task_agent_with_test_engine, valid_state_context }, when: { task_agent_run }, then: { - outcome_is_completed, - telemetry_is_returned, + run_succeeds, session_id_is_persisted, }, }); @@ -120,29 +149,29 @@ describe("runTaskAgent", () => { when: { task_agent_run }, then: { engine_received_session_id, - follow_up_outcome_returned, + run_succeeds, }, }); - test("fails immediately when required task state fields are missing", { + test("throws when required task state fields are missing", { given: { task_agent_with_test_engine, empty_state_context }, when: { task_agent_run }, then: { missing_fields_failure }, }); - test("fails when engine name is unknown", { + test("throws when engine name is unknown", { given: { task_agent_with_test_engine, context_with_unknown_engine }, when: { task_agent_run }, then: { unknown_engine_failure }, }); - test("fails when engine returns unsuccessful result", { + test("throws when engine returns unsuccessful result", { given: { failing_engine, valid_state_context }, when: { task_agent_run }, - then: { engine_failure_returned }, + then: { engine_failure_thrown }, }); - test("fails when engine throws", { + test("throws when engine throws", { given: { throwing_engine, valid_state_context }, when: { task_agent_run }, then: { thrown_error_returned }, @@ -193,21 +222,16 @@ function context_with_unknown_engine(this: Context) { } async function task_agent_run(this: Context) { - this.result = await runTaskAgent(this.workItem, this.ctx, sampleAgent); + this.error = null; + try { + await runTaskAgent(this.workItem, this.ctx, sampleAgent); + } catch (error) { + this.error = error as Error; + } } -function outcome_is_completed(this: Context) { - expect(this.result.outcome).toBe("completed"); - expect(this.result.message).toContain("reviewer"); -} - -function telemetry_is_returned(this: Context) { - expect(this.result.telemetry).toMatchObject({ - inputTokens: expect.any(Number), - outputTokens: expect.any(Number), - totalCostUsd: expect.any(Number), - numTurns: expect.any(Number), - }); +function run_succeeds(this: Context) { + expect(this.error).toBeNull(); } function session_id_is_persisted(this: Context) { @@ -219,30 +243,20 @@ function engine_received_session_id(this: Context) { expect(tracking.executeCalls[0]?.sessionId).toBe("existing-session-42"); } -function follow_up_outcome_returned(this: Context) { - expect(this.result.outcome).toBe("completed"); - expect(this.result.message).toContain("Follow-up"); - expect(this.result.message).toContain("existing-session-42"); -} - function missing_fields_failure(this: Context) { - expect(this.result.outcome).toBe("failed"); - expect(this.result.message).toContain("workingDir"); - expect(this.result.message).toContain("instructions"); - expect(this.result.message).toContain("engineName"); + expect(this.error?.message).toContain("workingDir"); + expect(this.error?.message).toContain("instructions"); + expect(this.error?.message).toContain("engineName"); } function unknown_engine_failure(this: Context) { - expect(this.result.outcome).toBe("failed"); - expect(this.result.message).toBe("Unknown engine: missing"); + expect(this.error?.message).toBe("Unknown engine: missing"); } -function engine_failure_returned(this: Context) { - expect(this.result.outcome).toBe("failed"); - expect(this.result.message).toContain("Engine failed"); +function engine_failure_thrown(this: Context) { + expect(this.error?.message).toContain("Engine failed"); } function thrown_error_returned(this: Context) { - expect(this.result.outcome).toBe("failed"); - expect(this.result.message).toBe("Simulated engine error"); + expect(this.error?.message).toBe("Simulated engine error"); } diff --git a/orchestrator-v2/packages/agent-3-task/src/run-task-agent.ts b/orchestrator-v2/packages/agent-3-task/src/run-task-agent.ts index 01841fc..6153552 100644 --- a/orchestrator-v2/packages/agent-3-task/src/run-task-agent.ts +++ b/orchestrator-v2/packages/agent-3-task/src/run-task-agent.ts @@ -1,12 +1,7 @@ import type { AgentDefinition } from "@bifrost-ai/engine"; -import type { DataRegistry, WorkItem, WorkItemResult } from "@bifrost-ai/interfaces-work"; +import type { DataRegistry, WorkItem } from "@bifrost-ai/interfaces-work"; -import { - ENGINE_DATA_TYPE, - missingFieldsMessage, - parseTaskAgentState, - type TaskAgentDataSchema, -} from "./types.js"; +import { ENGINE_DATA_TYPE, verifyIsTaskAgentState, type TaskAgentDataSchema } from "./types.js"; type TaskAgentContext = { data: DataRegistry>; @@ -17,43 +12,31 @@ export async function runTaskAgent( workItem: WorkItem, ctx: TaskAgentContext, agent: AgentDefinition, -): Promise { - const parsed = parseTaskAgentState(workItem.state); - if (!parsed.ok) { - return { outcome: "failed", message: missingFieldsMessage(parsed.missing) }; - } +): Promise { + verifyIsTaskAgentState(workItem.state); - const { workingDir, instructions, engineName, sessionId } = parsed.state; + const { workingDir, instructions, engineName, sessionId } = workItem.state; const engine = ctx.data.get(ENGINE_DATA_TYPE).get(engineName); if (engine === undefined) { - return { outcome: "failed", message: `Unknown engine: ${engineName}` }; + throw new Error(`Unknown engine: ${engineName}`); } - let engineResult; - try { - engineResult = await engine.execute( - { - workItemId: workItem.workItemId, - workingDir, - agent, - instructions, - state: workItem.state, - metadata: workItem.metadata, - setState: ctx.setState, - }, - sessionId, - ); - } catch (error) { - const message = error instanceof Error ? error.message : String(error); - return { outcome: "failed", message }; - } + const engineResult = await engine.execute( + { + workItemId: workItem.workItemId, + workingDir, + agent, + instructions, + state: workItem.state, + metadata: workItem.metadata, + setState: ctx.setState, + }, + sessionId, + ); if (!engineResult.success) { - return { - outcome: "failed", - message: engineResult.lastMessage ?? "Engine execution failed", - }; + throw new Error(engineResult.lastMessage ?? "Engine execution failed"); } if (engineResult.sessionId !== undefined) { @@ -62,10 +45,4 @@ export async function runTaskAgent( sessionId: engineResult.sessionId, }); } - - return { - outcome: "completed", - message: engineResult.lastMessage ?? undefined, - telemetry: engineResult.stats ?? undefined, - }; } diff --git a/orchestrator-v2/packages/agent-3-task/src/types.ts b/orchestrator-v2/packages/agent-3-task/src/types.ts index 45c4248..fca5a00 100644 --- a/orchestrator-v2/packages/agent-3-task/src/types.ts +++ b/orchestrator-v2/packages/agent-3-task/src/types.ts @@ -15,13 +15,9 @@ export type TaskAgentState = { sessionId?: string; }; -export type ParsedTaskAgentState = - | { ok: true; state: TaskAgentState } - | { ok: false; missing: string[] }; - const REQUIRED_FIELDS = ["workingDir", "instructions", "engineName"] as const; -export function parseTaskAgentState(taskState: Record): ParsedTaskAgentState { +export function getTaskAgentStateMissingFields(taskState: Record): string[] { const missing: string[] = []; for (const field of REQUIRED_FIELDS) { @@ -30,10 +26,6 @@ export function parseTaskAgentState(taskState: Record): ParsedT } } - if (missing.length > 0) { - return { ok: false, missing }; - } - const workingDir = taskState.workingDir; const instructions = taskState.instructions; const engineName = taskState.engineName; @@ -52,19 +44,16 @@ export function parseTaskAgentState(taskState: Record): ParsedT missing.push("sessionId"); } + return [...new Set(missing)]; +} + +export function verifyIsTaskAgentState( + taskState: Record, +): asserts taskState is TaskAgentState { + const missing = getTaskAgentStateMissingFields(taskState); if (missing.length > 0) { - return { ok: false, missing: [...new Set(missing)] }; + throw new Error(missingFieldsMessage(missing)); } - - return { - ok: true, - state: { - workingDir: workingDir as string, - instructions: instructions as string, - engineName: engineName as string, - ...(sessionId !== undefined ? { sessionId: sessionId as string } : {}), - }, - }; } export function missingFieldsMessage(missing: string[]): string { diff --git a/orchestrator-v2/packages/agent-4-workflow/package.json b/orchestrator-v2/packages/agent-4-workflow/package.json new file mode 100644 index 0000000..fe36bc8 --- /dev/null +++ b/orchestrator-v2/packages/agent-4-workflow/package.json @@ -0,0 +1,39 @@ +{ + "name": "@bifrost-ai/agent-4-workflow", + "version": "0.1.0", + "description": "Workflow Agent — DAG scheduling with dependencies", + "license": "MIT", + "files": [ + "dist" + ], + "type": "module", + "exports": { + ".": "./dist/index.mjs", + "./augment": "./dist/augment.mjs", + "./package.json": "./package.json" + }, + "publishConfig": { + "access": "public" + }, + "scripts": { + "build": "vp pack", + "dev": "vp pack --watch", + "test": "vp test", + "check": "vp check", + "prepublishOnly": "vp run build" + }, + "dependencies": { + "@bifrost-ai/interfaces-work": "workspace:*", + "@bifrost-ai/runner": "workspace:*" + }, + "devDependencies": { + "@bifrost-ai/orchestrator": "workspace:*", + "@bifrost-ai/protocol": "workspace:*", + "@types/node": "catalog:", + "@typescript/native-preview": "catalog:", + "bumpp": "catalog:", + "typescript": "catalog:", + "vite-plus": "catalog:", + "vitest-gwt": "catalog:" + } +} diff --git a/orchestrator-v2/packages/agent-4-workflow/src/augment.ts b/orchestrator-v2/packages/agent-4-workflow/src/augment.ts new file mode 100644 index 0000000..c2d331e --- /dev/null +++ b/orchestrator-v2/packages/agent-4-workflow/src/augment.ts @@ -0,0 +1,89 @@ +import type { ScriptFn } from "@bifrost-ai/interfaces-work"; +import { Runner } from "@bifrost-ai/runner"; + +import { flattenWorkflowBuilder } from "./flatten-workflow.js"; +import { createWorkflowScript } from "./create-workflow-agent.js"; +import type { ScriptRef } from "./step-refs.js"; +import { createStepDecorator } from "./step-wrapper.js"; +import type { WorkflowDefinition } from "./types.js"; +import { Workflow } from "./workflow.js"; + +declare module "@bifrost-ai/runner" { + // oxlint-disable-next-line typescript/consistent-type-definitions -- module augmentation + interface Runner { + registerWorkflowAgent(workflow: Workflow): WorkflowDefinition; + } +} + +Runner.prototype.registerWorkflowAgent = function registerWorkflowAgent( + this: Runner, + workflow: Workflow, +): WorkflowDefinition { + const definition = flattenWorkflowBuilder(workflow); + registerScriptSteps(this, workflow); + registerStepDecorators(this, definition); + validateDefinition(this, definition); + this.registerScript(definition.name, createWorkflowScript(definition)); + return definition; +}; + +function validateDefinition(runner: Runner, definition: WorkflowDefinition): void { + for (const step of definition.steps) { + if (step.innerKind === "task" && !runner.hasScript(step.innerName)) { + throw new Error(`Task agent not registered: ${step.innerName}`); + } + + for (const decoratorName of step.flow) { + if (decoratorName === step.id) { + continue; + } + if (!runner.hasDecorator(decoratorName)) { + throw new Error(`Decorator not registered: ${decoratorName}`); + } + } + } +} + +function registerScriptSteps(runner: Runner, workflow: Workflow): void { + for (const ref of collectScriptRefs(workflow)) { + if (!runner.hasScript(ref.displayName)) { + runner.registerScript(ref.displayName, createWorkflowInlineScript(ref)); + } + } +} + +function createWorkflowInlineScript(ref: ScriptRef): ScriptFn { + return async (workItem, ctx) => ref.fn({ workItem, cwd: ctx.cwd, setState: ctx.setState }); +} + +function registerStepDecorators(runner: Runner, definition: WorkflowDefinition): void { + for (const step of definition.steps) { + if (step.decoratorFns !== undefined) { + for (const [name, fn] of Object.entries(step.decoratorFns)) { + if (!runner.hasDecorator(name)) { + runner.registerDecorator(name, fn); + } + } + } + + if (!runner.hasDecorator(step.id)) { + runner.registerDecorator(step.id, createStepDecorator(step)); + } + } +} + +function collectScriptRefs(workflow: Workflow): ScriptRef[] { + const refs: ScriptRef[] = []; + for (const group of workflow.groups) { + for (const item of group) { + if (item instanceof Workflow) { + refs.push(...collectScriptRefs(item)); + continue; + } + if (item.type === "script") { + refs.push(item); + } + } + } + return refs; +} diff --git a/orchestrator-v2/packages/agent-4-workflow/src/create-workflow-agent.ts b/orchestrator-v2/packages/agent-4-workflow/src/create-workflow-agent.ts new file mode 100644 index 0000000..b697f43 --- /dev/null +++ b/orchestrator-v2/packages/agent-4-workflow/src/create-workflow-agent.ts @@ -0,0 +1,8 @@ +import type { ScriptFn } from "@bifrost-ai/interfaces-work"; + +import { runWorkflowAgent } from "./run-workflow-agent.js"; +import type { WorkflowDefinition } from "./types.js"; + +export function createWorkflowScript(definition: WorkflowDefinition): ScriptFn { + return async (workItem, ctx) => runWorkflowAgent(workItem, ctx, definition); +} diff --git a/orchestrator-v2/packages/agent-4-workflow/src/flatten-workflow.spec.ts b/orchestrator-v2/packages/agent-4-workflow/src/flatten-workflow.spec.ts new file mode 100644 index 0000000..bfefec4 --- /dev/null +++ b/orchestrator-v2/packages/agent-4-workflow/src/flatten-workflow.spec.ts @@ -0,0 +1,137 @@ +import type { DecoratorFn } from "@bifrost-ai/interfaces-work"; +import { describe, expect } from "vite-plus/test"; +import test from "vitest-gwt"; + +import { flattenWorkflowBuilder } from "./flatten-workflow.js"; +import { continueStep } from "./step-result.js"; +import { script, task } from "./step-refs.js"; +import { Workflow } from "./workflow.js"; + +const noopDecorator: DecoratorFn = async (_workItem, _ctx, next) => next(); + +type Context = { + definition: ReturnType; +}; + +describe("flattenWorkflowBuilder", () => { + test("linear workflow chains dependencies", { + given: { linear_workflow }, + then: { steps_chain_a_to_b_to_c }, + }); + + test("diamond workflow fans out and joins", { + given: { diamond_workflow }, + then: { d_depends_on_b_and_c }, + }); + + test("nested workflow flattens with namespaced ids", { + given: { nested_workflow }, + then: { nested_steps_are_namespaced }, + }); + + test("parallel steps with same name get unique ids", { + given: { parallel_same_name_workflow }, + then: { parallel_step_ids_are_unique }, + }); + + test("duplicate nested workflow names get distinct ids", { + given: { duplicate_nested_workflow }, + then: { nested_workflow_ids_are_distinct }, + }); + + test("step decorators are resolved into flow with step wrapper outermost", { + given: { decorated_workflow }, + then: { step_flow_includes_custom_decorators }, + }); +}); + +function linear_workflow(this: Context) { + const workflow = new Workflow({ name: "linear" }).step(task("a")).step(task("b")).step(task("c")); + this.definition = flattenWorkflowBuilder(workflow); +} + +function steps_chain_a_to_b_to_c(this: Context) { + expect(this.definition.steps).toHaveLength(3); + expect(this.definition.steps[0].dependsOn).toEqual([]); + expect(this.definition.steps[1].dependsOn).toEqual([this.definition.steps[0].id]); + expect(this.definition.steps[2].dependsOn).toEqual([this.definition.steps[1].id]); +} + +function diamond_workflow(this: Context) { + const workflow = new Workflow({ name: "diamond" }) + .step(task("a")) + .step(task("b"), task("c")) + .step(task("d")); + this.definition = flattenWorkflowBuilder(workflow); +} + +function d_depends_on_b_and_c(this: Context) { + const [a, b, c, d] = this.definition.steps; + expect(a.dependsOn).toEqual([]); + expect(b.dependsOn).toEqual([a.id]); + expect(c.dependsOn).toEqual([a.id]); + expect(d.dependsOn.sort()).toEqual([b.id, c.id].sort()); +} + +function nested_workflow(this: Context) { + const inner = new Workflow({ name: "inner" }) + .step(script(() => continueStep(), "someFn")) + .step(task("next")); + const workflow = new Workflow({ name: "outer" }) + .step(task("a")) + .step(task("b"), inner) + .step(script(() => continueStep(), "last")); + this.definition = flattenWorkflowBuilder(workflow); +} + +function nested_steps_are_namespaced(this: Context) { + const innerStep = this.definition.steps.find((step) => + step.id.includes("step2-2[inner]:step1-1[someFn]"), + ); + const lastStep = this.definition.steps.find((step) => step.id.includes("[last]")); + expect(innerStep).toBeDefined(); + expect(lastStep?.dependsOn.length).toBeGreaterThan(0); +} + +function parallel_same_name_workflow(this: Context) { + const workflow = new Workflow({ name: "parallel" }).step(task("same"), task("same")); + this.definition = flattenWorkflowBuilder(workflow); +} + +function parallel_step_ids_are_unique(this: Context) { + const ids = this.definition.steps.map((step) => step.id); + expect(ids).toHaveLength(2); + expect(new Set(ids).size).toBe(ids.length); + expect(ids[0]).not.toBe(ids[1]); +} + +function duplicate_nested_workflow(this: Context) { + const inner = new Workflow({ name: "inner" }).step(task("x")); + const workflow = new Workflow({ name: "outer" }).step(inner, inner).step(task("after")); + this.definition = flattenWorkflowBuilder(workflow); +} + +function nested_workflow_ids_are_distinct(this: Context) { + const innerSteps = this.definition.steps.filter((step) => step.id.includes("[inner]")); + expect(innerSteps).toHaveLength(2); + expect(innerSteps[0]?.id).not.toBe(innerSteps[1]?.id); + expect(innerSteps[0]?.dependsOn).toEqual([]); + expect(innerSteps[1]?.dependsOn).toEqual([]); + const afterStep = this.definition.steps.find((step) => step.id.includes("[after]")); + expect(afterStep?.dependsOn.sort()).toEqual(innerSteps.map((step) => step.id).sort()); +} + +function decorated_workflow(this: Context) { + const workflow = new Workflow({ name: "decorated" }).step(task("a"), ["logging"]).step( + script(() => continueStep(), "inline"), + [{ name: "metrics", fn: noopDecorator }], + ); + this.definition = flattenWorkflowBuilder(workflow); +} + +function step_flow_includes_custom_decorators(this: Context) { + const [taskStep, scriptStep] = this.definition.steps; + expect(taskStep?.flow).toEqual([taskStep?.id, "logging"]); + expect(scriptStep?.flow).toEqual([scriptStep?.id, "metrics"]); + expect(scriptStep?.decoratorFns?.metrics).toBeTypeOf("function"); +} diff --git a/orchestrator-v2/packages/agent-4-workflow/src/flatten-workflow.ts b/orchestrator-v2/packages/agent-4-workflow/src/flatten-workflow.ts new file mode 100644 index 0000000..ddbe29d --- /dev/null +++ b/orchestrator-v2/packages/agent-4-workflow/src/flatten-workflow.ts @@ -0,0 +1,85 @@ +import type { FlattenedStep, WorkflowDefinition } from "./types.js"; +import type { WorkflowStepInput } from "./step-refs.js"; +import { Workflow } from "./workflow.js"; + +export function flattenWorkflowBuilder(workflow: Workflow, parentPrefix = ""): WorkflowDefinition { + const name = workflow.name; + const prefix = parentPrefix.length > 0 ? `${parentPrefix}:${name}` : name; + const { steps } = flattenWorkflowGroup(workflow, prefix, []); + return { name, steps }; +} + +function flattenWorkflowGroup( + workflow: Workflow, + prefix: string, + entryDeps: string[], +): { steps: FlattenedStep[]; exitStepIds: string[] } { + const steps: FlattenedStep[] = []; + let previousExitIds = [...entryDeps]; + + for (const [groupIndex, group] of workflow.groups.entries()) { + const groupExitIds: string[] = []; + + for (const [itemIndex, item] of group.entries()) { + if (item instanceof Workflow) { + const nestedPrefix = `${prefix}:step${groupIndex + 1}-${itemIndex + 1}[${item.name}]`; + const nested = flattenWorkflowGroup(item, nestedPrefix, previousExitIds); + steps.push(...nested.steps); + if (nested.exitStepIds.length > 0) { + groupExitIds.push(...nested.exitStepIds); + } + continue; + } + + const stepId = buildStepId(prefix, groupIndex, itemIndex, item); + const { flow, decoratorFns } = resolveStepFlow(item, stepId); + steps.push({ + id: stepId, + innerKind: item.type === "task" ? "task" : "script", + innerName: item.type === "task" ? item.name : item.displayName, + dependsOn: [...previousExitIds], + flow, + ...(Object.keys(decoratorFns).length > 0 ? { decoratorFns } : {}), + }); + groupExitIds.push(stepId); + } + + previousExitIds = groupExitIds; + } + + return { steps, exitStepIds: previousExitIds }; +} + +function buildStepId( + prefix: string, + groupIndex: number, + itemIndex: number, + item: WorkflowStepInput, +): string { + const stepLabel = `step${groupIndex + 1}-${itemIndex + 1}`; + if (item.type === "script") { + return `${prefix}:${stepLabel}[${item.displayName}]`; + } + return `${prefix}:${stepLabel}[${item.name}]`; +} + +function resolveStepFlow( + step: WorkflowStepInput, + stepId: string, +): { flow: string[]; decoratorFns: NonNullable } { + const flow: string[] = [stepId]; + const decoratorFns: NonNullable = {}; + + for (const [index, decorator] of (step.decorators ?? []).entries()) { + if (typeof decorator === "string") { + flow.push(decorator); + continue; + } + + const name = decorator.name.length > 0 ? decorator.name : `${stepId}:decorator-${index}`; + flow.push(name); + decoratorFns[name] = decorator.fn; + } + + return { flow, decoratorFns }; +} diff --git a/orchestrator-v2/packages/agent-4-workflow/src/index.ts b/orchestrator-v2/packages/agent-4-workflow/src/index.ts new file mode 100644 index 0000000..39335cf --- /dev/null +++ b/orchestrator-v2/packages/agent-4-workflow/src/index.ts @@ -0,0 +1,31 @@ +export { createWorkflowScript } from "./create-workflow-agent.js"; +export { flattenWorkflowBuilder } from "./flatten-workflow.js"; +export { runWorkflowAgent } from "./run-workflow-agent.js"; +export { createStepDecorator, pauseWorkItem, runStepDecorator } from "./step-wrapper.js"; +export { continueStep, failStep, isStepResult, parseStepOutput, pauseStep } from "./step-result.js"; +export type { StepResult } from "./step-result.js"; +export { script, task } from "./step-refs.js"; +export type { + ScriptRef, + StepDecorator, + TaskRef, + WorkflowScriptFn, + WorkflowStepInput, +} from "./step-refs.js"; +export { Workflow } from "./workflow.js"; +export type { WorkflowGroupItem } from "./workflow.js"; +export type { + FlattenedStep, + StepTransition, + StepWrapperState, + WorkflowDefinition, + WorkflowPhase, + WorkflowState, + WorkflowStateParseResult, +} from "./types.js"; +export { + getWorkflowStateMissingFields, + missingFieldsMessage, + parseWorkflowState, + verifyIsWorkflowState, +} from "./types.js"; diff --git a/orchestrator-v2/packages/agent-4-workflow/src/run-workflow-agent.spec.ts b/orchestrator-v2/packages/agent-4-workflow/src/run-workflow-agent.spec.ts new file mode 100644 index 0000000..a59dd68 --- /dev/null +++ b/orchestrator-v2/packages/agent-4-workflow/src/run-workflow-agent.spec.ts @@ -0,0 +1,278 @@ +import type { + CreateDraftWorkItemInput, + ScriptContext, + WorkItem, + WorkItemSourceClient, + WorkItemStatus, +} from "@bifrost-ai/interfaces-work"; +import { describe, expect } from "vite-plus/test"; +import test from "vitest-gwt"; + +import { runWorkflowAgent } from "./run-workflow-agent.js"; +import type { WorkflowDefinition } from "./types.js"; + +type Context = { + workItem: WorkItem; + ctx: ScriptContext; + definition: WorkflowDefinition; + error: Error | null; + workItemSource: MockSource; +}; + +class MockSource implements WorkItemSourceClient { + public drafts: Array<{ + input: Parameters[0]; + id: string; + }> = []; + public started: string[] = []; + public dependencies: Array<{ workItemId: string; dependsOnWorkItemId: string }> = []; + public statuses = new Map(); + public paused: string[] = []; + public completed: string[] = []; + public failed: Array<{ workItemId: string; error: string }> = []; + private nextId = 1; + + async completeWorkItem(workItemId: string) { + this.completed.push(workItemId); + this.statuses.set(workItemId, "completed"); + } + + async failWorkItem(workItemId: string, error: string) { + this.failed.push({ workItemId, error }); + this.statuses.set(workItemId, "failed"); + } + + async pauseWorkItem(workItemId: string) { + this.paused.push(workItemId); + this.statuses.set(workItemId, "paused"); + } + + async createDraftWorkItem(input: CreateDraftWorkItemInput) { + const id = `child-${this.nextId}`; + this.nextId += 1; + this.drafts.push({ input, id }); + this.statuses.set(id, "draft"); + return id; + } + + async startWorkItem(workItemId: string) { + this.started.push(workItemId); + this.statuses.set(workItemId, "live"); + } + + async setDependency(workItemId: string, dependsOnWorkItemId: string) { + this.dependencies.push({ workItemId, dependsOnWorkItemId }); + } + + async getDependencies() { + return []; + } + + async getWorkItemStatus(workItemId: string) { + return this.statuses.get(workItemId) ?? "draft"; + } + + async setState() {} +} + +const linearDefinition: WorkflowDefinition = { + name: "linear", + steps: [ + { id: "step-a", innerKind: "task", innerName: "a", dependsOn: [], flow: ["step-a"] }, + { + id: "step-b", + innerKind: "task", + innerName: "b", + dependsOn: ["step-a"], + flow: ["step-b"], + }, + { + id: "step-c", + innerKind: "task", + innerName: "c", + dependsOn: ["step-b"], + flow: ["step-c"], + }, + ], +}; + +describe("runWorkflowAgent", () => { + test("schedule pass creates children and pauses", { + given: { schedule_fixture }, + when: { running_workflow }, + then: { workflow_is_paused, children_created_and_started }, + }); + + test("verify pass throws when a child failed", { + given: { verify_fixture_with_failed_child }, + when: { running_workflow }, + then: { workflow_throws }, + }); + + test("verify pass completes when all children completed", { + given: { verify_fixture_all_completed }, + when: { running_workflow }, + then: { run_succeeds }, + }); + + test("verify pass pauses when a child is still live", { + given: { verify_fixture_with_live_child }, + when: { running_workflow }, + then: { workflow_is_paused }, + }); +}); + +function schedule_fixture(this: Context) { + this.workItemSource = new MockSource(); + this.definition = linearDefinition; + this.workItem = { + workItemId: "workflow-1", + kind: "workflow", + name: "linear", + flow: [], + state: { + workingDir: "/tmp", + phase: "schedule", + }, + metadata: {}, + }; + this.ctx = makeCtx(this.workItemSource); +} + +function verify_fixture_with_failed_child(this: Context) { + this.workItemSource = new MockSource(); + this.definition = linearDefinition; + this.workItemSource.statuses.set("child-1", "completed"); + this.workItemSource.statuses.set("child-2", "failed"); + this.workItemSource.statuses.set("child-3", "completed"); + this.workItem = { + workItemId: "workflow-1", + kind: "workflow", + name: "linear", + flow: [], + state: { + workingDir: "/tmp", + phase: "verify", + childIds: { + "step-a": "child-1", + "step-b": "child-2", + "step-c": "child-3", + }, + }, + metadata: {}, + }; + this.ctx = makeCtx(this.workItemSource); +} + +function verify_fixture_all_completed(this: Context) { + this.workItemSource = new MockSource(); + this.definition = linearDefinition; + for (const id of ["child-1", "child-2", "child-3"]) { + this.workItemSource.statuses.set(id, "completed"); + } + this.workItem = { + workItemId: "workflow-1", + kind: "workflow", + name: "linear", + flow: [], + state: { + workingDir: "/tmp", + phase: "verify", + childIds: { + "step-a": "child-1", + "step-b": "child-2", + "step-c": "child-3", + }, + }, + metadata: {}, + }; + this.ctx = makeCtx(this.workItemSource); +} + +function verify_fixture_with_live_child(this: Context) { + this.workItemSource = new MockSource(); + this.definition = linearDefinition; + this.workItemSource.statuses.set("child-1", "completed"); + this.workItemSource.statuses.set("child-2", "live"); + this.workItemSource.statuses.set("child-3", "completed"); + this.workItem = { + workItemId: "workflow-1", + kind: "workflow", + name: "linear", + flow: [], + state: { + workingDir: "/tmp", + phase: "verify", + childIds: { + "step-a": "child-1", + "step-b": "child-2", + "step-c": "child-3", + }, + }, + metadata: {}, + }; + this.ctx = makeCtx(this.workItemSource); +} + +async function running_workflow(this: Context) { + this.error = null; + try { + await runWorkflowAgent(this.workItem, this.ctx, this.definition); + } catch (error) { + this.error = error as Error; + } +} + +function workflow_is_paused(this: Context) { + expect(this.error).toBeNull(); + expect(this.workItemSource.paused).toContain("workflow-1"); +} + +function children_created_and_started(this: Context) { + expect(this.workItemSource.drafts).toHaveLength(3); + expect(this.workItemSource.drafts[0]?.input).toMatchObject({ + kind: "task", + name: "a", + flow: ["step-a"], + state: { + workflowWorkItemId: "workflow-1", + workingDir: "/tmp", + }, + }); + expect(this.workItemSource.started).toEqual(["child-1", "child-2", "child-3"]); + expect(this.workItemSource.dependencies.some((dep) => dep.workItemId === "workflow-1")).toBe( + true, + ); +} + +function workflow_throws(this: Context) { + expect(this.error?.message).toContain("failed"); +} + +function run_succeeds(this: Context) { + expect(this.error).toBeNull(); +} + +function makeCtx(workItemSource: MockSource): ScriptContext { + const state: Record = {}; + return { + cwd: "/tmp", + data: { + get() { + return { + get() { + return undefined; + }, + has() { + return false; + }, + register() {}, + }; + }, + }, + workItemSource, + async setState(nextState) { + Object.assign(state, nextState); + }, + }; +} diff --git a/orchestrator-v2/packages/agent-4-workflow/src/run-workflow-agent.ts b/orchestrator-v2/packages/agent-4-workflow/src/run-workflow-agent.ts new file mode 100644 index 0000000..7fa67a5 --- /dev/null +++ b/orchestrator-v2/packages/agent-4-workflow/src/run-workflow-agent.ts @@ -0,0 +1,136 @@ +import type { ScriptContext, WorkItem } from "@bifrost-ai/interfaces-work"; + +import { pauseWorkItem } from "./step-wrapper.js"; +import type { WorkflowDefinition, WorkflowState } from "./types.js"; +import { verifyIsWorkflowState } from "./types.js"; + +export async function runWorkflowAgent( + workItem: WorkItem, + ctx: ScriptContext, + definition: WorkflowDefinition, +): Promise { + verifyIsWorkflowState(workItem.state); + + const state = workItem.state; + + if (workItem.name !== definition.name) { + throw new Error( + `Workflow definition mismatch: expected ${definition.name}, got ${workItem.name}`, + ); + } + + const phase = state.phase ?? "schedule"; + if (phase === "schedule") { + await schedulePass(workItem, ctx, definition, state); + return; + } + + await verifyPass(workItem, ctx, definition, state); +} + +async function schedulePass( + workItem: WorkItem, + ctx: ScriptContext, + definition: WorkflowDefinition, + state: WorkflowState, +): Promise { + const childIds = { ...state.childIds }; + + if (Object.keys(childIds).length === 0) { + for (const step of definition.steps) { + if (childIds[step.id] !== undefined) { + continue; + } + + const childId = await ctx.workItemSource.createDraftWorkItem({ + kind: step.innerKind, + name: step.innerName, + flow: step.flow, + state: { + workflowWorkItemId: workItem.workItemId, + workingDir: state.workingDir, + }, + metadata: { + workflowName: definition.name, + stepId: step.id, + }, + }); + childIds[step.id] = childId; + } + + await wireDependencies(ctx, definition, childIds); + await startChildren(ctx, workItem.workItemId, definition, childIds); + } + + await ctx.setState({ + ...workItem.state, + phase: "verify", + childIds, + }); + + await pauseWorkItem(ctx, workItem.workItemId); +} + +async function wireDependencies( + ctx: ScriptContext, + definition: WorkflowDefinition, + childIds: Record, +): Promise { + for (const step of definition.steps) { + const childId = childIds[step.id]; + if (childId === undefined) { + throw new Error(`Missing child for step ${step.id}`); + } + + for (const depStepId of step.dependsOn) { + const depChildId = childIds[depStepId]; + if (depChildId === undefined) { + throw new Error(`Missing dependency child for ${depStepId}`); + } + await ctx.workItemSource.setDependency(childId, depChildId); + } + } +} + +async function startChildren( + ctx: ScriptContext, + workflowWorkItemId: string, + definition: WorkflowDefinition, + childIds: Record, +): Promise { + for (const step of definition.steps) { + const childId = childIds[step.id]; + if (childId !== undefined) { + await ctx.workItemSource.startWorkItem(childId); + await ctx.workItemSource.setDependency(workflowWorkItemId, childId); + } + } +} + +async function verifyPass( + workItem: WorkItem, + ctx: ScriptContext, + definition: WorkflowDefinition, + state: WorkflowState, +): Promise { + const childIds = state.childIds; + if (childIds === undefined || Object.keys(childIds).length === 0) { + throw new Error("Workflow verify pass missing childIds"); + } + + for (const step of definition.steps) { + const childId = childIds[step.id]; + if (childId === undefined) { + throw new Error(`Missing child id for step ${step.id}`); + } + + const status = await ctx.workItemSource.getWorkItemStatus(childId); + if (status === "failed") { + throw new Error(`Step ${step.id} failed`); + } + if (status !== "completed") { + await pauseWorkItem(ctx, workItem.workItemId); + return; + } + } +} diff --git a/orchestrator-v2/packages/agent-4-workflow/src/step-refs.ts b/orchestrator-v2/packages/agent-4-workflow/src/step-refs.ts new file mode 100644 index 0000000..c5c6d69 --- /dev/null +++ b/orchestrator-v2/packages/agent-4-workflow/src/step-refs.ts @@ -0,0 +1,38 @@ +import type { DecoratorFn, ScriptContext, WorkItem } from "@bifrost-ai/interfaces-work"; + +import type { StepResult } from "./step-result.js"; + +export type StepDecorator = string | { name: string; fn: DecoratorFn }; + +export type TaskRef = { + type: "task"; + name: string; + decorators?: StepDecorator[]; +}; + +export type WorkflowScriptFn = (ctx: { + workItem: WorkItem; + cwd: string; + setState: ScriptContext["setState"]; +}) => Promise | StepResult; + +export type ScriptRef = { + type: "script"; + fn: WorkflowScriptFn; + displayName: string; + decorators?: StepDecorator[]; +}; + +export type WorkflowStepInput = TaskRef | ScriptRef; + +export function task(name: string): TaskRef { + return { type: "task", name }; +} + +export function script(fn: WorkflowScriptFn, displayName?: string): ScriptRef { + return { + type: "script", + fn, + displayName: displayName ?? (fn.name || "anonymous"), + }; +} diff --git a/orchestrator-v2/packages/agent-4-workflow/src/step-result.ts b/orchestrator-v2/packages/agent-4-workflow/src/step-result.ts new file mode 100644 index 0000000..5e23065 --- /dev/null +++ b/orchestrator-v2/packages/agent-4-workflow/src/step-result.ts @@ -0,0 +1,35 @@ +import type { ExecutionStats } from "@bifrost-ai/interfaces-work"; + +export type StepResult = + | { transition: "continue"; message?: string; telemetry?: ExecutionStats } + | { transition: "fail"; message?: string } + | { transition: "pause" }; + +export function continueStep(message?: string, telemetry?: ExecutionStats): StepResult { + return { transition: "continue", message, telemetry }; +} + +export function failStep(message?: string): StepResult { + return { transition: "fail", message }; +} + +export function pauseStep(): StepResult { + return { transition: "pause" }; +} + +export function isStepResult(value: unknown): value is StepResult { + if (value === null || typeof value !== "object") { + return false; + } + + const transition = (value as StepResult).transition; + return transition === "continue" || transition === "fail" || transition === "pause"; +} + +export function parseStepOutput(result: unknown): StepResult { + if (isStepResult(result)) { + return result; + } + + return { transition: "fail", message: "Invalid step result" }; +} diff --git a/orchestrator-v2/packages/agent-4-workflow/src/step-wrapper.spec.ts b/orchestrator-v2/packages/agent-4-workflow/src/step-wrapper.spec.ts new file mode 100644 index 0000000..9f4d017 --- /dev/null +++ b/orchestrator-v2/packages/agent-4-workflow/src/step-wrapper.spec.ts @@ -0,0 +1,159 @@ +import type { + CreateDraftWorkItemInput, + ScriptContext, + WorkItemSourceClient, + WorkItemStatus, + WorkItemDependency, +} from "@bifrost-ai/interfaces-work"; +import { describe, expect } from "vite-plus/test"; +import test from "vitest-gwt"; + +import { continueStep, failStep, pauseStep } from "./step-result.js"; +import { runStepDecorator } from "./step-wrapper.js"; +import type { StepWrapperState } from "./types.js"; + +type Context = { + error: Error | null; + workItemSource: MockSource; + innerResult: unknown; +}; + +class MockSource implements WorkItemSourceClient { + public workflowStateUpdates: Array<{ workItemId: string; state: Record }> = []; + public paused: Array<{ workItemId: string }> = []; + + async completeWorkItem(): Promise { + throw new Error("not implemented"); + } + + async failWorkItem(): Promise { + throw new Error("not implemented"); + } + + async pauseWorkItem(workItemId: string): Promise { + this.paused.push({ workItemId }); + } + + async createDraftWorkItem(_input: CreateDraftWorkItemInput): Promise { + throw new Error("not implemented"); + } + + async startWorkItem(_workItemId: string): Promise { + throw new Error("not implemented"); + } + + async setDependency(_workItemId: string, _dependsOnWorkItemId: string): Promise { + throw new Error("not implemented"); + } + + async getDependencies(_workItemId: string): Promise { + return []; + } + + async getWorkItemStatus(_workItemId: string): Promise { + return "live"; + } + + async setState(workItemId: string, state: Record): Promise { + this.workflowStateUpdates.push({ workItemId, state }); + } +} + +const wrapperState: StepWrapperState = { + workflowWorkItemId: "workflow-1", + workingDir: "/tmp", +}; + +describe("runStepDecorator", () => { + test("continue step result completes wrapper", { + given: { continue_step_fixture }, + when: { running_decorator }, + then: { run_succeeds }, + }); + + test("fail step result throws", { + given: { fail_step_fixture }, + when: { running_decorator }, + then: { fail_is_thrown }, + }); + + test("pause step result pauses work item", { + given: { pause_step_fixture }, + when: { running_decorator }, + then: { work_item_is_paused }, + }); + + test("thrown inner error is converted to fail transition", { + given: { thrown_error_fixture }, + when: { running_decorator }, + then: { fail_is_thrown }, + }); +}); + +function continue_step_fixture(this: Context) { + this.workItemSource = new MockSource(); + this.innerResult = continueStep("done"); +} + +function fail_step_fixture(this: Context) { + this.workItemSource = new MockSource(); + this.innerResult = failStep("boom"); +} + +function pause_step_fixture(this: Context) { + this.workItemSource = new MockSource(); + this.innerResult = pauseStep(); +} + +function thrown_error_fixture(this: Context) { + this.workItemSource = new MockSource(); + this.innerResult = new Error("boom"); +} + +async function running_decorator(this: Context) { + this.error = null; + try { + await runStepDecorator("step-child-1", wrapperState, makeCtx(this.workItemSource), async () => { + if (this.innerResult instanceof Error) { + throw this.innerResult; + } + return this.innerResult; + }); + } catch (error) { + this.error = error as Error; + } +} + +function run_succeeds(this: Context) { + expect(this.error).toBeNull(); +} + +function fail_is_thrown(this: Context) { + expect(this.error?.message).toBe("boom"); +} + +function work_item_is_paused(this: Context) { + expect(this.error).toBeNull(); + expect(this.workItemSource.paused).toEqual([{ workItemId: "step-child-1" }]); +} + +function makeCtx(workItemSource: MockSource): ScriptContext { + return { + cwd: "/tmp", + data: { + get() { + return { + get() { + return undefined; + }, + has() { + return false; + }, + register() {}, + }; + }, + }, + workItemSource, + async setState() {}, + }; +} diff --git a/orchestrator-v2/packages/agent-4-workflow/src/step-wrapper.ts b/orchestrator-v2/packages/agent-4-workflow/src/step-wrapper.ts new file mode 100644 index 0000000..b26958e --- /dev/null +++ b/orchestrator-v2/packages/agent-4-workflow/src/step-wrapper.ts @@ -0,0 +1,66 @@ +import type { DecoratorFn, ScriptContext } from "@bifrost-ai/interfaces-work"; + +import { parseStepOutput } from "./step-result.js"; +import type { StepResult } from "./step-result.js"; +import type { FlattenedStep, StepWrapperState } from "./types.js"; + +export async function pauseWorkItem(ctx: ScriptContext, workItemId: string): Promise { + await ctx.workItemSource.pauseWorkItem(workItemId); +} + +export function createStepDecorator(_step: FlattenedStep): DecoratorFn { + return async (workItem, ctx, next) => + runStepDecorator(workItem.workItemId, workItem.state, ctx, next); +} + +export async function runStepDecorator( + workItemId: string, + state: Record, + ctx: ScriptContext, + next: () => Promise, +): Promise { + verifyStepWrapperState(state); + + let rawResult: unknown; + try { + rawResult = await next(); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + await applyStepResult({ transition: "fail", message }, workItemId, ctx); + return; + } + + await applyStepResult(parseStepOutput(rawResult), workItemId, ctx); +} + +async function applyStepResult( + result: StepResult, + workItemId: string, + ctx: ScriptContext, +): Promise { + if (result.transition === "continue") { + return; + } + + if (result.transition === "pause") { + await pauseWorkItem(ctx, workItemId); + return; + } + + throw new Error(result.message ?? "fail"); +} + +function verifyStepWrapperState(state: Record): asserts state is StepWrapperState { + const required = ["workflowWorkItemId", "workingDir"] as const; + const missing: string[] = []; + + for (const field of required) { + if (!(field in state) || state[field] === undefined) { + missing.push(field); + } + } + + if (missing.length > 0) { + throw new Error(`Invalid step wrapper state: ${missing.join(", ")}`); + } +} diff --git a/orchestrator-v2/packages/agent-4-workflow/src/types.ts b/orchestrator-v2/packages/agent-4-workflow/src/types.ts new file mode 100644 index 0000000..748dfbd --- /dev/null +++ b/orchestrator-v2/packages/agent-4-workflow/src/types.ts @@ -0,0 +1,78 @@ +import type { DecoratorFn } from "@bifrost-ai/interfaces-work"; + +export type WorkflowPhase = "schedule" | "verify"; + +export type StepTransition = "continue" | "fail" | "pause"; + +export type FlattenedStep = { + id: string; + innerKind: "task" | "script"; + innerName: string; + dependsOn: string[]; + flow: string[]; + decoratorFns?: Record; +}; + +export type WorkflowDefinition = { + name: string; + steps: FlattenedStep[]; +}; + +export type WorkflowState = { + workingDir: string; + phase?: WorkflowPhase; + childIds?: Record; +}; + +export type StepWrapperState = { + workflowWorkItemId: string; + workingDir: string; +}; + +export type WorkflowStateParseResult = { ok: true } | { ok: false; missing: string[] }; + +export function getWorkflowStateMissingFields(taskState: Record): string[] { + const missing: string[] = []; + + const workingDir = taskState.workingDir; + if (typeof workingDir !== "string" || workingDir.length === 0) { + missing.push("workingDir"); + } + + const phase = taskState.phase; + const childIds = taskState.childIds; + + if (phase !== undefined && phase !== "schedule" && phase !== "verify") { + missing.push("phase"); + } + if ( + childIds !== undefined && + (childIds === null || typeof childIds !== "object" || Array.isArray(childIds)) + ) { + missing.push("childIds"); + } + + return [...new Set(missing)]; +} + +export function verifyIsWorkflowState( + taskState: Record, +): asserts taskState is WorkflowState { + const missing = getWorkflowStateMissingFields(taskState); + if (missing.length > 0) { + throw new Error(missingFieldsMessage(missing)); + } +} + +export function parseWorkflowState(taskState: Record): WorkflowStateParseResult { + const missing = getWorkflowStateMissingFields(taskState); + if (missing.length > 0) { + return { ok: false, missing }; + } + + return { ok: true }; +} + +export function missingFieldsMessage(missing: string[]): string { + return `Workflow agent state is missing required fields: ${missing.join(", ")}`; +} diff --git a/orchestrator-v2/packages/agent-4-workflow/src/workflow.integration.spec.ts b/orchestrator-v2/packages/agent-4-workflow/src/workflow.integration.spec.ts new file mode 100644 index 0000000..9fc506a --- /dev/null +++ b/orchestrator-v2/packages/agent-4-workflow/src/workflow.integration.spec.ts @@ -0,0 +1,120 @@ +import { mkdtemp, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { describe, expect } from "vite-plus/test"; +import test from "vitest-gwt"; +import { exportPrivateKeyPem, exportPublicKeyPem } from "@bifrost-ai/protocol"; +import { Runner } from "@bifrost-ai/runner"; +import "./augment.js"; +import { + authorizedRunnersFor, + createGraphMemoryWorkItemSource, + createIdentities, + startOrchestratorInBackground, + waitFor, +} from "@bifrost-ai/orchestrator/test-helpers"; + +import { continueStep } from "./step-result.js"; +import { script } from "./step-refs.js"; +import { Workflow } from "./workflow.js"; + +type Context = { + source: ReturnType; + runner: Runner; + abort: () => void; + done: Promise; + configPath: string; +}; + +describe("workflow agent integration", () => { + test("linear workflow completes after children run", { + given: { linear_integration_setup }, + when: { waiting_for_workflow_completion }, + then: { workflow_and_children_completed }, + }); +}); + +async function linear_integration_setup(this: Context) { + this.source = createGraphMemoryWorkItemSource([ + { + workItemId: "workflow-1", + kind: "workflow", + name: "linear-flow", + flow: [], + state: { + workingDir: "/tmp", + }, + metadata: {}, + }, + ]); + + const { orchestratorIdentity, runnerIdentity } = createIdentities(); + const background = await startOrchestratorInBackground({ + orchestratorIdentity, + authorizedRunners: authorizedRunnersFor(runnerIdentity), + workItemSource: this.source, + maxInFlightPerPeer: 4, + }); + this.abort = background.abort; + this.done = background.done; + + const configDir = await mkdtemp(join(tmpdir(), "workflow-runner-")); + this.configPath = join(configDir, "runner.yaml"); + await writeFile( + this.configPath, + [ + "orchestrator:", + ` url: ws://${background.address.host}:${background.address.port}`, + ` keyId: ${orchestratorIdentity.keyId}`, + ` publicKeyPem: |`, + ...exportPublicKeyPem(orchestratorIdentity.publicKey) + .split("\n") + .map((line) => ` ${line}`), + "identity:", + ` keyId: ${runnerIdentity.keyId}`, + ` privateKeyPem: |`, + ...exportPrivateKeyPem(runnerIdentity.privateKey) + .split("\n") + .map((line) => ` ${line}`), + ` publicKeyPem: |`, + ...exportPublicKeyPem(runnerIdentity.publicKey) + .split("\n") + .map((line) => ` ${line}`), + "heartbeatIntervalMs: 1000", + ].join("\n"), + ); + + this.runner = new Runner({ configPath: this.configPath }); + const workflow = new Workflow({ name: "linear-flow" }) + .step(script(() => continueStep(), "a")) + .step(script(() => continueStep(), "b")) + .step(script(() => continueStep(), "c")); + this.runner.registerWorkflowAgent(workflow); + await this.runner.start(); +} + +async function waiting_for_workflow_completion(this: Context) { + try { + await waitFor(() => this.source.completed.includes("workflow-1"), 10_000); + } catch (error) { + console.error("workflow debug", { + completed: this.source.completed, + failed: this.source.failed, + paused: this.source.paused, + startedOrder: this.source.startedOrder, + statuses: [...this.source.statuses.entries()], + }); + throw error; + } finally { + this.runner.close(); + this.source.abort(); + this.abort(); + } +} + +function workflow_and_children_completed(this: Context) { + expect(this.source.completed).toContain("workflow-1"); + const childCompleted = this.source.completed.filter((id) => id !== "workflow-1"); + expect(childCompleted).toEqual(["draft-1", "draft-2", "draft-3"]); +} diff --git a/orchestrator-v2/packages/agent-4-workflow/src/workflow.ts b/orchestrator-v2/packages/agent-4-workflow/src/workflow.ts new file mode 100644 index 0000000..aedf8c5 --- /dev/null +++ b/orchestrator-v2/packages/agent-4-workflow/src/workflow.ts @@ -0,0 +1,55 @@ +import type { StepDecorator, WorkflowStepInput } from "./step-refs.js"; + +export type WorkflowGroupItem = WorkflowStepInput | Workflow; + +export class Workflow { + public readonly name: string; + public readonly groups: WorkflowGroupItem[][] = []; + + public constructor(options: { name: string }) { + this.name = options.name; + } + + public step(...items: [...WorkflowGroupItem[], StepDecorator[]]): this; + public step(first: WorkflowGroupItem, ...rest: WorkflowGroupItem[]): this; + public step(...args: (WorkflowGroupItem | StepDecorator[])[]): this { + let decorators: StepDecorator[] = []; + let items: WorkflowGroupItem[]; + + const last = args.at(-1); + if (last !== undefined && isStepDecoratorArray(last)) { + decorators = last; + items = args.slice(0, -1) as WorkflowGroupItem[]; + } else { + items = args as WorkflowGroupItem[]; + } + + this.groups.push(items.map((item) => applyDecorators(item, decorators))); + return this; + } +} + +function applyDecorators(item: WorkflowGroupItem, decorators: StepDecorator[]): WorkflowGroupItem { + if (item instanceof Workflow || decorators.length === 0) { + return item; + } + + return { + ...item, + decorators: [...(item.decorators ?? []), ...decorators], + }; +} + +function isStepDecoratorArray(value: unknown): value is StepDecorator[] { + return Array.isArray(value) && value.every(isStepDecorator); +} + +function isStepDecorator(value: unknown): value is StepDecorator { + if (typeof value === "string") { + return true; + } + + return ( + value !== null && typeof value === "object" && "fn" in value && typeof value.fn === "function" + ); +} diff --git a/orchestrator-v2/packages/agent-4-workflow/tsconfig.json b/orchestrator-v2/packages/agent-4-workflow/tsconfig.json new file mode 100644 index 0000000..75eba9f --- /dev/null +++ b/orchestrator-v2/packages/agent-4-workflow/tsconfig.json @@ -0,0 +1,8 @@ +{ + "extends": "../../tsconfig.json", + "compilerOptions": { + "rootDir": "src", + "outDir": "dist" + }, + "include": ["src"] +} diff --git a/orchestrator-v2/packages/agent-4-workflow/vite.config.ts b/orchestrator-v2/packages/agent-4-workflow/vite.config.ts new file mode 100644 index 0000000..71fc502 --- /dev/null +++ b/orchestrator-v2/packages/agent-4-workflow/vite.config.ts @@ -0,0 +1,14 @@ +import { resolve } from "node:path"; +import { defineConfig } from "vite-plus"; + +export default defineConfig({ + test: { + testTimeout: 15_000, + }, + pack: { + entry: { + index: resolve(__dirname, "src/index.ts"), + augment: resolve(__dirname, "src/augment.ts"), + }, + }, +}); diff --git a/orchestrator-v2/packages/engine-claude-code/src/claude-code-engine.ts b/orchestrator-v2/packages/engine-claude-code/src/claude-code-engine.ts index 010bfbb..1f2e0d3 100644 --- a/orchestrator-v2/packages/engine-claude-code/src/claude-code-engine.ts +++ b/orchestrator-v2/packages/engine-claude-code/src/claude-code-engine.ts @@ -46,6 +46,7 @@ type ContentBlock = { const formatToolInput = (input: unknown): string => { if (!input || typeof input !== "object") { + // oxlint-disable-next-line typescript/no-base-to-string -- will never be "[object Object]" return String(input ?? ""); } const entries = Object.entries(input as Record).slice(0, 3); diff --git a/orchestrator-v2/packages/engine-cursor/src/cursor-engine.spec.ts b/orchestrator-v2/packages/engine-cursor/src/cursor-engine.spec.ts index 2faf8dc..fdb8525 100644 --- a/orchestrator-v2/packages/engine-cursor/src/cursor-engine.spec.ts +++ b/orchestrator-v2/packages/engine-cursor/src/cursor-engine.spec.ts @@ -110,7 +110,9 @@ describe("CursorEngine", () => { let mockResume: MockedFunction; beforeEach(() => { + // oxlint-disable-next-line typescript/unbound-method -- mocked mockCreate = vi.mocked(Agent.create) as MockedFunction; + // oxlint-disable-next-line typescript/unbound-method -- mocked mockResume = vi.mocked(Agent.resume) as MockedFunction; mockCreate.mockReset(); mockResume.mockReset(); diff --git a/orchestrator-v2/packages/engine-cursor/src/stream-preview.ts b/orchestrator-v2/packages/engine-cursor/src/stream-preview.ts index 2b89ddc..1460a0f 100644 --- a/orchestrator-v2/packages/engine-cursor/src/stream-preview.ts +++ b/orchestrator-v2/packages/engine-cursor/src/stream-preview.ts @@ -2,6 +2,7 @@ import type { SDKMessage } from "@cursor/sdk"; const formatToolInput = (input: unknown): string => { if (!input || typeof input !== "object") { + // oxlint-disable-next-line typescript/no-base-to-string -- will never be "[object Object]" return String(input ?? ""); } const entries = Object.entries(input as Record).slice(0, 3); diff --git a/orchestrator-v2/packages/interfaces-work/src/index.ts b/orchestrator-v2/packages/interfaces-work/src/index.ts index 942aa09..0852414 100644 --- a/orchestrator-v2/packages/interfaces-work/src/index.ts +++ b/orchestrator-v2/packages/interfaces-work/src/index.ts @@ -12,12 +12,12 @@ export type { WorkItemExecutionContext, WorkItemHandler, WorkItemHandlerRegistry, - WorkItemResult, WorkItemSource, + WorkItemSourceClient, + WorkItemStatus, } from "./types.js"; export { isWorkItemHandler, - isWorkItemResult, missingWorkItemFields, missingWorkItemFieldsMessage, isWorkItem as validateWorkItem, diff --git a/orchestrator-v2/packages/interfaces-work/src/types.ts b/orchestrator-v2/packages/interfaces-work/src/types.ts index d923ac5..fe33ce8 100644 --- a/orchestrator-v2/packages/interfaces-work/src/types.ts +++ b/orchestrator-v2/packages/interfaces-work/src/types.ts @@ -1,6 +1,7 @@ export type WorkItem = { workItemId: string; kind: string; + name: string; flow: string[]; state: Record; readonly metadata: Record; @@ -13,11 +14,14 @@ export type WorkItemDependency = { export type CreateDraftWorkItemInput = { kind: string; + name: string; flow?: string[]; state?: Record; metadata?: Record; }; +export type WorkItemStatus = "draft" | "live" | "paused" | "completed" | "failed"; + export type WorkItemSource = { watchWorkItems: () => AsyncGenerator; completeWorkItem: (workItemId: string) => Promise; @@ -28,11 +32,26 @@ export type WorkItemSource = { startWorkItem(workItemId: string): Promise; setDependency(workItemId: string, dependsOnWorkItemId: string, type?: string): Promise; getDependencies(workItemId: string): Promise; + getWorkItemStatus(workItemId: string): Promise; }; +export type WorkItemSourceClient = Pick< + WorkItemSource, + | "completeWorkItem" + | "failWorkItem" + | "pauseWorkItem" + | "createDraftWorkItem" + | "startWorkItem" + | "setDependency" + | "getDependencies" + | "getWorkItemStatus" + | "setState" +>; + export type ScriptContext = Record> = { cwd: string; data: DataRegistry; + workItemSource: WorkItemSourceClient; setState: (state: Record) => Promise; }; @@ -52,7 +71,7 @@ export type ScriptStack = Record>; }; -const REQUIRED_WORK_ITEM_FIELDS = ["workItemId", "kind", "flow"] as const; +const REQUIRED_WORK_ITEM_FIELDS = ["workItemId", "kind", "name", "flow"] as const; export function isWorkItem(value: unknown): value is WorkItem { if (value === null || typeof value !== "object") { @@ -65,6 +84,8 @@ export function isWorkItem(value: unknown): value is WorkItem { record.workItemId.length === 0 || typeof record.kind !== "string" || record.kind.length === 0 || + typeof record.name !== "string" || + record.name.length === 0 || !Array.isArray(record.flow) || !record.flow.every((entry) => typeof entry === "string" && entry.length > 0) || record.state === null || @@ -105,6 +126,9 @@ export function missingWorkItemFields(value: unknown): string[] { if (typeof record.kind === "string" && record.kind.length === 0) { missing.push("kind"); } + if (typeof record.name === "string" && record.name.length === 0) { + missing.push("name"); + } if (record.flow !== undefined && !Array.isArray(record.flow)) { missing.push("flow"); } @@ -152,19 +176,14 @@ export type WorkItemExecutionContext< > = { readonly data: DataRegistry; readonly handlers: WorkItemHandlerRegistry; + readonly workItemSource: WorkItemSourceClient; setState: (state: Record) => Promise; }; -export type WorkItemResult = { - outcome: "completed" | "failed" | "paused"; - message?: string; - telemetry?: ExecutionStats; -}; - export type WorkItemHandler = Record> = { kind: string; name: string; - run: (workItem: WorkItem, ctx: WorkItemExecutionContext) => Promise; + run: (workItem: WorkItem, ctx: WorkItemExecutionContext) => Promise; }; export function isWorkItemHandler(value: unknown): value is WorkItemHandler { @@ -179,14 +198,3 @@ export function isWorkItemHandler(value: unknown): value is WorkItemHandler { typeof record.run === "function" ); } - -export function isWorkItemResult(value: unknown): value is WorkItemResult { - if (value === null || typeof value !== "object") { - return false; - } - - const record = value as Partial; - return ( - record.outcome === "completed" || record.outcome === "failed" || record.outcome === "paused" - ); -} diff --git a/orchestrator-v2/packages/orchestrator/src/graph-memory-work-item-source.ts b/orchestrator-v2/packages/orchestrator/src/graph-memory-work-item-source.ts new file mode 100644 index 0000000..547809e --- /dev/null +++ b/orchestrator-v2/packages/orchestrator/src/graph-memory-work-item-source.ts @@ -0,0 +1,219 @@ +import type { + CreateDraftWorkItemInput, + WorkItem, + WorkItemDependency, + WorkItemSource, + WorkItemStatus, +} from "@bifrost-ai/interfaces-work"; + +const POLL_MS = 10; + +export type GraphMemoryWorkItemSource = WorkItemSource & { + completed: string[]; + failed: Array<{ workItemId: string; error: string }>; + paused: string[]; + states: Map>; + drafts: Map; + started: Set; + dependencies: Map; + items: Map; + statuses: Map; + startedOrder: string[]; + abort(): void; +}; + +export function createGraphMemoryWorkItemSource( + initialWorkItems: WorkItem[] = [], +): GraphMemoryWorkItemSource { + const completed: string[] = []; + const failed: Array<{ workItemId: string; error: string }> = []; + const paused: string[] = []; + const states = new Map>(); + const drafts = new Map(); + const started = new Set(); + const dependencies = new Map(); + const items = new Map(); + const statuses = new Map(); + const startedOrder: string[] = []; + const queued = new Set(); + const readyQueue: WorkItem[] = []; + let nextDraftId = 1; + let aborted = false; + + for (const workItem of initialWorkItems) { + registerItem(workItem, "live"); + started.add(workItem.workItemId); + } + + function registerItem(workItem: WorkItem, status: WorkItemStatus): void { + items.set(workItem.workItemId, { + workItemId: workItem.workItemId, + kind: workItem.kind, + name: workItem.name, + flow: [...workItem.flow], + state: { ...workItem.state }, + metadata: { ...workItem.metadata }, + }); + statuses.set(workItem.workItemId, status); + states.set(workItem.workItemId, { ...workItem.state }); + } + + function getStatus(workItemId: string): WorkItemStatus { + return statuses.get(workItemId) ?? "draft"; + } + + function isTerminal(status: WorkItemStatus): boolean { + return status === "completed" || status === "failed"; + } + + function depsSatisfied(workItemId: string): boolean { + const deps = dependencies.get(workItemId) ?? []; + return deps.every((dep) => isTerminal(getStatus(dep.workItemId))); + } + + function isRunnable(workItemId: string): boolean { + const status = getStatus(workItemId); + if (status !== "live") { + return false; + } + return depsSatisfied(workItemId); + } + + function enqueueIfReady(workItemId: string): void { + if (!isRunnable(workItemId) || queued.has(workItemId)) { + return; + } + const item = items.get(workItemId); + if (item === undefined) { + return; + } + queued.add(workItemId); + readyQueue.push({ + ...item, + state: { ...(states.get(workItemId) ?? item.state) }, + }); + } + + function scanForReady(): void { + for (const workItemId of items.keys()) { + enqueueIfReady(workItemId); + } + } + + function reevaluatePaused(): void { + for (const [workItemId, status] of statuses) { + if (status === "paused" && depsSatisfied(workItemId)) { + statuses.set(workItemId, "live"); + queued.delete(workItemId); + enqueueIfReady(workItemId); + } + } + } + + function onTerminal(workItemId: string): void { + queued.delete(workItemId); + reevaluatePaused(); + scanForReady(); + } + + for (const workItem of initialWorkItems) { + enqueueIfReady(workItem.workItemId); + } + + const source: GraphMemoryWorkItemSource = { + completed, + failed, + paused, + states, + drafts, + started, + dependencies, + items, + statuses, + startedOrder, + async *watchWorkItems() { + while (!aborted) { + scanForReady(); + while (readyQueue.length > 0) { + const workItem = readyQueue.shift(); + if (workItem === undefined) { + break; + } + yield workItem; + } + await delay(POLL_MS); + } + }, + async completeWorkItem(workItemId: string) { + completed.push(workItemId); + statuses.set(workItemId, "completed"); + onTerminal(workItemId); + }, + async failWorkItem(workItemId: string, error: string) { + failed.push({ workItemId, error }); + statuses.set(workItemId, "failed"); + onTerminal(workItemId); + }, + async pauseWorkItem(workItemId: string) { + paused.push(workItemId); + statuses.set(workItemId, "paused"); + queued.delete(workItemId); + const queuedIndex = readyQueue.findIndex((item) => item.workItemId === workItemId); + if (queuedIndex >= 0) { + readyQueue.splice(queuedIndex, 1); + } + }, + async setState(workItemId: string, state: Record) { + states.set(workItemId, state); + const item = items.get(workItemId); + if (item !== undefined) { + items.set(workItemId, { ...item, state: { ...state } }); + } + }, + async createDraftWorkItem(input: CreateDraftWorkItemInput) { + const workItemId = `draft-${nextDraftId}`; + nextDraftId += 1; + drafts.set(workItemId, input); + registerItem( + { + workItemId, + kind: input.kind, + name: input.name, + flow: [...(input.flow ?? [])], + state: input.state ?? {}, + metadata: input.metadata ?? {}, + }, + "draft", + ); + return workItemId; + }, + async startWorkItem(workItemId: string) { + started.add(workItemId); + startedOrder.push(workItemId); + statuses.set(workItemId, "live"); + enqueueIfReady(workItemId); + }, + async setDependency(workItemId: string, dependsOnWorkItemId: string, type = "blocks") { + const edges = dependencies.get(workItemId) ?? []; + edges.push({ workItemId: dependsOnWorkItemId, type }); + dependencies.set(workItemId, edges); + }, + async getDependencies(workItemId: string) { + return dependencies.get(workItemId) ?? []; + }, + async getWorkItemStatus(workItemId: string) { + return getStatus(workItemId); + }, + abort() { + aborted = true; + }, + }; + + return source; +} + +function delay(ms: number): Promise { + return new Promise((resolve) => { + setTimeout(resolve, ms); + }); +} diff --git a/orchestrator-v2/packages/orchestrator/src/index.ts b/orchestrator-v2/packages/orchestrator/src/index.ts index 92808e1..5cf0b63 100644 --- a/orchestrator-v2/packages/orchestrator/src/index.ts +++ b/orchestrator-v2/packages/orchestrator/src/index.ts @@ -6,4 +6,4 @@ export type { OrchestratorStartOptions, WorkItemMapper, } from "./orchestrator.js"; -export type { OrchestratorOptions, Scheduler } from "./types.js"; +export type { OrchestratorOptions } from "./types.js"; diff --git a/orchestrator-v2/packages/orchestrator/src/orchestrator.ts b/orchestrator-v2/packages/orchestrator/src/orchestrator.ts index 5cf3c02..9aa84c0 100644 --- a/orchestrator-v2/packages/orchestrator/src/orchestrator.ts +++ b/orchestrator-v2/packages/orchestrator/src/orchestrator.ts @@ -75,7 +75,7 @@ export class Orchestrator { const registry = new PeerRegistry({ heartbeatTimeoutMs, maxInFlightPerPeer }); const tracker = new DispatchTracker(); const results = new ResultHandler(workItemSource, tracker, registry); - const router = new RpcRouter(workItemSource, options.scheduler, results); + const router = new RpcRouter(workItemSource, results); const acks = new DispatchAckHandler(workItemSource, tracker, registry); const disconnectCleanup = peer.onPeerDisconnect((connected) => { diff --git a/orchestrator-v2/packages/orchestrator/src/rpc-router.ts b/orchestrator-v2/packages/orchestrator/src/rpc-router.ts index bf72f6b..6c537f4 100644 --- a/orchestrator-v2/packages/orchestrator/src/rpc-router.ts +++ b/orchestrator-v2/packages/orchestrator/src/rpc-router.ts @@ -1,14 +1,12 @@ -import type { WorkItemSource } from "@bifrost-ai/interfaces-work"; +import type { CreateDraftWorkItemInput, WorkItemSource } from "@bifrost-ai/interfaces-work"; import type { ConnectedPeer, FramePayload } from "@bifrost-ai/protocol"; import { sendRpcError, sendRpcResponse } from "./dispatcher.js"; import type { ResultHandler } from "./result-handler.js"; -import type { Scheduler } from "./types.js"; export class RpcRouter { constructor( private readonly workItemSource: WorkItemSource, - private readonly scheduler: Scheduler, private readonly results: ResultHandler, ) {} @@ -18,9 +16,6 @@ export class RpcRouter { } void this.route(peer, payload.id, payload.method, payload.params).catch((error) => { - // Backstop for the whole RPC surface: any handler that throws before it - // answers still gets a response back to the runner, so a bug in one path - // can't hang the runner. Per-handler catches below just refine the code. console.error(`Failed to route RPC ${payload.method}:`, error); sendRpcError(peer, payload.id, "INTERNAL_ERROR", error); }); @@ -45,8 +40,20 @@ export class RpcRouter { case "workItemSource.setState": await this.handleSetState(peer, requestId, params); return; - case "scheduler.call": - await this.handleSchedulerCall(peer, requestId, params); + case "workItemSource.createDraftWorkItem": + await this.handleCreateDraftWorkItem(peer, requestId, params); + return; + case "workItemSource.startWorkItem": + await this.handleStartWorkItem(peer, requestId, params); + return; + case "workItemSource.setDependency": + await this.handleSetDependency(peer, requestId, params); + return; + case "workItemSource.getDependencies": + await this.handleGetDependencies(peer, requestId, params); + return; + case "workItemSource.getWorkItemStatus": + await this.handleGetWorkItemStatus(peer, requestId, params); return; default: sendRpcError(peer, requestId, "METHOD_NOT_FOUND", `Unknown method: ${method}`); @@ -72,22 +79,107 @@ export class RpcRouter { } } - private async handleSchedulerCall( + private async handleCreateDraftWorkItem( + peer: ConnectedPeer, + requestId: string, + params: unknown, + ): Promise { + const parsed = readCreateDraftParams(params); + if (parsed === null) { + sendRpcError(peer, requestId, "INVALID_PARAMS", "input is required"); + return; + } + + try { + const workItemId = await this.workItemSource.createDraftWorkItem(parsed.input); + sendRpcResponse(peer, requestId, { workItemId }); + } catch (error) { + sendRpcError(peer, requestId, "SOURCE_ERROR", error); + } + } + + private async handleStartWorkItem( + peer: ConnectedPeer, + requestId: string, + params: unknown, + ): Promise { + const workItemId = readWorkItemId(params); + if (workItemId === null) { + sendRpcError(peer, requestId, "INVALID_PARAMS", "workItemId is required"); + return; + } + + try { + await this.workItemSource.startWorkItem(workItemId); + sendRpcResponse(peer, requestId, { ok: true }); + } catch (error) { + sendRpcError(peer, requestId, "SOURCE_ERROR", error); + } + } + + private async handleSetDependency( peer: ConnectedPeer, requestId: string, params: unknown, ): Promise { - const parsed = readSchedulerParams(params); + const parsed = readSetDependencyParams(params); if (parsed === null) { - sendRpcError(peer, requestId, "INVALID_PARAMS", "method and args are required"); + sendRpcError( + peer, + requestId, + "INVALID_PARAMS", + "workItemId and dependsOnWorkItemId are required", + ); return; } try { - const result = await this.scheduler.call(parsed.method, parsed.args); - sendRpcResponse(peer, requestId, result); + await this.workItemSource.setDependency( + parsed.workItemId, + parsed.dependsOnWorkItemId, + parsed.type, + ); + sendRpcResponse(peer, requestId, { ok: true }); } catch (error) { - sendRpcError(peer, requestId, "SCHEDULER_ERROR", error); + sendRpcError(peer, requestId, "SOURCE_ERROR", error); + } + } + + private async handleGetDependencies( + peer: ConnectedPeer, + requestId: string, + params: unknown, + ): Promise { + const workItemId = readWorkItemId(params); + if (workItemId === null) { + sendRpcError(peer, requestId, "INVALID_PARAMS", "workItemId is required"); + return; + } + + try { + const dependencies = await this.workItemSource.getDependencies(workItemId); + sendRpcResponse(peer, requestId, dependencies); + } catch (error) { + sendRpcError(peer, requestId, "SOURCE_ERROR", error); + } + } + + private async handleGetWorkItemStatus( + peer: ConnectedPeer, + requestId: string, + params: unknown, + ): Promise { + const workItemId = readWorkItemId(params); + if (workItemId === null) { + sendRpcError(peer, requestId, "INVALID_PARAMS", "workItemId is required"); + return; + } + + try { + const status = await this.workItemSource.getWorkItemStatus(workItemId); + sendRpcResponse(peer, requestId, { status }); + } catch (error) { + sendRpcError(peer, requestId, "SOURCE_ERROR", error); } } } @@ -111,13 +203,68 @@ function readSetStateParams( }; } -function readSchedulerParams(params: unknown): { method: string; args: unknown } | null { +function readWorkItemId(params: unknown): string | null { if (params === null || typeof params !== "object") { return null; } - const record = params as { method?: unknown; args?: unknown }; - if (typeof record.method !== "string") { + const record = params as { workItemId?: unknown }; + return typeof record.workItemId === "string" ? record.workItemId : null; +} + +function readCreateDraftParams(params: unknown): { input: CreateDraftWorkItemInput } | null { + if (params === null || typeof params !== "object") { return null; } - return { method: record.method, args: record.args }; + const record = params as { input?: unknown }; + if (record.input === null || typeof record.input !== "object") { + return null; + } + const input = record.input as Partial; + if (typeof input.kind !== "string") { + return null; + } + if (typeof input.name !== "string") { + return null; + } + if ( + input.flow !== undefined && + (!Array.isArray(input.flow) || + !input.flow.every((entry) => typeof entry === "string" && entry.length > 0)) + ) { + return null; + } + if ( + input.state !== undefined && + (input.state === null || typeof input.state !== "object" || Array.isArray(input.state)) + ) { + return null; + } + if ( + input.metadata !== undefined && + (input.metadata === null || typeof input.metadata !== "object" || Array.isArray(input.metadata)) + ) { + return null; + } + return { input: record.input as CreateDraftWorkItemInput }; +} + +function readSetDependencyParams( + params: unknown, +): { workItemId: string; dependsOnWorkItemId: string; type?: string } | null { + if (params === null || typeof params !== "object") { + return null; + } + const record = params as { + workItemId?: unknown; + dependsOnWorkItemId?: unknown; + type?: unknown; + }; + if (typeof record.workItemId !== "string" || typeof record.dependsOnWorkItemId !== "string") { + return null; + } + return { + workItemId: record.workItemId, + dependsOnWorkItemId: record.dependsOnWorkItemId, + ...(typeof record.type === "string" ? { type: record.type } : {}), + }; } diff --git a/orchestrator-v2/packages/orchestrator/src/test-helpers.ts b/orchestrator-v2/packages/orchestrator/src/test-helpers.ts index 71ccbba..1772b9d 100644 --- a/orchestrator-v2/packages/orchestrator/src/test-helpers.ts +++ b/orchestrator-v2/packages/orchestrator/src/test-helpers.ts @@ -3,12 +3,12 @@ import type { WorkItem, WorkItemDependency, WorkItemSource, + WorkItemStatus, } from "@bifrost-ai/interfaces-work"; import type { FramePayload, PeerIdentity, RunnerPeer } from "@bifrost-ai/protocol"; import { createRunnerPeer, generateKeyPair } from "@bifrost-ai/protocol"; import { Orchestrator } from "./orchestrator.js"; -import type { Scheduler } from "./types.js"; export type MemoryWorkItemSource = WorkItemSource & { completed: string[]; @@ -36,6 +36,9 @@ export function createMemoryWorkItemSource(workItems: WorkItem[]): MemoryWorkIte const drafts = new Map(); const started = new Set(workItems.map((workItem) => workItem.workItemId)); const dependencies = new Map(); + const statuses = new Map( + workItems.map((workItem) => [workItem.workItemId, "live"]), + ); let nextDraftId = 1; return { @@ -53,12 +56,15 @@ export function createMemoryWorkItemSource(workItems: WorkItem[]): MemoryWorkIte }, async completeWorkItem(workItemId: string) { completed.push(workItemId); + statuses.set(workItemId, "completed"); }, async failWorkItem(workItemId: string, error: string) { failed.push({ workItemId, error }); + statuses.set(workItemId, "failed"); }, async pauseWorkItem(workItemId: string) { paused.push(workItemId); + statuses.set(workItemId, "paused"); }, async setState(workItemId: string, state: Record) { states.set(workItemId, state); @@ -67,10 +73,12 @@ export function createMemoryWorkItemSource(workItems: WorkItem[]): MemoryWorkIte const workItemId = `draft-${nextDraftId}`; nextDraftId += 1; drafts.set(workItemId, input); + statuses.set(workItemId, "draft"); return workItemId; }, async startWorkItem(workItemId: string) { started.add(workItemId); + statuses.set(workItemId, "live"); }, async setDependency(workItemId: string, dependsOnWorkItemId: string, type = "blocks") { const edges = dependencies.get(workItemId) ?? []; @@ -80,13 +88,8 @@ export function createMemoryWorkItemSource(workItems: WorkItem[]): MemoryWorkIte async getDependencies(workItemId: string) { return dependencies.get(workItemId) ?? []; }, - }; -} - -export function createNoopScheduler(): Scheduler { - return { - async call() { - return { ok: true }; + async getWorkItemStatus(workItemId: string) { + return statuses.get(workItemId) ?? "draft"; }, }; } @@ -130,8 +133,7 @@ export async function waitFor( export async function startOrchestratorInBackground(options: { orchestratorIdentity: PeerIdentity; authorizedRunners: Map; - workItemSource: MemoryWorkItemSource; - scheduler?: Scheduler; + workItemSource: WorkItemSource; maxInFlightPerPeer?: number; }): Promise<{ abort: () => void; @@ -148,7 +150,6 @@ export async function startOrchestratorInBackground(options: { const handle = await orchestrator.start({ identity: options.orchestratorIdentity, authorizedRunners: options.authorizedRunners, - scheduler: options.scheduler ?? createNoopScheduler(), maxInFlightPerPeer: options.maxInFlightPerPeer, abortSignal: abortController.signal, }); @@ -292,9 +293,13 @@ function waitForRpcResponse(runner: RunnerPeer, id: string): Promise { export function sampleWorkItem(workItemId: string): WorkItem { return { workItemId, - kind: "echo", + kind: "script", + name: "echo", flow: [], state: {}, metadata: {}, }; } + +export { createGraphMemoryWorkItemSource } from "./graph-memory-work-item-source.js"; +export type { GraphMemoryWorkItemSource } from "./graph-memory-work-item-source.js"; diff --git a/orchestrator-v2/packages/orchestrator/src/types.ts b/orchestrator-v2/packages/orchestrator/src/types.ts index b1d7ec7..21206bf 100644 --- a/orchestrator-v2/packages/orchestrator/src/types.ts +++ b/orchestrator-v2/packages/orchestrator/src/types.ts @@ -2,15 +2,10 @@ import type { WorkItemSource } from "@bifrost-ai/interfaces-work"; import type { FramePayload, PeerIdentity } from "@bifrost-ai/protocol"; import type { KeyObject } from "node:crypto"; -export type Scheduler = { - call(method: string, params: unknown): Promise; -}; - export type OrchestratorOptions = { identity: PeerIdentity; authorizedRunners: ReadonlyMap; workItemSource: WorkItemSource; - scheduler: Scheduler; host?: string; port?: number; heartbeatTimeoutMs?: number; diff --git a/orchestrator-v2/packages/runner/README.md b/orchestrator-v2/packages/runner/README.md index ed9ca00..11dc7e2 100644 --- a/orchestrator-v2/packages/runner/README.md +++ b/orchestrator-v2/packages/runner/README.md @@ -46,7 +46,6 @@ await runner.start(); ### Lower-level exports -- `registerScriptAgent(runner, name, fn)` — adapt legacy `{ workItem, cwd, setState }` scripts - `composeStack`, `executeScriptStack`, `resolveStack` — in-process stack execution - `createScriptContext` — build RPC-backed `ScriptContext` for a dispatch - `failOnError`, `FAIL_ON_ERROR_DECORATOR` — built-in error-handling convention diff --git a/orchestrator-v2/packages/runner/src/conventions/complete-on-success.ts b/orchestrator-v2/packages/runner/src/conventions/complete-on-success.ts new file mode 100644 index 0000000..d4ac716 --- /dev/null +++ b/orchestrator-v2/packages/runner/src/conventions/complete-on-success.ts @@ -0,0 +1,11 @@ +import type { DecoratorFn } from "@bifrost-ai/interfaces-work"; + +export const COMPLETE_ON_SUCCESS_DECORATOR = "completeOnSuccess"; + +export const completeOnSuccess: DecoratorFn = async (workItem, ctx, next) => { + await next(); + const status = await ctx.workItemSource.getWorkItemStatus(workItem.workItemId); + if (status === "live") { + await ctx.workItemSource.completeWorkItem(workItem.workItemId); + } +}; diff --git a/orchestrator-v2/packages/runner/src/conventions/fail-on-error.ts b/orchestrator-v2/packages/runner/src/conventions/fail-on-error.ts index 3f559ce..ba8650b 100644 --- a/orchestrator-v2/packages/runner/src/conventions/fail-on-error.ts +++ b/orchestrator-v2/packages/runner/src/conventions/fail-on-error.ts @@ -2,11 +2,12 @@ import type { DecoratorFn } from "@bifrost-ai/interfaces-work"; export const FAIL_ON_ERROR_DECORATOR = "failOnError"; -export const failOnError: DecoratorFn = async (_workItem, _ctx, next) => { +export const failOnError: DecoratorFn = async (workItem, ctx, next) => { try { - return await next(); + await next(); } catch (error) { + console.error("Script execution failed:", error); const message = error instanceof Error ? error.message : String(error); - return { outcome: "failed", message }; + await ctx.workItemSource.failWorkItem(workItem.workItemId, message); } }; diff --git a/orchestrator-v2/packages/runner/src/conventions/lifecycle-decorators.spec.ts b/orchestrator-v2/packages/runner/src/conventions/lifecycle-decorators.spec.ts new file mode 100644 index 0000000..c4ba3c9 --- /dev/null +++ b/orchestrator-v2/packages/runner/src/conventions/lifecycle-decorators.spec.ts @@ -0,0 +1,146 @@ +import type { DecoratorFn, ScriptContext, WorkItem } from "@bifrost-ai/interfaces-work"; +import { describe, expect } from "vite-plus/test"; +import test from "vitest-gwt"; + +import { completeOnSuccess } from "./complete-on-success.js"; +import { failOnError } from "./fail-on-error.js"; + +type Context = { + workItem: WorkItem; + ctx: ScriptContext; + workItemSource: { + completed: string[]; + failed: Array<{ workItemId: string; error: string }>; + status: string; + }; + error: Error | null; +}; + +const baseWorkItem = (): WorkItem => ({ + workItemId: "wi-1", + kind: "script", + name: "task", + flow: [], + state: {}, + metadata: {}, +}); + +function makeCtx(workItemSource: Context["workItemSource"]): ScriptContext { + return { + cwd: "/tmp", + data: { + get() { + return { + register() {}, + get() { + return undefined; + }, + has() { + return false; + }, + }; + }, + }, + workItemSource: { + async completeWorkItem(workItemId: string) { + workItemSource.completed.push(workItemId); + workItemSource.status = "completed"; + }, + async failWorkItem(workItemId: string, error: string) { + workItemSource.failed.push({ workItemId, error }); + workItemSource.status = "failed"; + }, + async pauseWorkItem() { + workItemSource.status = "paused"; + }, + async createDraftWorkItem() { + return "draft-1"; + }, + async startWorkItem() {}, + async setDependency() {}, + async getDependencies() { + return []; + }, + async getWorkItemStatus() { + return workItemSource.status as "live"; + }, + async setState() {}, + }, + async setState() {}, + }; +} + +describe("lifecycle decorators", () => { + test("failOnError records failure via workItemSource", { + given: { failing_script_context }, + when: { running_fail_on_error }, + then: { failure_recorded }, + }); + + test("completeOnSuccess completes when status is still live", { + given: { succeeding_script_context }, + when: { running_complete_on_success }, + then: { work_item_completed }, + }); + + test("completeOnSuccess skips complete when status is paused", { + given: { paused_script_context }, + when: { running_complete_on_success }, + then: { work_item_not_completed }, + }); +}); + +function failing_script_context(this: Context) { + this.workItem = baseWorkItem(); + this.workItemSource = { completed: [], failed: [], status: "live" }; + this.ctx = makeCtx(this.workItemSource); +} + +function succeeding_script_context(this: Context) { + this.workItem = baseWorkItem(); + this.workItemSource = { completed: [], failed: [], status: "live" }; + this.ctx = makeCtx(this.workItemSource); +} + +function paused_script_context(this: Context) { + this.workItem = baseWorkItem(); + this.workItemSource = { completed: [], failed: [], status: "paused" }; + this.ctx = makeCtx(this.workItemSource); +} + +async function running_fail_on_error(this: Context) { + this.error = null; + const decorator = failOnError as DecoratorFn; + try { + await decorator(this.workItem, this.ctx, async () => { + throw new Error("boom"); + }); + } catch (error) { + this.error = error as Error; + } +} + +async function running_complete_on_success(this: Context) { + this.error = null; + const decorator = completeOnSuccess as DecoratorFn; + try { + await decorator(this.workItem, this.ctx, async () => undefined); + } catch (error) { + this.error = error as Error; + } +} + +function failure_recorded(this: Context) { + expect(this.error).toBeNull(); + expect(this.workItemSource.failed).toEqual([{ workItemId: "wi-1", error: "boom" }]); +} + +function work_item_completed(this: Context) { + expect(this.error).toBeNull(); + expect(this.workItemSource.completed).toEqual(["wi-1"]); +} + +function work_item_not_completed(this: Context) { + expect(this.error).toBeNull(); + expect(this.workItemSource.completed).toEqual([]); +} diff --git a/orchestrator-v2/packages/runner/src/dispatch-handler.ts b/orchestrator-v2/packages/runner/src/dispatch-handler.ts index 9c0d1d1..d4509f2 100644 --- a/orchestrator-v2/packages/runner/src/dispatch-handler.ts +++ b/orchestrator-v2/packages/runner/src/dispatch-handler.ts @@ -66,40 +66,6 @@ async function handleDispatch>( sendRpcResponse(peer, payload.id, { accepted: true }); - try { - const { workItem: liveWorkItem, ctx } = createScriptContext(workItem, stack.rpc, stack.data); - const result = await executeScriptStack(liveWorkItem, ctx, resolved); - - switch (result.outcome) { - case "completed": - await stack.rpc.call("workItem.complete", { workItemId: workItem.workItemId }); - break; - case "failed": - await stack.rpc.call("workItem.fail", { - workItemId: workItem.workItemId, - message: result.message ?? "failed", - }); - break; - case "paused": - await stack.rpc.call("workItem.pause", { workItemId: workItem.workItemId }); - break; - default: - await stack.rpc.call("workItem.fail", { - workItemId: workItem.workItemId, - message: `Unknown outcome: ${String(result.outcome)}`, - }); - break; - } - } catch (error) { - const message = error instanceof Error ? error.message : String(error); - console.error("Dispatch execution failed:", error); - try { - await stack.rpc.call("workItem.fail", { - workItemId: workItem.workItemId, - message, - }); - } catch (failError) { - console.error("Failed to report work item failure:", failError); - } - } + const { workItem: liveWorkItem, ctx } = createScriptContext(workItem, stack.rpc, stack.data); + await executeScriptStack(liveWorkItem, ctx, resolved); } diff --git a/orchestrator-v2/packages/runner/src/index.ts b/orchestrator-v2/packages/runner/src/index.ts index 597ae18..957cc7a 100644 --- a/orchestrator-v2/packages/runner/src/index.ts +++ b/orchestrator-v2/packages/runner/src/index.ts @@ -1,17 +1,17 @@ export { Runner } from "./runner.js"; export { Registry } from "./registry.js"; export { createDataRegistry } from "./data-registry.js"; -export { registerScriptAgent, type LegacyScriptFn, type ScriptFn } from "./script-agent.js"; +export type { ScriptFn } from "@bifrost-ai/interfaces-work"; export { discoverConfigPath, loadRunnerConfig, resolveRunnerOptions } from "./config-loader.js"; export { createScriptContext } from "./script-context.js"; -export { - composeStack, - executeScriptStack, - normalizeScriptResult, - resolveStack, -} from "./script-stack.js"; +export { composeStack, executeScriptStack, resolveStack } from "./script-stack.js"; +export { createRpcWorkItemSourceClient } from "./work-item-source-client.js"; export { createRpcClient } from "./rpc-client.js"; export type { RpcClient } from "./rpc-client.js"; +export { + COMPLETE_ON_SUCCESS_DECORATOR, + completeOnSuccess, +} from "./conventions/complete-on-success.js"; export { FAIL_ON_ERROR_DECORATOR, failOnError } from "./conventions/fail-on-error.js"; export type { IdentityConfig, diff --git a/orchestrator-v2/packages/runner/src/runner.spec.ts b/orchestrator-v2/packages/runner/src/runner.spec.ts index 08fd6c6..8a5fc50 100644 --- a/orchestrator-v2/packages/runner/src/runner.spec.ts +++ b/orchestrator-v2/packages/runner/src/runner.spec.ts @@ -35,14 +35,15 @@ type Context = { const echoScript: ScriptFn = async (workItem, ctx) => { const message = workItem.metadata.message as string; await ctx.setState({ echoed: message }); - return { outcome: "completed" }; }; const failScript: ScriptFn = async () => { throw new Error("boom"); }; -const pauseScript: ScriptFn = async () => ({ outcome: "paused" }); +const pauseScript: ScriptFn = async (workItem, ctx) => { + await ctx.workItemSource.pauseWorkItem(workItem.workItemId); +}; describe("runner", () => { withAspect(setup_identities, teardown_runner); @@ -142,7 +143,8 @@ function work_item_source_with_echo(this: Context) { this.workItemSource = createMemoryWorkItemSource([ { workItemId: "work-item-1", - kind: "echo", + kind: "script", + name: "echo", flow: [], state: {}, metadata: { message: "hello" }, @@ -154,7 +156,8 @@ function work_item_source_with_fail(this: Context) { this.workItemSource = createMemoryWorkItemSource([ { workItemId: "work-item-fail", - kind: "fail", + kind: "script", + name: "fail", flow: [], state: {}, metadata: {}, @@ -166,7 +169,8 @@ function work_item_source_with_pause(this: Context) { this.workItemSource = createMemoryWorkItemSource([ { workItemId: "work-item-pause", - kind: "pause", + kind: "script", + name: "pause", flow: [], state: {}, metadata: {}, diff --git a/orchestrator-v2/packages/runner/src/runner.ts b/orchestrator-v2/packages/runner/src/runner.ts index 426d49d..516cc56 100644 --- a/orchestrator-v2/packages/runner/src/runner.ts +++ b/orchestrator-v2/packages/runner/src/runner.ts @@ -1,22 +1,20 @@ -import type { - DataRegistry, - DecoratorFn, - ScriptFn, - WorkItemResult, -} from "@bifrost-ai/interfaces-work"; +import type { DataRegistry, DecoratorFn, ScriptFn } from "@bifrost-ai/interfaces-work"; import { createRunnerPeer, type RunnerPeer } from "@bifrost-ai/protocol"; import { resolveRunnerOptions } from "./config-loader.js"; +import { + COMPLETE_ON_SUCCESS_DECORATOR, + completeOnSuccess, +} from "./conventions/complete-on-success.js"; import { FAIL_ON_ERROR_DECORATOR, failOnError } from "./conventions/fail-on-error.js"; import { registerDispatchHandler } from "./dispatch-handler.js"; import { startHeartbeat, type HeartbeatHandle } from "./heartbeat.js"; import { createRpcClient } from "./rpc-client.js"; import { createDataRegistry } from "./data-registry.js"; import { Registry } from "./registry.js"; -import { registerScriptAgent as adaptLegacyScript, type LegacyScriptFn } from "./script-agent.js"; import type { RunnerOptions } from "./types.js"; -const DEFAULT_CONVENTIONS = [FAIL_ON_ERROR_DECORATOR] as const; +const DEFAULT_CONVENTIONS = [COMPLETE_ON_SUCCESS_DECORATOR, FAIL_ON_ERROR_DECORATOR] as const; export class Runner = Record> { private readonly options: RunnerOptions; @@ -33,6 +31,7 @@ export class Runner = Record); this.registerDecorator(FAIL_ON_ERROR_DECORATOR, failOnError as DecoratorFn); + this.registerDecorator(COMPLETE_ON_SUCCESS_DECORATOR, completeOnSuccess as DecoratorFn); } registerScript(kind: string, fn: ScriptFn): void { @@ -52,10 +51,6 @@ export class Runner = Record = Record) => Promise; -}) => Promise | WorkItemResult; - -export type { ScriptFn }; - -export function registerScriptAgent>( - runner: Runner, - name: string, - fn: LegacyScriptFn, -): void { - const script: ScriptFn = async (workItem, ctx) => - fn({ workItem, cwd: ctx.cwd, setState: ctx.setState }); - - runner.registerScript(name, script); -} diff --git a/orchestrator-v2/packages/runner/src/script-context.ts b/orchestrator-v2/packages/runner/src/script-context.ts index f017ddb..97ea1dd 100644 --- a/orchestrator-v2/packages/runner/src/script-context.ts +++ b/orchestrator-v2/packages/runner/src/script-context.ts @@ -1,6 +1,7 @@ import type { DataRegistry, ScriptContext, WorkItem } from "@bifrost-ai/interfaces-work"; import type { RpcClient } from "./rpc-client.js"; +import { createRpcWorkItemSourceClient } from "./work-item-source-client.js"; export function resolveScriptCwd(workItem: WorkItem): string { return typeof workItem.state.workingDir === "string" && workItem.state.workingDir.length > 0 @@ -12,6 +13,7 @@ export function createLiveWorkItem(workItem: WorkItem): WorkItem { return { workItemId: workItem.workItemId, kind: workItem.kind, + name: workItem.name, flow: [...workItem.flow], metadata: workItem.metadata, state: { ...workItem.state }, @@ -28,12 +30,14 @@ export function createScriptContext>( const ctx: ScriptContext = { cwd: resolveScriptCwd(liveWorkItem), data, + workItemSource: createRpcWorkItemSourceClient(rpc), async setState(nextState) { - Object.assign(liveWorkItem.state, nextState); + const mergedState = { ...liveWorkItem.state, ...nextState }; await rpc.call("workItemSource.setState", { workItemId: workItem.workItemId, - state: liveWorkItem.state, + state: mergedState, }); + Object.assign(liveWorkItem.state, nextState); }, }; diff --git a/orchestrator-v2/packages/runner/src/script-stack.spec.ts b/orchestrator-v2/packages/runner/src/script-stack.spec.ts index 8f347f4..e51fe1d 100644 --- a/orchestrator-v2/packages/runner/src/script-stack.spec.ts +++ b/orchestrator-v2/packages/runner/src/script-stack.spec.ts @@ -3,11 +3,7 @@ import { describe, expect } from "vite-plus/test"; import test from "vitest-gwt"; import { Registry } from "./registry.js"; -import { - executeScriptStack, - normalizeScriptResult, - resolveStack, -} from "./script-stack.js"; +import { executeScriptStack, resolveStack } from "./script-stack.js"; type Context = { workItem: WorkItem; @@ -19,7 +15,8 @@ type Context = { const baseWorkItem = (): WorkItem => ({ workItemId: "wi-1", - kind: "hunt", + kind: "script", + name: "hunt", flow: [], state: {}, metadata: {}, @@ -40,6 +37,23 @@ const scriptContext = { }; }, }, + workItemSource: { + async completeWorkItem() {}, + async failWorkItem() {}, + async pauseWorkItem() {}, + async createDraftWorkItem() { + return "draft-1"; + }, + async startWorkItem() {}, + async setDependency() {}, + async getDependencies() { + return []; + }, + async getWorkItemStatus() { + return "live" as const; + }, + async setState() {}, + }, async setState() {}, }; @@ -79,16 +93,6 @@ describe("script-stack", () => { when: { resolving_unknown_decorator }, then: { decorator_error_thrown }, }); - - test("normalizeScriptResult treats WorkItemResult as-is", { - when: { normalizing_work_item_result }, - then: { result_is_preserved }, - }); - - test("normalizeScriptResult treats unknown as completed", { - when: { normalizing_unknown }, - then: { outcome_is_completed }, - }); }); function a_script_registry_with_hunt(this: Context) { @@ -97,17 +101,16 @@ function a_script_registry_with_hunt(this: Context) { this.decorators = new Registry(); this.scripts.register("hunt", async () => { this.result = "meat"; - return this.result; }); } async function executing_without_conventions(this: Context) { const stack = resolveStack(this.workItem, this.scripts, this.decorators, []); - this.result = await executeScriptStack(this.workItem, scriptContext, stack); + await executeScriptStack(this.workItem, scriptContext, stack); } function script_ran(this: Context) { - expect(this.result).toEqual({ outcome: "completed" }); + expect(this.result).toBe("meat"); } function nested_decorators(this: Context) { @@ -119,7 +122,6 @@ function nested_decorators(this: Context) { this.scripts.register("hunt", async () => { order.push("script"); - return "done"; }); this.decorators.register("outer", async (_wi, _ctx, next) => { @@ -161,7 +163,7 @@ function skip_decorator(this: Context) { state.scriptRan = true; }); - this.decorators.register("skip", async () => "skipped"); + this.decorators.register("skip", async () => undefined); } async function executing_short_circuit(this: Context) { @@ -185,14 +187,14 @@ function flaky_script_and_retry_decorator(this: Context) { if (state.attempts < 3) { throw new Error("not yet"); } - return "ok"; }); this.decorators.register("retry", async (_wi, _ctx, next) => { let tries = 0; while (true) { try { - return await next(); + await next(); + return; } catch (error) { if (++tries >= 3) { throw error; @@ -234,7 +236,7 @@ function script_without_decorator(this: Context) { this.workItem = { ...baseWorkItem(), flow: ["missing"] }; this.scripts = new Registry(); this.decorators = new Registry(); - this.scripts.register("hunt", async () => "ok"); + this.scripts.register("hunt", async () => undefined); } function resolving_unknown_decorator(this: Context) { @@ -249,19 +251,3 @@ function resolving_unknown_decorator(this: Context) { function decorator_error_thrown(this: Context) { expect(this.error?.message).toBe("Unknown decorator: missing"); } - -function normalizing_work_item_result(this: Context) { - this.result = normalizeScriptResult({ outcome: "paused" }); -} - -function result_is_preserved(this: Context) { - expect(this.result).toEqual({ outcome: "paused" }); -} - -function normalizing_unknown(this: Context) { - this.result = normalizeScriptResult(undefined); -} - -function outcome_is_completed(this: Context) { - expect(this.result).toEqual({ outcome: "completed" }); -} diff --git a/orchestrator-v2/packages/runner/src/script-stack.ts b/orchestrator-v2/packages/runner/src/script-stack.ts index 73f5bf7..b2dd957 100644 --- a/orchestrator-v2/packages/runner/src/script-stack.ts +++ b/orchestrator-v2/packages/runner/src/script-stack.ts @@ -1,11 +1,4 @@ -import type { - DecoratorFn, - ScriptContext, - ScriptFn, - WorkItem, - WorkItemResult, -} from "@bifrost-ai/interfaces-work"; -import { isWorkItemResult } from "@bifrost-ai/interfaces-work"; +import type { DecoratorFn, ScriptContext, ScriptFn, WorkItem } from "@bifrost-ai/interfaces-work"; import type { Registry } from "./registry.js"; @@ -20,9 +13,9 @@ export function resolveStack = Record>, conventions: readonly string[], ): ResolvedStack { - const script = scripts.get(workItem.kind); + const script = scripts.get(workItem.name); if (script === undefined) { - throw new Error(`Unknown script: ${workItem.kind}`); + throw new Error(`Unknown script: ${workItem.name}`); } const decoratorNames = [...conventions, ...workItem.flow]; @@ -44,7 +37,7 @@ export function composeStack>( ctx: ScriptContext, script: ScriptFn, decoratorFns: DecoratorFn[], -): () => Promise { +): () => Promise { let inner: () => Promise = () => script(workItem, ctx); for (const decorator of decoratorFns.toReversed()) { @@ -52,23 +45,16 @@ export function composeStack>( inner = () => decorator(workItem, ctx, next); } - return inner; -} - -export function normalizeScriptResult(value: unknown): WorkItemResult { - if (isWorkItemResult(value)) { - return value; - } - - return { outcome: "completed" }; + return async () => { + await inner(); + }; } export async function executeScriptStack>( workItem: WorkItem, ctx: ScriptContext, stack: ResolvedStack, -): Promise { +): Promise { const run = composeStack(workItem, ctx, stack.script, stack.decorators); - const result = await run(); - return normalizeScriptResult(result); + await run(); } diff --git a/orchestrator-v2/packages/runner/src/work-item-source-client.ts b/orchestrator-v2/packages/runner/src/work-item-source-client.ts new file mode 100644 index 0000000..cf15a4b --- /dev/null +++ b/orchestrator-v2/packages/runner/src/work-item-source-client.ts @@ -0,0 +1,60 @@ +import type { + CreateDraftWorkItemInput, + WorkItemDependency, + WorkItemSourceClient, + WorkItemStatus, +} from "@bifrost-ai/interfaces-work"; + +import type { RpcClient } from "./rpc-client.js"; + +export function createRpcWorkItemSourceClient(rpc: RpcClient): WorkItemSourceClient { + return { + async completeWorkItem(workItemId: string) { + await rpc.call("workItem.complete", { workItemId }); + }, + async failWorkItem(workItemId: string, message: string) { + await rpc.call("workItem.fail", { workItemId, message }); + }, + async pauseWorkItem(workItemId: string) { + await rpc.call("workItem.pause", { workItemId }); + }, + async createDraftWorkItem(input: CreateDraftWorkItemInput) { + const result = await rpc.call("workItemSource.createDraftWorkItem", { input }); + if ( + typeof result !== "object" || + result === null || + typeof (result as { workItemId?: unknown }).workItemId !== "string" + ) { + throw new Error("workItemSource.createDraftWorkItem returned invalid workItemId"); + } + return (result as { workItemId: string }).workItemId; + }, + async startWorkItem(workItemId: string) { + await rpc.call("workItemSource.startWorkItem", { workItemId }); + }, + async setDependency(workItemId: string, dependsOnWorkItemId: string, type?: string) { + await rpc.call("workItemSource.setDependency", { workItemId, dependsOnWorkItemId, type }); + }, + async getDependencies(workItemId: string) { + const result = await rpc.call("workItemSource.getDependencies", { workItemId }); + if (!Array.isArray(result)) { + throw new Error("workItemSource.getDependencies returned invalid dependencies"); + } + return result as WorkItemDependency[]; + }, + async getWorkItemStatus(workItemId: string) { + const result = await rpc.call("workItemSource.getWorkItemStatus", { workItemId }); + if ( + typeof result !== "object" || + result === null || + typeof (result as { status?: unknown }).status !== "string" + ) { + throw new Error("workItemSource.getWorkItemStatus returned invalid status"); + } + return (result as { status: WorkItemStatus }).status; + }, + async setState(workItemId: string, state: Record) { + await rpc.call("workItemSource.setState", { workItemId, state }); + }, + }; +} diff --git a/orchestrator-v2/packages/work-item-source-bifrost/src/bifrost-work-item-source.spec.ts b/orchestrator-v2/packages/work-item-source-bifrost/src/bifrost-work-item-source.spec.ts index 929b9cf..06d3b26 100644 --- a/orchestrator-v2/packages/work-item-source-bifrost/src/bifrost-work-item-source.spec.ts +++ b/orchestrator-v2/packages/work-item-source-bifrost/src/bifrost-work-item-source.spec.ts @@ -112,7 +112,8 @@ describe("BifrostWorkItemSource", () => { await cleanup(); expect(workItems).toHaveLength(1); - expect(workItems[0].kind).toBe("implementer"); + expect(workItems[0].kind).toBe("task"); + expect(workItems[0].name).toBe("implementer"); expect(workItems[0].flow).toEqual([]); }); @@ -241,7 +242,8 @@ describe("BifrostWorkItemSource", () => { }); const workItemId = await source.createDraftWorkItem({ - kind: "implementer", + kind: "task", + name: "implementer", metadata: { title: "New work item", description: "Do the thing" }, }); @@ -255,6 +257,13 @@ describe("BifrostWorkItemSource", () => { body: expect.stringContaining("agent:implementer"), }), ); + expect(global.fetch).toHaveBeenCalledWith( + expect.stringContaining("/api/create-rune"), + expect.objectContaining({ + method: "POST", + body: expect.stringContaining("kind:task"), + }), + ); }); }); @@ -365,7 +374,8 @@ describe("BifrostWorkItemSource", () => { expect(workItem).toBeDefined(); expect(workItem!.workItemId).toBe("rune-1"); - expect(workItem!.kind).toBe("implementer"); + expect(workItem!.kind).toBe("task"); + expect(workItem!.name).toBe("implementer"); expect(workItem!.flow).toEqual([]); expect(workItem!.metadata.description).toBe("Test description"); expect(workItem!.metadata.dependencies).toEqual([ @@ -419,4 +429,44 @@ describe("BifrostWorkItemSource", () => { expect(pollInterval).toBe(1000); }); }); + + describe("mapRuneStatus", () => { + it.each([ + ["draft", "draft"], + ["open", "live"], + ["claimed", "live"], + ["fulfilled", "completed"], + ["sealed", "failed"], + ["shattered", "failed"], + ["unknown-status", "live"], + ] as const)("maps %s to %s", async (status, expected) => { + const { BifrostWorkItemSource } = await import("./bifrost-work-item-source.js"); + expect(BifrostWorkItemSource.mapRuneStatus(status)).toBe(expected); + }); + }); + + describe("getWorkItemStatus", () => { + it.each([ + ["draft", "draft"], + ["open", "live"], + ["claimed", "live"], + ["fulfilled", "completed"], + ["sealed", "failed"], + ["shattered", "failed"], + ["unknown-status", "live"], + ] as const)("returns %s for rune status %s", async (runeStatus, expected) => { + const { source, cleanup } = await createTestSource(); + + global.fetch = vi.fn().mockResolvedValue({ + ok: true, + json: async () => runeDetailFixture({ status: runeStatus }), + }); + + const status = await source.getWorkItemStatus("rune-1"); + + await cleanup(); + + expect(status).toBe(expected); + }); + }); }); diff --git a/orchestrator-v2/packages/work-item-source-bifrost/src/bifrost-work-item-source.ts b/orchestrator-v2/packages/work-item-source-bifrost/src/bifrost-work-item-source.ts index c23a861..acd1f4c 100644 --- a/orchestrator-v2/packages/work-item-source-bifrost/src/bifrost-work-item-source.ts +++ b/orchestrator-v2/packages/work-item-source-bifrost/src/bifrost-work-item-source.ts @@ -3,6 +3,7 @@ import type { WorkItem, WorkItemDependency, WorkItemSource, + WorkItemStatus, } from "@bifrost-ai/interfaces-work"; import { BifrostHttpClient } from "./client/bifrost-http-client.js"; import { loadConfig } from "./config/config-loader.js"; @@ -163,6 +164,29 @@ export class BifrostWorkItemSource implements WorkItemSource { })); } + public async getWorkItemStatus(workItemId: string): Promise { + const client = await this.#getClient(); + const detail = await client.getRune(workItemId); + return BifrostWorkItemSource.mapRuneStatus(detail.status); + } + + public static mapRuneStatus(status: string): WorkItemStatus { + switch (status) { + case "draft": + return "draft"; + case "open": + case "claimed": + return "live"; + case "fulfilled": + return "completed"; + case "sealed": + case "shattered": + return "failed"; + default: + return "live"; + } + } + public static extractAgentName(tags: string[]): string | null { const agentTag = tags.find((tag) => tag.startsWith("agent:")); if (!agentTag) { @@ -171,6 +195,14 @@ export class BifrostWorkItemSource implements WorkItemSource { return agentTag.slice(6); } + public static extractAgentKind(tags: string[]): string { + const kindTag = tags.find((tag) => tag.startsWith("kind:")); + if (!kindTag) { + return "task"; + } + return kindTag.slice(5); + } + public static extractDecoratorNames(tags: string[]): string[] { return tags .filter((tag) => tag.startsWith("decorator:")) @@ -180,7 +212,7 @@ export class BifrostWorkItemSource implements WorkItemSource { public static buildCreateRuneRequest(input: CreateDraftWorkItemInput): CreateRuneRequest { const metadata = input.metadata ?? {}; const title = - typeof metadata.title === "string" && metadata.title.length > 0 ? metadata.title : input.kind; + typeof metadata.title === "string" && metadata.title.length > 0 ? metadata.title : input.name; const description = typeof metadata.description === "string" ? metadata.description : undefined; const priority = typeof metadata.priority === "number" ? metadata.priority : 1; const parentId = typeof metadata.parentId === "string" ? metadata.parentId : undefined; @@ -189,7 +221,8 @@ export class BifrostWorkItemSource implements WorkItemSource { const tags = Array.isArray(metadata.tags) ? [...(metadata.tags as string[])] : ([] as string[]); - tags.push(`agent:${input.kind}`); + tags.push(`agent:${input.name}`); + tags.push(`kind:${input.kind}`); for (const decorator of input.flow ?? []) { tags.push(`decorator:${decorator}`); @@ -209,7 +242,8 @@ export class BifrostWorkItemSource implements WorkItemSource { public static mapToWorkItem(rune: RuneDetail, agentName: string): WorkItem { return { workItemId: rune.id, - kind: agentName, + kind: BifrostWorkItemSource.extractAgentKind(rune.tags ?? []), + name: agentName, flow: BifrostWorkItemSource.extractDecoratorNames(rune.tags ?? []), state: { ...rune.state }, metadata: rune as unknown as Record, diff --git a/orchestrator-v2/pnpm-lock.yaml b/orchestrator-v2/pnpm-lock.yaml index 56d2c71..79ce78b 100644 --- a/orchestrator-v2/pnpm-lock.yaml +++ b/orchestrator-v2/pnpm-lock.yaml @@ -259,6 +259,38 @@ importers: '@bifrost-ai/work-item-source-bifrost': specifier: workspace:* version: link:../../packages/work-item-source-bifrost + devDependencies: + typescript: + specifier: 'catalog:' + version: 6.0.3 + + examples/lvl4: + dependencies: + '@bifrost-ai/agent-3-task': + specifier: workspace:* + version: link:../../packages/agent-3-task + '@bifrost-ai/agent-4-workflow': + specifier: workspace:* + version: link:../../packages/agent-4-workflow + '@bifrost-ai/engine-cursor': + specifier: workspace:* + version: link:../../packages/engine-cursor + '@bifrost-ai/interfaces-work': + specifier: workspace:* + version: link:../../packages/interfaces-work + '@bifrost-ai/orchestrator': + specifier: workspace:* + version: link:../../packages/orchestrator + '@bifrost-ai/runner': + specifier: workspace:* + version: link:../../packages/runner + '@bifrost-ai/work-item-source-bifrost': + specifier: workspace:* + version: link:../../packages/work-item-source-bifrost + devDependencies: + typescript: + specifier: 'catalog:' + version: 6.0.3 packages/agent-3-task: dependencies: @@ -294,6 +326,40 @@ importers: specifier: 'catalog:' version: 4.1.0(vitest@4.1.9(@types/node@25.9.4)(@vitest/browser-preview@4.1.9)(vite@8.1.2(@types/node@25.9.4)(jiti@2.7.0)(yaml@2.9.0))) + packages/agent-4-workflow: + dependencies: + '@bifrost-ai/interfaces-work': + specifier: workspace:* + version: link:../interfaces-work + '@bifrost-ai/runner': + specifier: workspace:* + version: link:../runner + devDependencies: + '@bifrost-ai/orchestrator': + specifier: workspace:* + version: link:../orchestrator + '@bifrost-ai/protocol': + specifier: workspace:* + version: link:../protocol + '@types/node': + specifier: 'catalog:' + version: 25.9.4 + '@typescript/native-preview': + specifier: 'catalog:' + version: 7.0.0-dev.20260707.2 + bumpp: + specifier: 'catalog:' + version: 11.1.0 + typescript: + specifier: 'catalog:' + version: 6.0.3 + vite-plus: + specifier: 'catalog:' + version: 0.2.1(@types/node@25.9.4)(jiti@2.7.0)(typescript@6.0.3)(vite@8.1.2(@types/node@25.9.4)(jiti@2.7.0)(yaml@2.9.0))(yaml@2.9.0) + vitest-gwt: + specifier: 'catalog:' + version: 4.1.0(vitest@4.1.9(@types/node@25.9.4)(@vitest/browser-preview@4.1.9)(vite@8.1.2(@types/node@25.9.4)(jiti@2.7.0)(yaml@2.9.0))) + packages/engine: devDependencies: '@types/node':