diff --git a/.changeset/temporal-node-activity-heartbeat-timeout.md b/.changeset/temporal-node-activity-heartbeat-timeout.md new file mode 100644 index 000000000..cc3a26561 --- /dev/null +++ b/.changeset/temporal-node-activity-heartbeat-timeout.md @@ -0,0 +1,5 @@ +--- +'@workflowbuilder/temporal': minor +--- + +`ActivityProfile` (and `nodeActivityProfiles`) gains an optional `heartbeatTimeout` key for long-running activities that need Temporal to detect a stalled or crashed worker faster than `startToCloseTimeout` alone allows. diff --git a/apps/ai-studio/README.md b/apps/ai-studio/README.md index 383877afe..78f9bfe4c 100644 --- a/apps/ai-studio/README.md +++ b/apps/ai-studio/README.md @@ -24,3 +24,30 @@ This is a sibling to `apps/demo`, not a layer over it. They share the SDK; nothi | Backend | None (pure SPA) | Required (Hono + Temporal) | | Plugin model | Plugins decorate the editor | Direct JSX composition; one slim plugin for node markers | | Dev port | 4200 | 4201 | + +## Agent Harness node + +`Agent Harness` (`ai-studio/agent-harness`) delegates a workflow step to an external +autonomous coding-agent CLI (GitHub Copilot in v1) instead of a single bounded LLM call. +Its property panel has four accordion sections: **General** (label, prompt, provider), +**Execution** (model, effort, context, idle timeout), **Tools** (tools preset, +allowed/denied tool lists), and **Advanced** (output format, MCP config, skills, +sub-agents, max budget, mutates-checkout, persist-session). + +**Functional in v1:** `prompt`, `provider` (`copilot` only), `model`, `effort`, +`idle_timeout`, `mutatesCheckout`, `agents`. + +**Rendered but NOT YET SUPPORTED in v1** (backend deferred, but per issue #147's "fields +must still render" requirement they are intentionally present rather than hidden — +this is not an oversight): `context` beyond `'fresh'` (`'shared'`/`'resume'` need session +persistence), `output_format` (no structured-output enforcement yet), `mcp`, `skills`, +`maxBudgetUsd`, `persistSession`. Each of these fields' label/placeholder in the panel +says "not yet supported" so this is visible in the UI itself, not just in this doc. + +**`idle_timeout` is in milliseconds, not seconds.** `300` means 300ms; for a 5-minute +timeout set `idle_timeout: 300000`. + +**Demo template:** "Agent Harness Demo" (`data/agent-harness-flow.ts`) — a +`trigger` → `agent-harness` → `visualize` chain that asks the agent to write a +`plan.md` outlining rate-limiting approaches and summarize the tradeoffs, exercising +both file tools and a returned text result. diff --git a/apps/ai-studio/src/data/agent-harness-flow.ts b/apps/ai-studio/src/data/agent-harness-flow.ts new file mode 100644 index 000000000..b50eb6eb2 --- /dev/null +++ b/apps/ai-studio/src/data/agent-harness-flow.ts @@ -0,0 +1,100 @@ +import type { DiagramModel, TemplateModel } from '@workflowbuilder/sdk'; + +const diagram: DiagramModel = { + name: 'Agent Harness Demo', + diagram: { + nodes: [ + { + id: 'trigger-1', + type: 'start-node', + position: { x: 0, y: 300 }, + data: { + segments: [], + isStartNode: true, + properties: { + label: 'Start', + description: 'Kicks off the agent harness demo.', + inputPrompt: `Create a file plan.md outlining three approaches to rate-limiting an HTTP API, then summarize the tradeoffs.`, + }, + type: 'ai-studio/trigger', + icon: 'Lightning', + }, + selected: false, + measured: { width: 258, height: 63 }, + dragging: false, + }, + { + id: 'agent-harness-1', + type: 'node', + position: { x: 380, y: 300 }, + data: { + segments: [], + properties: { + label: 'Agent Harness', + description: 'Delegates the task to Copilot via the agent harness CLI.', + prompt: `Create a file plan.md outlining three approaches to rate-limiting an HTTP API, then summarize the tradeoffs.`, + provider: 'copilot', + model: 'auto', + context: 'fresh', + toolsMode: 'all', + mutatesCheckout: false, + persistSession: false, + idle_timeout: 300_000, + }, + type: 'ai-studio/agent-harness', + icon: 'Terminal', + }, + selected: false, + measured: { width: 258, height: 123 }, + dragging: false, + }, + { + id: 'visualize-1', + type: 'node', + position: { x: 760, y: 300 }, + data: { + segments: [], + properties: { + label: 'Visualize', + description: 'Renders the agent output (auto-detects the format).', + mode: 'auto', + }, + type: 'ai-studio/visualize', + icon: 'Eye', + }, + selected: false, + measured: { width: 258, height: 123 }, + dragging: false, + }, + ], + edges: [ + { + source: 'trigger-1', + sourceHandle: 'source', + target: 'agent-harness-1', + targetHandle: 'target', + type: 'labelEdge', + id: 'edge-trigger-agent-harness', + data: {}, + }, + { + source: 'agent-harness-1', + sourceHandle: 'source', + target: 'visualize-1', + targetHandle: 'target', + type: 'labelEdge', + id: 'edge-agent-harness-visualize', + data: {}, + }, + ], + viewport: { x: 180, y: 150, zoom: 0.7 }, + }, + layoutDirection: 'RIGHT', +}; + +export const agentHarnessFlow: TemplateModel = { + id: 306, + name: 'Agent Harness Demo', + value: diagram, + icon: 'Terminal', +}; diff --git a/apps/ai-studio/src/data/ai-studio-templates.ts b/apps/ai-studio/src/data/ai-studio-templates.ts index a603b4fe7..cf4d2b403 100644 --- a/apps/ai-studio/src/data/ai-studio-templates.ts +++ b/apps/ai-studio/src/data/ai-studio-templates.ts @@ -1,5 +1,6 @@ import type { TemplateModel } from '@workflowbuilder/sdk'; +import { agentHarnessFlow } from './agent-harness-flow'; import { aiDebateFlow } from './ai-debate-flow'; import { contentRepurposerFlow } from './content-repurposer-flow'; import { meetingNotesFlow } from './meeting-notes-flow'; @@ -12,4 +13,5 @@ export const aiStudioTemplates: TemplateModel[] = [ contentRepurposerFlow, meetingNotesFlow, researchFlow, + agentHarnessFlow, ]; diff --git a/apps/ai-studio/src/data/node-types.ts b/apps/ai-studio/src/data/node-types.ts index 48ad1faef..758dd152a 100644 --- a/apps/ai-studio/src/data/node-types.ts +++ b/apps/ai-studio/src/data/node-types.ts @@ -1,5 +1,6 @@ import type { PaletteItemOrGroup } from '@workflowbuilder/sdk'; +import { agentHarnessPaletteItem } from '../nodes/agent-harness'; import { aiAgentPaletteItem } from '../nodes/ai-agent'; import { decisionPaletteItem } from '../nodes/decision'; import { triggerPaletteItem } from '../nodes/trigger'; @@ -9,6 +10,12 @@ export const aiStudioNodeTypes: PaletteItemOrGroup[] = [ { label: 'AI Studio', isOpen: true, - groupItems: [triggerPaletteItem, aiAgentPaletteItem, decisionPaletteItem, visualizePaletteItem], + groupItems: [ + triggerPaletteItem, + aiAgentPaletteItem, + agentHarnessPaletteItem, + decisionPaletteItem, + visualizePaletteItem, + ], }, ]; diff --git a/apps/ai-studio/src/nodes/agent-harness/default-properties-data.ts b/apps/ai-studio/src/nodes/agent-harness/default-properties-data.ts new file mode 100644 index 000000000..0fcd2c306 --- /dev/null +++ b/apps/ai-studio/src/nodes/agent-harness/default-properties-data.ts @@ -0,0 +1,15 @@ +import type { NodeDataProperties } from '@workflowbuilder/sdk'; + +import type { AgentHarnessSchema } from './schema'; + +export const defaultPropertiesData: NodeDataProperties = { + label: 'Agent Harness', + description: '', + prompt: '', + provider: 'copilot', + model: 'auto', + context: 'fresh', + toolsMode: 'none', + mutatesCheckout: false, + persistSession: false, +}; diff --git a/apps/ai-studio/src/nodes/agent-harness/index.ts b/apps/ai-studio/src/nodes/agent-harness/index.ts new file mode 100644 index 000000000..9fe8d03d9 --- /dev/null +++ b/apps/ai-studio/src/nodes/agent-harness/index.ts @@ -0,0 +1,24 @@ +import { NodeType, type PaletteItem } from '@workflowbuilder/sdk'; + +import { defaultPropertiesData } from './default-properties-data'; +import { type AgentHarnessSchema, schema } from './schema'; +import { uischema } from './uischema'; + +export const agentHarnessPaletteItem: PaletteItem = { + label: 'Agent Harness', + description: 'Delegate a step to an autonomous coding-agent CLI (GitHub Copilot)', + type: 'ai-studio/agent-harness', + icon: 'Terminal', + templateType: NodeType.Node, + defaultPropertiesData, + schema, + uischema, + // Lets `{{ nodes..response }}` references resolve to a real mention instead of a "missing mention" pill. + outputSchema: { + type: 'default', + properties: { + response: { type: 'string', label: 'Response', description: 'The text produced by the agent run' }, + tokens: { type: 'object', label: 'Tokens', description: 'Token usage reported by the provider' }, + }, + }, +}; diff --git a/apps/ai-studio/src/nodes/agent-harness/schema.ts b/apps/ai-studio/src/nodes/agent-harness/schema.ts new file mode 100644 index 000000000..14516d4d6 --- /dev/null +++ b/apps/ai-studio/src/nodes/agent-harness/schema.ts @@ -0,0 +1,94 @@ +import { sharedProperties } from '@workflowbuilder/sdk'; +import type { NodeSchema } from '@workflowbuilder/sdk'; + +// Only Copilot is functionally wired in v1 (see `apps/execution-worker/src/agent-harness/registry.ts`). +// Kept as a single-entry option list, rather than a plain string, so the Select still renders — a +// second provider is a one-line addition here once its backend lands (per issue #147 parity intent). +const providerOptions = [{ label: 'GitHub Copilot', value: 'copilot' }]; + +// Copilot's own effort ladder (`COPILOT_EFFORTS` in providers/copilot/config.ts) is narrower than the +// full cross-provider `EffortRung` union — only offer the rungs this provider actually honors. +const effortOptions = [ + { label: 'Low', value: 'low' }, + { label: 'Medium', value: 'medium' }, + { label: 'High', value: 'high' }, + { label: 'Extra high', value: 'xhigh' }, +]; + +// 'shared'/'resume' are NOT YET SUPPORTED (session persistence is out of scope for v1, see the +// implementation plan §2) — rendered anyway per issue #147 ("fields must still render"). +const contextOptions = [ + { label: 'Fresh (new session)', value: 'fresh' }, + { label: 'Shared (not yet supported)', value: 'shared' }, + { label: 'Resume (not yet supported)', value: 'resume' }, +]; + +// UI-only convenience preset; expanding it into allowedTools/deniedTools is a manual step for now +// (no automatic preset -> field wiring in this milestone, see nodes/agent-harness/index.ts handoff). +const toolsModeOptions = [ + { label: 'None', value: 'none' }, + { label: 'Read-only', value: 'read-only' }, + { label: 'Edit-only', value: 'edit-only' }, + { label: 'All tools', value: 'all' }, +]; + +export const schema = { + type: 'object', + properties: { + ...sharedProperties, + prompt: { + type: 'string', + }, + provider: { + type: 'string', + options: providerOptions, + }, + model: { + type: 'string', + }, + effort: { + type: 'string', + options: effortOptions, + }, + context: { + type: 'string', + options: contextOptions, + }, + idle_timeout: { + type: 'number', + }, + toolsMode: { + type: 'string', + options: toolsModeOptions, + }, + allowedTools: { + type: 'string', + }, + deniedTools: { + type: 'string', + }, + output_format: { + type: 'string', + }, + mcp: { + type: 'string', + }, + skills: { + type: 'string', + }, + agents: { + type: 'string', + }, + maxBudgetUsd: { + type: 'number', + }, + mutatesCheckout: { + type: 'boolean', + }, + persistSession: { + type: 'boolean', + }, + }, +} satisfies NodeSchema; + +export type AgentHarnessSchema = typeof schema; diff --git a/apps/ai-studio/src/nodes/agent-harness/uischema.ts b/apps/ai-studio/src/nodes/agent-harness/uischema.ts new file mode 100644 index 000000000..15358762f --- /dev/null +++ b/apps/ai-studio/src/nodes/agent-harness/uischema.ts @@ -0,0 +1,143 @@ +import { getScope } from '@workflowbuilder/sdk'; +import type { UISchema } from '@workflowbuilder/sdk'; + +import type { AgentHarnessSchema } from './schema'; + +const scope = getScope; + +// The SDK's `UISchema` control elements have no `description`/help-text slot (see +// `packages/sdk/src/types/controls.ts` — only `label`/`placeholder`). The plan's suggested +// per-field "description" mechanism does not exist, so out-scoped fields communicate their +// "not yet supported" status through the `label` and `placeholder` text instead. This is a +// documented deviation from the plan's assumption, not an omission — see the M7 handoff. +export const uischema: UISchema = { + type: 'VerticalLayout', + elements: [ + { + type: 'Accordion', + label: 'General', + elements: [ + { + type: 'Text', + scope: scope('properties.label'), + label: 'Title', + placeholder: 'Node Title...', + }, + { + type: 'TextArea', + scope: scope('properties.prompt'), + label: 'Prompt', + placeholder: 'Describe the task for the agent... supports {{ nodes..output }} references', + minRows: 5, + maxRows: 14, + }, + { + type: 'Select', + scope: scope('properties.provider'), + label: 'Provider', + }, + ], + }, + { + type: 'Accordion', + label: 'Execution', + elements: [ + { + type: 'Text', + scope: scope('properties.model'), + label: 'Model', + placeholder: 'auto', + }, + { + type: 'Select', + scope: scope('properties.effort'), + label: 'Effort', + }, + { + type: 'Select', + scope: scope('properties.context'), + label: 'Context (only "Fresh" is supported today)', + }, + { + type: 'Text', + scope: scope('properties.idle_timeout'), + label: 'Idle timeout (ms)', + placeholder: 'Defaults to 30 minutes', + }, + ], + }, + { + type: 'Accordion', + label: 'Tools', + elements: [ + { + type: 'Select', + scope: scope('properties.toolsMode'), + label: 'Tools preset (UI convenience — reflect manually into the lists below)', + }, + { + type: 'Text', + scope: scope('properties.allowedTools'), + label: 'Allowed tools (comma-separated)', + placeholder: 'e.g. read_file, write_file', + }, + { + type: 'Text', + scope: scope('properties.deniedTools'), + label: 'Denied tools (comma-separated)', + placeholder: 'e.g. shell_exec', + }, + ], + }, + { + type: 'Accordion', + label: 'Advanced', + elements: [ + { + type: 'TextArea', + scope: scope('properties.output_format'), + label: 'Output format (JSON — not yet supported, backend does not enforce it)', + placeholder: '{ "type": "json" }', + minRows: 3, + maxRows: 8, + }, + { + type: 'Text', + scope: scope('properties.mcp'), + label: 'MCP config path (not yet supported)', + placeholder: '/path/to/mcp.json', + }, + { + type: 'Text', + scope: scope('properties.skills'), + label: 'Skills (comma-separated, not yet supported)', + placeholder: 'e.g. code-review, testing', + }, + { + type: 'TextArea', + scope: scope('properties.agents'), + label: 'Sub-agents (JSON)', + placeholder: '{ "reviewer": { "description": "...", "prompt": "..." } }', + minRows: 3, + maxRows: 8, + }, + { + type: 'Text', + scope: scope('properties.maxBudgetUsd'), + label: 'Max budget (USD, not yet supported)', + placeholder: 'e.g. 5', + }, + { + type: 'Switch', + scope: scope('properties.mutatesCheckout'), + label: 'Mutates checkout (node may write to the working tree)', + }, + { + type: 'Switch', + scope: scope('properties.persistSession'), + label: 'Persist session (not yet supported)', + }, + ], + }, + ], +}; diff --git a/apps/ai-studio/src/nodes/ai-agent/default-properties-data.ts b/apps/ai-studio/src/nodes/ai-agent/default-properties-data.ts index b4483d606..f105f0823 100644 --- a/apps/ai-studio/src/nodes/ai-agent/default-properties-data.ts +++ b/apps/ai-studio/src/nodes/ai-agent/default-properties-data.ts @@ -7,4 +7,5 @@ export const defaultPropertiesData: NodeDataProperties = { description: '', systemPrompt: '', webSearch: false, + provider: 'auto', }; diff --git a/apps/ai-studio/src/nodes/ai-agent/schema.ts b/apps/ai-studio/src/nodes/ai-agent/schema.ts index 969af44b5..2cc5a56e8 100644 --- a/apps/ai-studio/src/nodes/ai-agent/schema.ts +++ b/apps/ai-studio/src/nodes/ai-agent/schema.ts @@ -1,6 +1,15 @@ import { sharedProperties } from '@workflowbuilder/sdk'; import type { NodeSchema } from '@workflowbuilder/sdk'; +// Kept as a plain string (not an enum) so a provider added later, or a custom +// OpenAI-compatible endpoint, doesn't require a schema migration to unlock. +const providerOptions = [ + { label: 'Auto', value: 'auto' }, + { label: 'OpenRouter', value: 'openrouter' }, + { label: 'OpenAI', value: 'openai' }, + { label: 'Anthropic', value: 'anthropic' }, +]; + export const schema = { type: 'object', properties: { @@ -11,6 +20,13 @@ export const schema = { webSearch: { type: 'boolean', }, + model: { + type: 'string', + }, + provider: { + type: 'string', + options: providerOptions, + }, }, } satisfies NodeSchema; diff --git a/apps/ai-studio/src/nodes/ai-agent/uischema.ts b/apps/ai-studio/src/nodes/ai-agent/uischema.ts index 06abc4aa4..134c0b01b 100644 --- a/apps/ai-studio/src/nodes/ai-agent/uischema.ts +++ b/apps/ai-studio/src/nodes/ai-agent/uischema.ts @@ -27,5 +27,16 @@ export const uischema: UISchema = { scope: scope('properties.webSearch'), label: 'Web search (let the agent look things up)', }, + { + type: 'Text', + scope: scope('properties.model'), + label: 'Model', + placeholder: 'Inherit workflow/deployment default', + }, + { + type: 'Select', + scope: scope('properties.provider'), + label: 'Provider', + }, ], }; diff --git a/apps/execution-worker/.env.example b/apps/execution-worker/.env.example index 94e2ecd4c..7c17aee7f 100644 --- a/apps/execution-worker/.env.example +++ b/apps/execution-worker/.env.example @@ -9,3 +9,7 @@ AI_MODEL=mistralai/mistral-small-3.2-24b-instruct # free key at https://tavily.com (free tier ~1000 searches/month). Leave empty # to disable: agents with web search toggled on still run, just without the tool. TAVILY_API_KEY= + +# Queue polled only by the specialized worker (start:specialized). A node type is +# routed here via its activity profile's `taskQueue`; unset means the default queue. +SPECIALIZED_TASK_QUEUE=workflow-execution-specialized diff --git a/apps/execution-worker/README.md b/apps/execution-worker/README.md index a9aa5d107..ebacaea49 100644 --- a/apps/execution-worker/README.md +++ b/apps/execution-worker/README.md @@ -35,6 +35,63 @@ See `.env.example`. Required: | `TEMPORAL_ADDRESS` | Temporal server address | `127.0.0.1:7233` | | `AI_MODEL` | OpenRouter model ID | `anthropic/claude-3.5-haiku` | +Optional, per-provider direct-routing keys — none is declared here or in `env.ts`; each is read +directly by its own `@ai-sdk/` package only when an `ai-agent` node selects that `provider` +explicitly (or `'auto'` infers it from the model-id prefix). Presence is checked with +`Boolean(process.env.)`; the key value itself is never read, stored, or logged by this app. + +| Provider | Env var | +| ------------ | ------------------------------ | +| `openai` | `OPENAI_API_KEY` | +| `anthropic` | `ANTHROPIC_API_KEY` | +| `google` | `GOOGLE_GENERATIVE_AI_API_KEY` | +| `xai` | `XAI_API_KEY` | +| `mistral` | `MISTRAL_API_KEY` | +| `cohere` | `COHERE_API_KEY` | +| `deepseek` | `DEEPSEEK_API_KEY` | +| `moonshotai` | `MOONSHOT_API_KEY` | +| `groq` | `GROQ_API_KEY` | +| `togetherai` | `TOGETHER_API_KEY` | +| `fireworks` | `FIREWORKS_API_KEY` | +| `perplexity` | `PERPLEXITY_API_KEY` | +| `cerebras` | `CEREBRAS_API_KEY` | +| `deepinfra` | `DEEPINFRA_API_KEY` | + +An `ai-agent` node's `model`/`provider` config fields fall back to `env.AI_MODEL`/`'auto'` when +unset (`apps/execution-worker/src/model-provider.ts`). `provider` accepts free text for a value +not yet in this table — that currently fails at execution with a clear error until support (a +table row plus its `@ai-sdk/` dependency) is added. + +## `ai-studio/agent-harness` node + +Delegates a workflow step to an external autonomous coding-agent CLI (v1: GitHub Copilot +only), as opposed to `ai-studio/ai-agent`'s single bounded LLM call. It is longer-running +(minutes, not seconds), side-effecting (may write files in a working directory), and +shells out to an external CLI/SDK rather than calling a model API directly. See +[`src/agent-harness/README.md`](./src/agent-harness/README.md) for the provider +architecture (ported from [coleam00/Archon](https://github.com/coleam00/Archon), MIT). + +**Env vars** (both optional): + +| Var | Purpose | Default | +| ---------------------- | ----------------------------------------------------- | ---------------------------------------------------------- | +| `COPILOT_GITHUB_TOKEN` | GitHub token used to authenticate the `copilot` CLI | Falls back to the ambient `copilot login` session if unset | +| `COPILOT_CLI_PATH` | Overrides binary resolution (skips the `PATH` lookup) | Resolved via `PATH` | + +**Activity profile** (`'ai-studio/agent-harness'` in `engines/temporal/worker.ts`): + +| Field | Value | Why | +| ----------------------- | ------- | ------------------------------------------------------------------------------------------------------------------------------ | +| `startToCloseTimeout` | `'45m'` | The CLI can run a genuinely long agentic task. | +| `retry.maximumAttempts` | `1` | Zero automatic retries — this is a side-effecting node; retrying it could re-run mutations. | +| `heartbeatTimeout` | `'5s'` | Without heartbeating, Temporal cancellation is not detected promptly (see `packages/temporal/README.md` § "heartbeatTimeout"). | + +**Operational constraints:** + +- No retries. A failed run is a failed run; re-triggering the workflow is the user's decision, not the platform's. +- Cancellation is handled by the SDK, not a manual process-group kill: the activity aborts the run via the Copilot SDK's own `session.abort()`/`client.stop()`, which cleanly terminates the underlying subprocess tree. This was verified empirically (zero orphaned `copilot` processes across repeated cancellation tests) — no `spawn(detached)+process.kill(-pid)` workaround was needed, unlike Archon's own implementation which targets a different (Bun-compiled) binary shape. +- **`idle_timeout` is in milliseconds, not seconds.** A value of `300` means 300ms, not 5 minutes — for 5 minutes, set `idle_timeout: 300000`. This has bitten someone during E2E testing already (a `300` intended as "5 minutes" produced an almost-instant timeout); the UI label ("Idle timeout (ms)") is correct, but easy to misread under time pressure. + ## Structure ``` @@ -44,6 +101,7 @@ src/ └── engines/ └── temporal/ ├── worker.ts # Worker bootstrap: executors + store, handed to WorkflowBuilderPlugin + ├── specialized-worker.ts # Activity-only worker for a taskQueue-routed subset of node types └── workflows.ts # One-line re-export of runWorkflow for Temporal's bundler ``` @@ -51,6 +109,20 @@ The workflow itself, the activity contract and the event emitter live in [`@workflowbuilder/temporal`](../../packages/temporal/README.md). This app only supplies what is its own: one executor per node type and the database as the store port. +## Per-node-type task queue routing + +A node type's activity profile can carry `taskQueue`, which pins it to a queue other +than the default (`plugin.taskQueue`). `worker.ts` keeps polling the default queue for +everything else; `specialized-worker.ts` is a second, activity-only entrypoint (no +`workflowsPath` — Temporal supports activity-only workers) that polls +`SPECIALIZED_TASK_QUEUE` and registers only the node type(s) routed there. Run it with +`pnpm --filter execution-worker start:specialized`, or as the `worker-specialized` +compose service (see `deploy/ai-studio/README.md`). + +This is a deployment-only change: `runGraph` and the graph model never see a taskQueue, +they only affect which worker process a node's `executeNode` activity is scheduled on. +A profile with no `taskQueue` behaves exactly as before. + ## Temporal specifics - **Task queue:** `workflow-execution`, read from `plugin.taskQueue` so the backend and the worker cannot drift apart. Both default to the same constant in the package. diff --git a/apps/execution-worker/package.json b/apps/execution-worker/package.json index b0ee15a9b..b53881ac2 100644 --- a/apps/execution-worker/package.json +++ b/apps/execution-worker/package.json @@ -7,6 +7,7 @@ "dev": "tsx watch --env-file=.env ./src/engines/temporal/worker.ts", "start": "tsx --env-file=.env ./src/engines/temporal/worker.ts", "start:prod": "tsx ./src/engines/temporal/worker.ts", + "start:specialized": "tsx ./src/engines/temporal/specialized-worker.ts", "typecheck": "tsc --noEmit", "lint": "eslint", "lint:fix": "eslint --fix", @@ -14,7 +15,23 @@ "test:watch": "vitest" }, "dependencies": { + "@ai-sdk/anthropic": "^3.0.118", + "@ai-sdk/cerebras": "^2.0.81", + "@ai-sdk/cohere": "^3.0.61", + "@ai-sdk/deepinfra": "^2.0.79", + "@ai-sdk/deepseek": "^2.0.64", + "@ai-sdk/fireworks": "^2.0.85", + "@ai-sdk/google": "^3.0.122", + "@ai-sdk/groq": "^3.0.66", + "@ai-sdk/mistral": "^3.0.64", + "@ai-sdk/moonshotai": "^2.0.56", + "@ai-sdk/openai": "^3.0.112", + "@ai-sdk/perplexity": "^3.0.60", + "@ai-sdk/togetherai": "^2.0.81", + "@ai-sdk/xai": "^3.0.132", + "@github/copilot-sdk": "^1.0.13", "@openrouter/ai-sdk-provider": "^2.5.0", + "@temporalio/activity": "catalog:", "@temporalio/worker": "catalog:", "@temporalio/workflow": "catalog:", "@workflow-builder/execution-core": "workspace:*", diff --git a/apps/execution-worker/src/activities/agent-harness.test.ts b/apps/execution-worker/src/activities/agent-harness.test.ts new file mode 100644 index 000000000..97752f7d3 --- /dev/null +++ b/apps/execution-worker/src/activities/agent-harness.test.ts @@ -0,0 +1,328 @@ +import { execFile } from 'node:child_process'; +import { mkdtemp, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { promisify } from 'node:util'; +import { afterEach, describe, expect, test } from 'vitest'; + +import type { ExecutionContext } from '@workflow-builder/execution-core'; + +import type { IAgentProvider, MessageChunk, ProviderCapabilities } from '../agent-harness/types'; +import { type AgentHarnessNode, executeAgentHarness } from './agent-harness'; + +const execFileAsync = promisify(execFile); + +const FAKE_CAPABILITIES: ProviderCapabilities = { + sessionResume: false, + mcp: false, + hooks: false, + skills: false, + agents: false, + toolRestrictions: false, + structuredOutput: false, + envInjection: false, + costControl: false, + effortControl: false, + fallbackModel: false, + sandbox: false, + settingSources: false, + nativeTools: false, + containerExec: false, +}; + +/** Fake IAgentProvider driven by a hand-written async generator — no real CLI/SDK. */ +class FakeProvider implements IAgentProvider { + /** Captures the exact prompt string `executeAgentHarness` sent, for assertion in tests. */ + public receivedPrompt: string | undefined; + + constructor(private readonly chunks: MessageChunk[] | (() => AsyncGenerator)) {} + + getType(): string { + return 'fake'; + } + + getCapabilities(): ProviderCapabilities { + return FAKE_CAPABILITIES; + } + + async *sendQuery(prompt: string): AsyncGenerator { + this.receivedPrompt = prompt; + if (typeof this.chunks === 'function') { + yield* this.chunks(); + return; + } + for (const chunk of this.chunks) { + yield chunk; + } + } +} + +class ThrowingProvider implements IAgentProvider { + constructor( + private readonly error: Error, + private readonly beforeThrow?: MessageChunk[], + ) {} + + getType(): string { + return 'fake-throwing'; + } + + getCapabilities(): ProviderCapabilities { + return FAKE_CAPABILITIES; + } + + async *sendQuery(): AsyncGenerator { + for (const chunk of this.beforeThrow ?? []) { + yield chunk; + } + throw this.error; + } +} + +function makeContext(overrides: Partial = {}): ExecutionContext { + return { + workflowId: 'wf-1', + executionId: 'exec-1', + triggerPayload: {}, + nodeOutputs: {}, + variables: {}, + global: {}, + ...overrides, + }; +} + +function makeNode(overrides: Partial = {}): AgentHarnessNode { + return { + id: 'agent-harness-1', + type: 'ai-studio/agent-harness', + config: { + prompt: 'hello', + provider: 'fake', + ...overrides, + }, + }; +} + +const scratchDirectoriesToClean: string[] = []; +afterEach(async () => { + await Promise.all( + scratchDirectoriesToClean.splice(0).map((directory) => rm(directory, { recursive: true, force: true })), + ); +}); + +describe('executeAgentHarness', () => { + test('accumulates text, tokens, and warnings into the final output shape', async () => { + const provider = new FakeProvider([ + { type: 'system', content: 'a warning' }, + { type: 'assistant', content: 'Hello ' }, + { type: 'assistant', content: 'world' }, + { type: 'result', sessionId: 's1', tokens: { input: 10, output: 5 } }, + ]); + + const result = await executeAgentHarness(makeNode(), makeContext(), { + getProvider: () => provider, + }); + + expect(result.output.response).toBe('Hello world'); + expect(result.output.tokens).toEqual({ input: 10, output: 5 }); + expect(result.output.warnings).toEqual(['a warning']); + }); + + test("discards a prior turn's text on assistant_turn_boundary, keeping only the final turn's response", async () => { + const provider = new FakeProvider([ + { type: 'assistant', content: "I'll inspect the working directory, then create plan.md." }, + { type: 'tool', toolName: 'writeFile' }, + { type: 'assistant_turn_boundary' }, + { type: 'assistant', content: 'Created plan.md with three rate-limiting approaches.' }, + { type: 'result', sessionId: 's1' }, + ]); + + const result = await executeAgentHarness(makeNode(), makeContext(), { + getProvider: () => provider, + }); + + expect(result.output.response).toBe('Created plan.md with three rate-limiting approaches.'); + }); + + test('resolves {{namespace.path}} references in the prompt before sending it to the provider', async () => { + const provider = new FakeProvider([{ type: 'assistant', content: 'ok' }]); + + await executeAgentHarness( + makeNode({ prompt: 'Classify: {{trigger.inputPrompt}}' }), + makeContext({ triggerPayload: { inputPrompt: 'Charged twice, need a refund.' } }), + { getProvider: () => provider }, + ); + + expect(provider.receivedPrompt).toContain('Charged twice, need a refund.'); + expect(provider.receivedPrompt).not.toContain('{{trigger.inputPrompt}}'); + }); + + test('prepends upstream node outputs to the prompt, matching ai-agent.ts parity, even without an explicit reference', async () => { + const provider = new FakeProvider([{ type: 'assistant', content: 'ok' }]); + + await executeAgentHarness( + makeNode({ prompt: 'Classify the ticket above.' }), + makeContext({ nodeOutputs: { 'trigger-1': { response: 'Charged twice, need a refund.' } } }), + { getProvider: () => provider }, + ); + + expect(provider.receivedPrompt).toContain('Charged twice, need a refund.'); + expect(provider.receivedPrompt).toContain('Classify the ticket above.'); + }); + + test('classifies a FATAL provider error as a permanent (non-retryable) error', async () => { + const provider = new ThrowingProvider(new Error('401 unauthorized'), [{ type: 'assistant', content: 'partial' }]); + + await expect(executeAgentHarness(makeNode(), makeContext(), { getProvider: () => provider })).rejects.toMatchObject( + { name: 'PermanentNodeExecutionError' }, + ); + }); + + test('classifies a TRANSIENT provider error as a transient (retryable) error', async () => { + const provider = new ThrowingProvider(new Error('ECONNRESET while streaming')); + + await expect(executeAgentHarness(makeNode(), makeContext(), { getProvider: () => provider })).rejects.toMatchObject( + { name: 'TransientNodeExecutionError' }, + ); + }); + + test('cleans up the scratch dir even when the provider throws mid-stream', async () => { + let capturedCwd: string | undefined; + class CapturingThrowingProvider implements IAgentProvider { + getType(): string { + return 'fake-capturing'; + } + getCapabilities(): ProviderCapabilities { + return FAKE_CAPABILITIES; + } + async *sendQuery(_prompt: string, cwd: string): AsyncGenerator { + capturedCwd = cwd; + yield { type: 'assistant', content: 'x' }; + throw new Error('boom'); + } + } + + await expect( + executeAgentHarness(makeNode(), makeContext(), { + getProvider: () => new CapturingThrowingProvider(), + }), + ).rejects.toThrow(); + + expect(capturedCwd).toBeDefined(); + await expect(execFileAsync('test', ['-d', capturedCwd!])).rejects.toBeDefined(); + }); + + test('does not tear down accumulation early when a result arrives while background_tasks is non-empty', async () => { + const provider = new FakeProvider(async function* () { + yield { type: 'assistant', content: 'first turn. ' } as MessageChunk; + yield { type: 'background_tasks', tasks: [{ taskId: 't1', taskType: 'sub', description: 'd' }] }; + // A `result` chunk while a background task is still live must NOT end the stream. + yield { type: 'result', sessionId: 's-early' }; + yield { type: 'assistant', content: 'second turn.' }; + yield { type: 'background_tasks', tasks: [] }; + yield { type: 'result', sessionId: 's-final', tokens: { input: 1, output: 1 } }; + }); + + const result = await executeAgentHarness(makeNode(), makeContext(), { + getProvider: () => provider, + }); + + expect(result.output.response).toBe('first turn. second turn.'); + }); + + test('assertCheckoutUntouched fails a mutatesCheckout: false node when the working tree changed', async () => { + const repoDirectory = await mkdtemp(path.join(tmpdir(), 'agent-harness-git-')); + scratchDirectoriesToClean.push(repoDirectory); + await execFileAsync('git', ['init', '-q'], { cwd: repoDirectory }); + await execFileAsync('git', ['config', 'user.email', 'test@test.dev'], { cwd: repoDirectory }); + await execFileAsync('git', ['config', 'user.name', 'test'], { cwd: repoDirectory }); + await writeFile(path.join(repoDirectory, 'committed.txt'), 'v1'); + await execFileAsync('git', ['add', '-A'], { cwd: repoDirectory }); + await execFileAsync('git', ['commit', '-q', '-m', 'init'], { cwd: repoDirectory }); + + class MutatingProvider implements IAgentProvider { + getType(): string { + return 'fake-mutating'; + } + getCapabilities(): ProviderCapabilities { + return FAKE_CAPABILITIES; + } + async *sendQuery(_prompt: string, cwd: string): AsyncGenerator { + await writeFile(path.join(cwd, 'mutated.txt'), 'unexpected'); + yield { type: 'assistant', content: 'done' }; + yield { type: 'result', sessionId: 's1' }; + } + } + + await expect( + executeAgentHarness( + makeNode({ mutatesCheckout: false }), + makeContext({ variables: { workdir: repoDirectory } }), + { getProvider: () => new MutatingProvider() }, + ), + ).rejects.toMatchObject({ name: 'PermanentNodeExecutionError' }); + }); + + test('empty output fails the run', async () => { + const provider = new FakeProvider([{ type: 'result', sessionId: 's1' }]); + + await expect(executeAgentHarness(makeNode(), makeContext(), { getProvider: () => provider })).rejects.toThrow(); + }); + + test('empty prompt fails fast as a permanent error instead of waiting on the idle timeout', async () => { + const provider = new FakeProvider([{ type: 'assistant', content: 'should never run' }]); + + const start = Date.now(); + await expect( + executeAgentHarness(makeNode({ prompt: ' ' }), makeContext(), { getProvider: () => provider }), + ).rejects.toMatchObject({ classification: 'permanent' }); + expect(Date.now() - start).toBeLessThan(1000); + }); + + test('idle timeout with partial output salvages a successful result with a warning (Archon L3087-3135)', async () => { + class StallingProvider implements IAgentProvider { + getType(): string { + return 'fake-stalling'; + } + getCapabilities(): ProviderCapabilities { + return FAKE_CAPABILITIES; + } + async *sendQuery(): AsyncGenerator { + yield { type: 'assistant', content: 'partial' }; + // Never yields again — the idle timeout must fire and end the stream. + await new Promise(() => {}); + } + } + + const start = Date.now(); + const result = await executeAgentHarness(makeNode({ idle_timeout: 200 }), makeContext(), { + getProvider: () => new StallingProvider(), + }); + expect(Date.now() - start).toBeLessThan(2000); + expect(result.output.response).toBe('partial'); + expect(result.output.warnings?.some((w) => w.includes('idle timeout'))).toBe(true); + }); + + test('idle timeout with zero output fails the run', async () => { + class SilentStallingProvider implements IAgentProvider { + getType(): string { + return 'fake-silent-stalling'; + } + getCapabilities(): ProviderCapabilities { + return FAKE_CAPABILITIES; + } + async *sendQuery(): AsyncGenerator { + await new Promise(() => {}); + yield { type: 'assistant', content: 'unreachable' }; + } + } + + const start = Date.now(); + await expect( + executeAgentHarness(makeNode({ idle_timeout: 200 }), makeContext(), { + getProvider: () => new SilentStallingProvider(), + }), + ).rejects.toThrow(); + expect(Date.now() - start).toBeLessThan(2000); + }); +}); diff --git a/apps/execution-worker/src/activities/agent-harness.ts b/apps/execution-worker/src/activities/agent-harness.ts new file mode 100644 index 000000000..9d01f54e9 --- /dev/null +++ b/apps/execution-worker/src/activities/agent-harness.ts @@ -0,0 +1,444 @@ +/** + * Temporal activity adaptation for the `ai-studio/agent-harness` node type. + * + * Ported from Archon's `packages/workflows/src/dag-executor.ts` per + * PORTING-MAP.md §5 — the file is 12,098 LOC and is NOT ported wholesale; + * only the slices listed there are extracted and reshaped into a single + * activity-shaped function. See that table for exact line ranges and + * fidelity (verbatim / trimmed / adapted / skipped) per slice. + * + * Adaptations (A2/A3, per the implementation plan §6 M5): + * - Archon's DB-poll cancel check + activity heartbeat DB write are dropped + * entirely in favor of Temporal's own cancellation + heartbeat machinery + * (`Context.current().cancellationSignal` / `.heartbeat()`), which this + * activity already runs inside of (the plugin's `executeNode` activity — + * see `packages/temporal/src/activities.ts` — calls this executor). + * - Process-group kill (A3): the Copilot SDK owns its own subprocess and + * does not expose its pid (verified in M4 against `@github/copilot-sdk`'s + * type definitions), so `spawn(..., { detached: true })` + + * `process.kill(-pid, 'SIGTERM')` has no attachment point. Cancellation + * instead goes through the SDK's own `session.abort()` (already wired end + * to end: `provider.sendQuery`'s `abortSignal` option → `bridgeSession` → + * `session.abort()` → `client.stop()`). Verified empirically in this + * milestone's manual E2E — see the M5 handoff for the measured result. + * - Retry loop (Archon L1054-1205) is skipped: v1 runs a node with a single + * attempt (`maximumAttempts: 1`, set in M6's activity profile). + * - Structured-output reask loop (Archon L2968-3085) is skipped — out of + * scope per the porting plan §2. + * - UI streaming fan-out (Archon's `safeSendMessage`/`sendStructuredEvent` + * calls throughout the L2353-2700 switch) is dropped — out of scope for + * this activity; only text/token/warning accumulation is kept. + */ +import { Context } from '@temporalio/activity'; +import { execFile } from 'node:child_process'; +import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { promisify } from 'node:util'; + +import { + type ExecutionContext, + type LoggerPort, + NodeExecutionError, + resolveTemplate, +} from '@workflow-builder/execution-core'; + +import type { ResolvedCredential } from '../agent-harness/credentials/delivery'; +import { deliverCredential } from '../agent-harness/credentials/delivery'; +import { getAgentProvider } from '../agent-harness/registry'; +import { classifyError, toHostNodeExecutionError } from '../agent-harness/shared/error-classification'; +import { STEP_IDLE_TIMEOUT_MS, withIdleTimeout } from '../agent-harness/shared/idle-timeout'; +import type { IAgentProvider, MessageChunk, TokenUsage } from '../agent-harness/types'; +import { mergeTokenUsage } from '../agent-harness/types'; +import type { AgentHarnessNode } from '../domain/ai-studio-nodes'; + +const execFileAsync = promisify(execFile); + +export type { AgentHarnessNode } from '../domain/ai-studio-nodes'; + +export interface AgentHarnessResult { + output: { + response: string; + tokens?: TokenUsage; + warnings?: string[]; + }; +} + +export interface AgentHarnessDeps { + logger?: LoggerPort; + /** + * Already-resolved credential for `node.config.credentialVendor`, or + * `undefined` to rely on ambient/env auth. Credential storage/retrieval + * (looking a vendor id up in a vault) is not yet wired end-to-end — that + * is a separate concern from M3's `deliverCredential` (env/file shaping), + * which this activity does call. Callers (M6+) inject the resolved + * credential here once that lookup exists. + */ + credential?: ResolvedCredential; + /** Provider factory override, defaults to the M4 registry's `getAgentProvider`. */ + getProvider?: (providerId: string) => IAgentProvider; + /** + * Override for the Temporal activity context (heartbeat/cancellation). + * Tests and the manual-invocation smoke test (see the M5 handoff) supply + * this directly since `Context.current()` throws outside a real activity. + * Production callers should omit it — `Context.current()` is used. + */ + activityContext?: { heartbeat: () => void; cancellationSignal: AbortSignal }; +} + +// ─── Checkout snapshot (Archon dag-executor.ts L1234-1331 — verbatim) ────── + +function checkoutSnapshotExcludes(...directories: readonly string[]): readonly string[] { + return directories.map((d) => path.resolve(d)); +} + +function isInsideAny(absPath: string, directories: readonly string[]): boolean { + return directories.some((d) => absPath === d || absPath.startsWith(d + path.sep)); +} + +/** Path operands of one `git status --porcelain` line (`XY path` or `XY old -> new`). */ +function unquotePathSegment(value: string): string { + return value.startsWith('"') && value.endsWith('"') ? value.slice(1, -1) : value; +} + +function porcelainPaths(line: string): string[] { + const body = line.slice(3); + return (body.includes(' -> ') ? body.split(' -> ') : [body]).map(unquotePathSegment); +} + +/** + * Snapshot the working tree's dirty state (`git status --porcelain`) for a + * node's `mutatesCheckout: false` assertion, dropping entries under + * `excludeDirs`. Returns `undefined` when the check cannot run — cwd outside + * a repo, or git failing — so a broken assertion degrades to no check rather + * than breaking unrelated runs. + */ +export async function snapshotCheckout( + cwd: string, + excludeDirectories: readonly string[], +): Promise { + try { + const { stdout } = await execFileAsync( + 'git', + ['-c', 'core.quotePath=false', 'status', '--porcelain', '--untracked-files=normal'], + { cwd, timeout: 10_000 }, + ); + const relevant = stdout + .split('\n') + .filter((line) => line.length > 3) + .filter((line) => !porcelainPaths(line).some((p) => isInsideAny(path.resolve(cwd, p), excludeDirectories))); + return relevant.join('\n'); + } catch { + return undefined; + } +} + +/** + * Enforce a node's `mutatesCheckout: false` declaration: when the node ran + * successfully but the pre-run snapshot changed, throw a non-retryable + * (permanent) error naming the node and listing what moved. Called OUTSIDE + * any retry path (there is none in v1 — this is naturally satisfied — but + * kept as an explicit invariant for when retry lands: retrying a node that + * provably mutates the checkout would only multiply the damage). + */ +export async function assertCheckoutUntouched( + nodeId: string, + cwd: string, + excludeDirectories: readonly string[], + before: string | undefined, +): Promise { + if (before === undefined) return; + const after = await snapshotCheckout(cwd, excludeDirectories); + if (after === undefined || after === before) return; + const changedPaths = after.split('\n').filter(Boolean).flatMap(porcelainPaths).slice(0, 10).join(', '); + throw toHostNodeExecutionError( + 'FATAL', + 'agent_harness.mutates_checkout_violation', + `Node '${nodeId}' declared 'mutatesCheckout: false' but modified the working tree: ${changedPaths}`, + ); +} + +// ─── Background-task drain tracker (Archon L1019-1047 — verbatim) ───────── + +/** + * Tracks the provider's live background-Agent-task set (Archon #2083). A + * `result` chunk arriving while the set is non-empty must NOT tear down the + * stream — the provider holds its subprocess open to let the tasks finish + * and runs a follow-up turn to integrate their output. + */ +function createBackgroundTaskTracker(): { + update(tasks: { taskId: string }[]): void; + shouldBreakOnResult(): boolean; + ids(): string[]; +} { + const live = new Set(); + return { + update(tasks): void { + live.clear(); + for (const t of tasks) live.add(t.taskId); + }, + shouldBreakOnResult(): boolean { + return live.size === 0; + }, + ids(): string[] { + return [...live]; + }, + }; +} + +// ─── Workdir resolution (A6) ──────────────────────────────────────────────── + +/** + * A6: prefer a workdir the caller's workflow context already resolved (no + * existing `ExecutionContext` field carries one today — `ai-agent`, the only + * other AI Studio executor, never touches the filesystem — so this checks an + * optional `variables.workdir` convention M6 may adopt), else fall back to a + * fresh scratch directory. The caller owns cleanup of an externally-supplied + * workdir; a scratch dir created here is removed in this activity's own + * `finally`. + */ +async function resolveWorkdir(context: ExecutionContext): Promise<{ cwd: string; isScratch: boolean }> { + const fromContext = context.variables?.['workdir']; + if (typeof fromContext === 'string' && fromContext.length > 0) { + return { cwd: fromContext, isScratch: false }; + } + const cwd = await mkdtemp(path.join(tmpdir(), 'agent-harness-')); + return { cwd, isScratch: true }; +} + +// ─── Activity context resolution ─────────────────────────────────────────── + +function resolveActivityContext(override: AgentHarnessDeps['activityContext']): { + heartbeat: () => void; + cancellationSignal: AbortSignal; +} { + if (override) return override; + try { + const context = Context.current(); + return { heartbeat: () => context.heartbeat(), cancellationSignal: context.cancellationSignal }; + } catch { + // No real Temporal activity context (unit test / manual smoke-test + // invocation) — heartbeat is a no-op and cancellation never fires unless + // the caller wires its own AbortController via `activityContext`. + return { heartbeat: () => {}, cancellationSignal: new AbortController().signal }; + } +} + +// ─── Main activity ────────────────────────────────────────────────────────── + +const HEARTBEAT_INTERVAL_MS = 1000; + +export async function executeAgentHarness( + node: AgentHarnessNode, + context: ExecutionContext, + deps: AgentHarnessDeps = {}, +): Promise { + const log = deps.logger; + const activityContext = resolveActivityContext(deps.activityContext); + + const { cwd, isScratch } = await resolveWorkdir(context); + const artifactsDirectory = path.join(cwd, '.agent-harness-artifacts'); + + const abortController = new AbortController(); + const onCancelled = (): void => abortController.abort(); + activityContext.cancellationSignal.addEventListener('abort', onCancelled, { once: true }); + if (activityContext.cancellationSignal.aborted) abortController.abort(); + + const heartbeatTimer = setInterval(() => { + try { + activityContext.heartbeat(); + } catch { + // Best-effort; a throwing heartbeat must never crash the node run. + } + }, HEARTBEAT_INTERVAL_MS); + + let idleTimedOut = false; + let cancelled = false; + + try { + // Credential delivery (M3) — env vars merged into the provider request; + // any files land under the scratch/workdir's artifacts directory. + let credentialEnv: Record = {}; + if (node.config.credentialVendor && deps.credential) { + await mkdir(artifactsDirectory, { recursive: true }); + const delivery = deliverCredential(node.config.credentialVendor, deps.credential, { + artifactsDir: artifactsDirectory, + }); + credentialEnv = delivery.env; + for (const file of delivery.files ?? []) { + await mkdir(path.dirname(file.path), { recursive: true }); + await writeFile(file.path, file.contents); + } + } + + // Parity with `ai-agent.ts`: resolve `{{namespace.path}}` references, then + // prepend upstream node outputs so a prompt with no explicit reference + // still receives trigger/upstream content (mirrors ai-agent's fallback + // `userPrompt` — see ai-agent.ts for the twin of this block). + const resolvedPrompt = resolveTemplate(node.config.prompt, context); + const previousOutputs = Object.entries(context.nodeOutputs); + const contextBlock = + previousOutputs.length > 0 + ? `Context from previous steps:\n\n${previousOutputs + .map(([nodeId, output]) => { + const text = + typeof output === 'string' + ? output + : typeof output === 'object' && + output !== null && + typeof (output as Record)['response'] === 'string' + ? ((output as Record)['response'] as string) + : JSON.stringify(output); + return `[${nodeId}]:\n${text}`; + }) + .join('\n\n')}\n\n---\n\n` + : ''; + const finalPrompt = `${contextBlock}${resolvedPrompt}`; + + // Fail fast rather than relying on the idle timeout (default 30 min) to + // eventually notice nothing was ever sent to the provider. + if (finalPrompt.trim() === '') { + throw toHostNodeExecutionError('FATAL', 'agent_harness.empty_prompt', `Node '${node.id}' has an empty prompt.`); + } + + const provider = (deps.getProvider ?? getAgentProvider)(node.config.provider); + + const excludeDirectories = checkoutSnapshotExcludes(artifactsDirectory); + const mutatesCheckout = node.config.mutatesCheckout; + const checkoutSnapshotBefore = + mutatesCheckout === false ? await snapshotCheckout(cwd, excludeDirectories) : undefined; + + const idleTimeoutMs = node.config.idle_timeout ?? STEP_IDLE_TIMEOUT_MS; + + let responseText = ''; + let tokens: TokenUsage | undefined; + const warnings: string[] = []; + let sawError: { message: string } | undefined; + + const backgroundTasks = createBackgroundTaskTracker(); + + const stream = provider.sendQuery(finalPrompt, cwd, undefined, { + abortSignal: abortController.signal, + nodeConfig: node.config, + env: credentialEnv, + }); + + for await (const chunk of withIdleTimeout(stream, idleTimeoutMs, () => { + idleTimedOut = true; + abortController.abort(); + })) { + responseText = accumulateChunk(chunk, responseText, warnings, backgroundTasks, (sawError_) => { + sawError = sawError_; + }); + if (chunk.type === 'result') { + tokens = mergeTokenUsage([tokens, chunk.tokens].filter((t): t is TokenUsage => t !== undefined)); + if (backgroundTasks.shouldBreakOnResult()) break; + } + } + + cancelled = abortController.signal.aborted && !idleTimedOut; + + // OUTSIDE any retry path (there is none in v1) — a mutation turns a + // "successful" stream into a non-retryable failure. + if (mutatesCheckout === false && !cancelled) { + await assertCheckoutUntouched(node.id, cwd, excludeDirectories, checkoutSnapshotBefore); + } + + if (cancelled) { + throw toHostNodeExecutionError('TRANSIENT', 'agent_harness.cancelled', `Node '${node.id}' was cancelled.`); + } + + if (sawError) { + const error = new Error(sawError.message); + const errorType = classifyError(error); + throw toHostNodeExecutionError(errorType, 'agent_harness.provider_error', sawError.message, { + cause: error, + }); + } + + if (responseText.trim() === '') { + const message = idleTimedOut + ? `Node '${node.id}' timed out with no output (idle for ${String(idleTimeoutMs / 60_000)} min).` + : `Node '${node.id}' produced no assistant output.`; + throw toHostNodeExecutionError('TRANSIENT', 'agent_harness.empty_output', message); + } + + // Partial-output salvage (Archon L3087-3098): idle timeout with non-empty + // output completes successfully rather than failing — the agent likely + // finished but the subprocess didn't exit cleanly — surfaced as a warning + // instead of the UI fan-out message Archon sends (out of scope here). + if (idleTimedOut) { + warnings.push( + `Node '${node.id}' completed via idle timeout (no output for ${String(idleTimeoutMs / 60_000)} min). The AI likely finished but the subprocess didn't exit cleanly.`, + ); + } + + return { + output: { + response: responseText, + ...(tokens ? { tokens } : {}), + ...(warnings.length > 0 ? { warnings } : {}), + }, + }; + } catch (error) { + if (error instanceof NodeExecutionError) { + throw error; + } + const error_ = error instanceof Error ? error : new Error(String(error)); + const errorType = classifyError(error_); + log?.error('agent_harness.run_failed', { nodeId: node.id, error: { message: error_.message } }); + throw toHostNodeExecutionError(errorType, 'agent_harness.failed', error_.message, { cause: error_ }); + } finally { + clearInterval(heartbeatTimer); + activityContext.cancellationSignal.removeEventListener('abort', onCancelled); + if (isScratch) { + await rm(cwd, { recursive: true, force: true }).catch(() => { + // Best-effort — a leftover scratch dir under the OS tmpdir is not + // worth failing an otherwise-completed (or already-failed) run over. + }); + } + } +} + +/** + * Accumulate one MessageChunk into the running text/warnings state (Archon + * dag-executor.ts L2353-2700, trimmed — the UI streaming fan-out calls in + * that switch are dropped; only accumulation survives). + */ +function accumulateChunk( + chunk: MessageChunk, + responseText: string, + warnings: string[], + backgroundTasks: ReturnType, + markError: (error: { message: string } | undefined) => void, +): string { + switch (chunk.type) { + case 'assistant': { + return responseText + chunk.content; + } + case 'assistant_turn_boundary': { + // A new turn starts (e.g. after a tool call): only the LAST turn's + // text is the final answer — a planning preamble before a tool call + // must not survive concatenated onto the summary that follows it. + return ''; + } + case 'system': { + warnings.push(chunk.content); + return responseText; + } + case 'background_tasks': { + backgroundTasks.update(chunk.tasks); + return responseText; + } + case 'result': { + if (chunk.isError && chunk.errorSubtype !== 'success') { + const detail = chunk.errors?.length ? ` — ${chunk.errors.join('; ')}` : ''; + markError({ message: `Node failed: SDK returned ${chunk.errorSubtype ?? 'unknown'}${detail}` }); + } + return responseText; + } + default: { + return responseText; + } + } +} diff --git a/apps/execution-worker/src/activities/ai-agent.test.ts b/apps/execution-worker/src/activities/ai-agent.test.ts index 277f987b9..936b32b3f 100644 --- a/apps/execution-worker/src/activities/ai-agent.test.ts +++ b/apps/execution-worker/src/activities/ai-agent.test.ts @@ -1,10 +1,11 @@ import { APICallError } from 'ai'; import { MockLanguageModelV3 } from 'ai/test'; -import { describe, expect, it } from 'vitest'; +import { describe, expect, it, vi } from 'vitest'; import type { ExecutionContext } from '@workflow-builder/execution-core'; import type { AiAgentNode } from '../domain/ai-studio-nodes'; +import type { OpenRouterClient } from '../model-provider'; import { executeAiAgent } from './ai-agent'; function context(): ExecutionContext { @@ -64,3 +65,65 @@ describe('executeAiAgent', () => { expect(model.doGenerateCalls).toHaveLength(1); }); }); + +function mockModel(text: string) { + return new MockLanguageModelV3({ + doGenerate: { + content: [{ type: 'text', text }], + finishReason: { unified: 'stop', raw: undefined }, + usage: { + inputTokens: { total: undefined, noCache: undefined, cacheRead: undefined, cacheWrite: undefined }, + outputTokens: { total: undefined, text: undefined, reasoning: undefined }, + }, + warnings: [], + }, + }); +} + +function fakeOpenrouter(chat: (id: string) => unknown): OpenRouterClient { + return { chat: vi.fn(chat) } as unknown as OpenRouterClient; +} + +describe('executeAiAgent model/provider fallback chain', () => { + it("uses node.config.model over deps.defaultModel when resolving via 'auto'/openrouter", async () => { + const chat = vi.fn(() => mockModel('ok')); + const openrouter = fakeOpenrouter(chat); + const node: AiAgentNode = { + id: 'agent1', + type: 'ai-studio/ai-agent', + config: { systemPrompt: 'p', model: 'node-model' }, + }; + + await executeAiAgent(node, context(), { openrouter, defaultModel: 'env-model' }); + + expect(chat).toHaveBeenCalledWith('node-model'); + }); + + it('falls back to deps.defaultModel when node.config.model is unset', async () => { + const chat = vi.fn(() => mockModel('ok')); + const openrouter = fakeOpenrouter(chat); + const node: AiAgentNode = { + id: 'agent1', + type: 'ai-studio/ai-agent', + config: { systemPrompt: 'p' }, + }; + + await executeAiAgent(node, context(), { openrouter, defaultModel: 'env-model' }); + + expect(chat).toHaveBeenCalledWith('env-model'); + }); + + it('throws when node.config.provider selects a known provider whose API key is missing', async () => { + const openrouter = fakeOpenrouter(() => mockModel('unused')); + delete process.env['OPENAI_API_KEY']; + const node: AiAgentNode = { + id: 'agent1', + type: 'ai-studio/ai-agent', + config: { systemPrompt: 'p', model: 'gpt-4o-mini', provider: 'openai' }, + }; + + await expect(executeAiAgent(node, context(), { openrouter, defaultModel: 'env-model' })).rejects.toThrow( + 'OPENAI_API_KEY', + ); + }); +}); diff --git a/apps/execution-worker/src/activities/ai-agent.ts b/apps/execution-worker/src/activities/ai-agent.ts index 313b3f424..390ee43bf 100644 --- a/apps/execution-worker/src/activities/ai-agent.ts +++ b/apps/execution-worker/src/activities/ai-agent.ts @@ -3,13 +3,21 @@ import { generateText, stepCountIs } from 'ai'; import { type ExecutionContext, type LoggerPort, resolveTemplate } from '@workflow-builder/execution-core'; import type { AiAgentNode } from '../domain/ai-studio-nodes'; +import type { OpenRouterClient } from '../model-provider'; +import { resolveModel } from '../model-provider'; import { createWebSearchTool } from '../tools/web-search'; // Bounds the agentic tool loop so a misbehaving model can't run up cost. const MAX_TOOL_STEPS = 4; type AiAgentDeps = { - model: Parameters[0]['model']; + // A fixed model bypasses per-node resolution entirely (used by tests). + // A default model id/provider plus an OpenRouter client resolves per node, + // via node.config.model/provider falling back to these defaults. + model?: Parameters[0]['model']; + defaultModel?: string; + defaultProvider?: string; + openrouter?: OpenRouterClient; logger?: LoggerPort; tavilyApiKey?: string; }; @@ -44,9 +52,17 @@ export async function executeAiAgent(node: AiAgentNode, context: ExecutionContex const webSearchEnabled = node.config.webSearch === true && Boolean(deps.tavilyApiKey); const tools = webSearchEnabled ? { webSearch: createWebSearchTool(deps.tavilyApiKey!) } : undefined; + const model = + deps.model ?? + (await resolveModel( + node.config.model ?? deps.defaultModel!, + node.config.provider ?? deps.defaultProvider ?? 'auto', + deps.openrouter!, + )); + try { const result = await generateText({ - model: deps.model, + model, // Temporal's activity retry policy owns retries; SDK retries on top would // multiply model calls (up to 3x per activity attempt). maxRetries: 0, diff --git a/apps/execution-worker/src/agent-harness/README.md b/apps/execution-worker/src/agent-harness/README.md new file mode 100644 index 000000000..59c4514b5 --- /dev/null +++ b/apps/execution-worker/src/agent-harness/README.md @@ -0,0 +1,104 @@ +# agent-harness + +## Attribution + +Portions of this module are derived from [coleam00/Archon](https://github.com/coleam00/Archon) +(MIT © 2025-2026 Cole Medin). Archon is not depended upon as a library — its provider +abstraction, credential delivery, structured-output handling, and Copilot adapter are +adapted (not copied wholesale) to fit workflowbuilder's Temporal-activity / hexagonal-ports +architecture. See the upstream repository for the original implementation and its `LICENSE`. + +Modules derived from Archon (ported incrementally, milestones M1-M9 of the porting plan): + +- Provider contract types (`types.ts`, `errors.ts`) — from `packages/providers/src/{types,errors}.ts` +- Provider registry (`registry.ts`) — from `packages/providers/src/registry.ts` +- Credential delivery (`credentials/delivery.ts`, `credentials/catalog.ts`) — from + `packages/core/src/credentials/{delivery,catalog}.ts` +- Shared execution utilities (`shared/*.ts`) — from `packages/providers/src/shared/*.ts`, + `packages/workflows/src/utils/idle-timeout.ts`, and `packages/workflows/src/executor-shared.ts` +- Copilot provider adapter (`providers/copilot/*.ts`) — from `community/copilot/*.ts` + +## Architecture intent + +This directory hosts a worker-local port of Archon's provider abstraction, scoped to a +single functional provider (GitHub Copilot) for v1. It backs the `ai-studio/agent-harness` +node type that delegates a workflow step to an external autonomous coding-agent CLI. +Execution runs inside a Temporal activity (`activities/agent-harness.ts`) rather than +Archon's own DAG executor, since activity code must remain replay-safe and Archon's +execution model (Bun-compiled binary, multi-tenant credential vault) does not map onto +workflowbuilder's architecture. + +## Directory structure + +| Path | What it does | +| -------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | +| `types.ts` | Provider contract types: `TokenUsage`/`mergeTokenUsage`, the `MessageChunk` union, `NodeConfig` (the cross-provider parity surface), `ProviderCapabilities`, `IAgentProvider`. | +| `errors.ts` | Provider-level error classes. | +| `registry.ts` | Static provider registration (single entry: Copilot) + capability lookup. | +| `credentials/delivery.ts` | Turns a resolved credential into `{ env, files? }` for a given vendor; only `github-copilot` is live, `anthropic`/`openai` are kept as commented reference cases. | +| `credentials/catalog.ts` | Declares which credential kinds each vendor accepts (trimmed to Copilot). | +| `shared/binary-resolution.ts` | Generic "is this an executable file" helper used by provider CLI resolution. | +| `shared/run-config.ts` | Shared run-config validation helpers (`assertKnownRunConfigKeys`, etc.). | +| `shared/idle-timeout.ts` | `withIdleTimeout` — wraps an async generator so it aborts if no chunk arrives within a window; `STEP_IDLE_TIMEOUT_MS` default. | +| `shared/error-classification.ts` | `classifyError`/`isRateLimitError`/`formatSubprocessFailure` — maps raw error text onto FATAL/TRANSIENT/UNKNOWN. | +| `providers/copilot/capabilities.ts` | Copilot's honest capability flags (what it actually supports). | +| `providers/copilot/config.ts` | Lenient (`parseCopilotConfig`) and strict (`parseCopilotRunConfig`) config parsers, effort clamping. | +| `providers/copilot/binary-resolver.ts` | Resolves the `copilot` CLI binary: `COPILOT_CLI_PATH` env → config path → `PATH` lookup → throw. | +| `providers/copilot/event-bridge.ts` | Maps the Copilot SDK's native event stream onto the `MessageChunk` union, incl. usage normalization. | +| `providers/copilot/provider.ts` | `IAgentProvider` implementation: env/token resolution, `NodeConfig` → `SessionConfig` translation, `sendQuery`. | + +## Per-file provenance + +Every ported file's Archon source (consolidated from the porting plan's `PORTING-MAP.md`): + +| workflowbuilder file | Archon source | Fidelity | +| -------------------------------------- | ---------------------------------------------------------------- | ---------------------------------------------------------- | +| `types.ts`, `errors.ts` | `packages/providers/src/{types,errors}.ts` | Trimmed / Verbatim | +| `registry.ts` | `packages/providers/src/registry.ts` | Adapted (static map, not lazy dynamic-import) | +| `credentials/delivery.ts` | `packages/core/src/credentials/delivery.ts` | Trimmed (Copilot case only, others as commented reference) | +| `credentials/catalog.ts` | `packages/core/src/credentials/catalog.ts` | Trimmed (Copilot only) | +| `shared/binary-resolution.ts` | `packages/providers/src/shared/binary-resolution.ts` | Verbatim | +| `shared/run-config.ts` | `packages/providers/src/shared/run-config.ts` | Verbatim | +| `shared/idle-timeout.ts` | `packages/workflows/src/utils/idle-timeout.ts` | Verbatim | +| `shared/error-classification.ts` | `packages/workflows/src/executor-shared.ts` (selected functions) | Verbatim | +| `providers/copilot/capabilities.ts` | `community/copilot/capabilities.ts` | Verbatim | +| `providers/copilot/config.ts` | `community/copilot/config.ts` | Verbatim | +| `providers/copilot/binary-resolver.ts` | `community/copilot/binary-resolver.ts` | Trimmed ~205 → ~50 LOC (A4) | +| `providers/copilot/event-bridge.ts` | `community/copilot/event-bridge.ts` | Verbatim | +| `providers/copilot/provider.ts` | `community/copilot/provider.ts` | Trimmed (A4 import style, OAuth branches dropped) | +| `../activities/agent-harness.ts` | `packages/workflows/src/dag-executor.ts` (selected slices) | Adapted into a Temporal activity (A1/A2/A3) | + +See [`PORTING-MAP.md`](/tmp/opencode/workflowbuilder/PORTING-MAP.md) (outside this repo, plan working directory) for the full line-range breakdown. + +## Adaptations from Archon (A1-A8) + +This is a port, not a redesign — but a handful of host-architecture mismatches forced +deliberate deviations. These are the _only_ permitted deviations; anything else that +differs from Archon is a defect. + +| # | Archon does | We do instead | Why | +| --- | ---------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| A1 | Runs the streaming loop inside its own `dag-executor` process | Runs inside a **Temporal activity** function | `runGraph` is sandboxed and replay-deterministic; no I/O may touch it. | +| A2 | `withIdleTimeout` is the only liveness mechanism | Additionally calls `Context.current().heartbeat()` on an interval, with `heartbeatTimeout` set on the activity profile | Spike-verified: without heartbeating, `handle.cancel()` is silently ignored and the activity runs to completion. | +| A3 | `child.kill()` / SDK abort | The Copilot SDK's own `session.abort()`/`client.stop()` — **no manual process-group kill was implemented** | See "SDK abort supersedes A3" below — empirically the SDK's own abort cleanly kills the underlying process tree; a manual `spawn(detached)+process.kill(-pid)` workaround was not needed. | +| A4 | 6-step binary resolution chain for `bun --compile` binaries | Plain `node_modules`/`PATH` resolution + a single `COPILOT_CLI_PATH` env override | Worker is a normal Node/Docker process; Archon's chain solves a problem we don't have. | +| A5 | Encrypted multi-tenant credential vault (DB rows, envelope encryption) | Read credential from worker env config | Single-tenant reference stack. Delivery is ported faithfully; storage is not. | +| A6 | `cwd` from a `codebases` DB row (`kind: 'repo' \| 'folder'`) | Workflow-context lookup → scratch temp dir fallback | No codebase concept exists yet; Archon's `kind: 'folder'` fallback semantics map directly onto our fallback. | +| A7 | Hand-rolled React `NodeInspector.tsx` (tabs, raw `