diff --git a/apps/backend/src/routes/workflows.ts b/apps/backend/src/routes/workflows.ts index cb7906c85..96e773285 100644 --- a/apps/backend/src/routes/workflows.ts +++ b/apps/backend/src/routes/workflows.ts @@ -255,7 +255,7 @@ export function createWorkflowsRoutes( executionId: execution.id, definition, triggerPayload: body.triggerPayload ?? {}, - variables: {}, // server-side globals (secrets, env) populated here later + variables: {}, global: {}, }); diff --git a/apps/docs/src/content/docs/guides/use-variable-picker.mdx b/apps/docs/src/content/docs/guides/use-variable-picker.mdx index 5118bf50e..c1f7cd514 100644 --- a/apps/docs/src/content/docs/guides/use-variable-picker.mdx +++ b/apps/docs/src/content/docs/guides/use-variable-picker.mdx @@ -84,7 +84,7 @@ Typing `{{` opens the suggestions panel, which only lists ancestor-node properti ### `variables.` -Globals and secrets available at the start of each run. **Manual entry only - not surfaced in the picker today.** Type the reference yourself: +Non-secret server-side run config that your backend sets when it submits the run; the reference backend sends an empty object. Keep secrets out. Every value here is recorded in the run's execution history. **Manual entry only - not surfaced in the picker today.** Type the reference yourself: ``` {{variables.apiBaseUrl}} diff --git a/apps/execution-worker/README.md b/apps/execution-worker/README.md index a9aa5d107..21b9c472a 100644 --- a/apps/execution-worker/README.md +++ b/apps/execution-worker/README.md @@ -26,14 +26,15 @@ Requires Postgres + Temporal running. Start them with `pnpm infra:up`. ## Environment -See `.env.example`. Required: +See `.env.example`. -| Var | Purpose | Default | -| -------------------- | ---------------------------------- | ---------------------------------------------------- | -| `OPENROUTER_API_KEY` | AI agent activities (**required**) | — | -| `DATABASE_URL` | Execution events + status | `postgresql://wb:wb@127.0.0.1:5432/workflow_builder` | -| `TEMPORAL_ADDRESS` | Temporal server address | `127.0.0.1:7233` | -| `AI_MODEL` | OpenRouter model ID | `anthropic/claude-3.5-haiku` | +| Var | Purpose | Default | +| -------------------- | ----------------------------------- | ---------------------------------------------------- | +| `OPENROUTER_API_KEY` | AI agent activities (**required**) | — | +| `DATABASE_URL` | Execution events + status | `postgresql://wb:wb@127.0.0.1:5432/workflow_builder` | +| `TEMPORAL_ADDRESS` | Temporal server address | `127.0.0.1:7233` | +| `AI_MODEL` | OpenRouter model ID | `mistralai/mistral-small-3.2-24b-instruct` | +| `TAVILY_API_KEY` | AI agent web-search tool (optional) | — | ## Structure @@ -61,6 +62,17 @@ own: one executor per node type and the database as the store port. - **Editing the package:** the worker imports its built `dist`, so run `pnpm build:temporal` after changing `packages/temporal/src`. - **Deploys that change the emitted event set:** drain in-flight runs first. Replaying an old run's history against a new emit sequence diverges — see [`replay-audit.md`](../../packages/execution-core/replay-audit.md) rule 9. +## The AI agent's tool loop + +The reference AI Agent node runs the AI SDK's tool loop inside one activity. The loop stops after `MAX_TOOL_STEPS` steps (`src/activities/ai-agent.ts`). At the cap `generateText` returns normally, so the node completes with whatever the last step produced, possibly empty text, rather than failing. The cap only applies when tools are on, which needs both the node's `webSearch` config flag and a `TAVILY_API_KEY`. Without them the activity makes one model call and there is no loop. + +This is deliberate: `runGraph` walks a DAG, so it cannot express a loop whose length the model decides at run time. What it costs: + +- **A retry re-runs the whole loop.** Tool calls included, so a tool with side effects can run more than once for one node. `generateText` runs with `maxRetries: 0`, so one activity attempt is exactly one pass. To rule the re-run out, give `ai-studio/ai-agent` its own profile through `createRunWorkflow({ nodeActivityProfiles })`, setting `retry.maximumAttempts: 1`. Every profile also has to declare `startToCloseTimeout`, so pass both. See the package README. +- **Temporal records nothing until the activity returns.** If the worker dies mid-loop, every finished step is lost, and the whole loop shares the node profile's single `startToCloseTimeout` (see Temporal specifics above). + +There is no durable per-step option today. A fixed sequence of model calls can be split into one node per call, each its own activity that Temporal records and resumes on its own. A model-driven loop cannot: a cycle through the start node is rejected before the run starts, and any other cycle fails the run with `Workflow stalled` once the reachable nodes have run. No tool-call node ships, so a durable tool call is an executor you write yourself. + ## Adding a new engine 1. Create `src/engines//` with: diff --git a/packages/execution-core/README.md b/packages/execution-core/README.md index 81f4587d2..aa8c91a30 100644 --- a/packages/execution-core/README.md +++ b/packages/execution-core/README.md @@ -205,9 +205,9 @@ Failure always wins. An unhandled node failure returns before the terminal check ## Recorded step inputs and payload redaction -Every `node_started` event records what the step was given: `{ config, visibleNodeIds }` — the node's frozen config plus the ids of every output visible at start (the same wave snapshot the executor receives; executors and templates may read any completed node's output, not just direct predecessors). The output _values_ are deliberately not copied: each is already recorded exactly once, at a lower sequence, in its own `node_completed` event — or `node_failed`, for a failure absorbed by errorPolicy `'continue'`/`'errorRoute'` — so copying them would grow Postgres, Temporal history, and the SSE stream quadratically with graph depth while adding no information. `reconstructNodeInputs(events, nodeId)` joins the ids back to the values and returns `{ config, nodeOutputs }`, a faithful record for diagnosing a failure or replaying a step with corrected input. `variables`, `global`, and `triggerPayload` are not recorded: the first is the server-side secrets bag, and the trigger payload is already frozen on the execution row. Events recorded before inputs were captured carry no payload, which is why `NodeStartedEvent.payload` is optional — `reconstructNodeInputs` returns `undefined` for those. +Every `node_started` event records what the step was given: `{ config, visibleNodeIds }` — the node's frozen config plus the ids of every output visible at start (the same wave snapshot the executor receives; executors and templates may read any completed node's output, not just direct predecessors). The output _values_ are deliberately not copied: each is already recorded exactly once, at a lower sequence, in its own `node_completed` event — or `node_failed`, for a failure absorbed by errorPolicy `'continue'`/`'errorRoute'` — so copying them would grow Postgres, Temporal history, and the SSE stream quadratically with graph depth while adding no information. `reconstructNodeInputs(events, nodeId)` joins the ids back to the values and returns `{ config, nodeOutputs }`, a faithful record for diagnosing a failure or replaying a step with corrected input. `variables`, `global`, and `triggerPayload` are not recorded. The first two are identical for every step of the run, and the trigger payload is already frozen on the execution row. Events recorded before inputs were captured carry no payload, which is why `NodeStartedEvent.payload` is optional — `reconstructNodeInputs` returns `undefined` for those. -Every payload — inputs, outputs, errors — passes through `redactSensitive` (`redact.ts`) before it crosses `EventEmitterPort`. Event history is immutable on every surface it lands on (Postgres, SSE, and Temporal's own history via activity args), so secrets must never reach it in the first place; redacting inside the runner, before the emit activity, is what keeps all three surfaces clean. Matching is key-based (`SENSITIVE_KEY_RULES`: `apiKey`, `secret`, `password`, `*token`, …) and replaces the whole subtree under a matched key with `'[REDACTED]'`. The walk is copy-on-write — the unredacted objects continue on into execution — and depth-capped (see replay-audit rule 8). Secrets arriving through _values_ rather than keys (e.g. a resolved `{{variables.x}}` template) are not caught; that belongs to the variables feature (follow-up: value-based-redaction). Encrypting Temporal's own history, where `executeNode` activity args still carry real values, is planned separately (follow-up: temporal-payload-codec). +Every payload — inputs, outputs, errors — passes through `redactSensitive` (`redact.ts`) before it crosses `EventEmitterPort`. Event history is immutable on every surface it lands on (Postgres, SSE, and Temporal's own history via activity args), so secrets must never reach it in the first place; redacting inside the runner, before the emit activity, is what keeps every emitted payload clean on all three. Matching is key-based (`SENSITIVE_KEY_RULES`: `apiKey`, `secret`, `password`, `*token`, …) and replaces the whole subtree under a matched key with `'[REDACTED]'`. The walk is copy-on-write — the original objects continue on into execution — and depth-capped (see replay-audit rule 8). A secret arriving as a _value_ rather than under a matched key — a resolved template, say — is not caught (follow-up: value-based-redaction). Encrypting Temporal's own history, where `executeNode` activity args still carry real values, is planned separately (follow-up: temporal-payload-codec). ## Template references diff --git a/packages/execution-core/src/execution-context.ts b/packages/execution-core/src/execution-context.ts index de61f7ed1..2579146f2 100644 --- a/packages/execution-core/src/execution-context.ts +++ b/packages/execution-core/src/execution-context.ts @@ -3,10 +3,10 @@ export type ExecutionContext = { readonly executionId: string; readonly triggerPayload: Record; readonly nodeOutputs: Record; - // Server-side globals/secrets the backend injects at execution start. - // Values cross into Temporal event history unredacted via `executeNode` - // activity args — encrypting that history is planned - // (follow-up: temporal-payload-codec). + /** + * Non-secret server-side run config. Everything here reaches Temporal Event History + * unredacted. See the `@workflowbuilder/temporal` README, "What Event History records". + */ readonly variables: Record; // Global variables defined manually in the builder readonly global: Record; diff --git a/packages/execution-core/src/graph-runner.ts b/packages/execution-core/src/graph-runner.ts index 7943ab920..fe4b3c676 100644 --- a/packages/execution-core/src/graph-runner.ts +++ b/packages/execution-core/src/graph-runner.ts @@ -58,9 +58,9 @@ export async function runGraph( runner: ActivityRunnerPort, rawEvents: EventEmitterPort, ): Promise { - // Every payload is redacted before it crosses the emit boundary — event history - // (DB, SSE, Temporal's own history via activity args) is immutable, so secrets - // must never reach it in the first place. + // Every emitted payload is redacted before it crosses the emit boundary — event + // history (DB, SSE, Temporal's record of the emitEvent args) is immutable, so + // secrets must never reach it in the first place. const events = withRedactedPayloads(rawEvents); const adjacency = buildAdjacencyMap(input.definition.nodes, input.definition.edges); diff --git a/packages/execution-core/src/ports/workflow-engine.port.ts b/packages/execution-core/src/ports/workflow-engine.port.ts index daed6c6f4..1c8806923 100644 --- a/packages/execution-core/src/ports/workflow-engine.port.ts +++ b/packages/execution-core/src/ports/workflow-engine.port.ts @@ -5,6 +5,8 @@ export type WorkflowExecutionInput = { executionId: string; definition: WorkflowDefinition; triggerPayload: Record; + // Non-secret server-side run config. Everything here reaches Temporal Event History + // unredacted. See the `@workflowbuilder/temporal` README, "What Event History records". variables: Record; global: Record; }; diff --git a/packages/execution-core/src/redact.ts b/packages/execution-core/src/redact.ts index 8310c2762..0eacf5d0f 100644 --- a/packages/execution-core/src/redact.ts +++ b/packages/execution-core/src/redact.ts @@ -9,10 +9,9 @@ import type { EventEmitterPort } from './ports/event-emitter.port'; // shared across sibling nodes — mutating here would feed '[REDACTED]' into // execution itself. // -// Matching is key-based only. Secrets arriving through VALUES (e.g. a resolved -// `{{variables.x}}` template) are not caught — that belongs to the variables -// feature (follow-up: value-based-redaction). Encrypting Temporal's own event -// history, where activity args land unredacted, is planned separately +// Matching is key-based only. A secret arriving as a VALUE rather than under a +// matched key is not caught (follow-up: value-based-redaction). Encrypting Temporal's +// own event history, where executeNode args land unredacted, is planned separately // (follow-up: temporal-payload-codec). export const REDACTED = '[REDACTED]'; diff --git a/packages/temporal/README.md b/packages/temporal/README.md index 0d94a4bad..e0104279f 100644 --- a/packages/temporal/README.md +++ b/packages/temporal/README.md @@ -189,6 +189,62 @@ Filling in `node.label` belongs to whatever builds the `WorkflowExecutionInput`, The attempt count is an upper bound rather than a promise. An executor that throws `PermanentNodeExecutionError` — for a rejected API key, say — is not retried at all: the activity adapter marks the failure non-retryable, which Temporal honours regardless of the profile. `TransientNodeExecutionError` says the opposite, that another attempt is worth making, but it does not raise the limit; the profile still caps it. Anything thrown unclassified retries exactly as it always has. Both classes are re-exported from this package, and a classified failure also records its error code and the attempt it died on in the `node_failed` event — see [`execution-core`](../execution-core/README.md#transient-vs-permanent-failures) for when to throw which. +## What Event History records + +Temporal writes every argument you give it into Event History. Replay reads that record back, so nothing can change it later. The record lives while the run is open, then for the namespace's retention period after the run closes. + +This package hands Temporal: + +- The workflow input. It carries the whole graph definition, so every node's `config` is in the record before the first node runs. +- The arguments and the return value of every `executeNode` call. The arguments are the node, `variables`, `global`, `triggerPayload`, and the output of every node that finished before the current wave started. +- The arguments of every `updateStatus` call, including the `errorMessage` of a failed run. +- The payload of every `emitEvent` call `runGraph` makes. This is the only one that `execution-core` redacts first. +- The Temporal Summary of every labelled node activity, copied from `node.label`. +- Every failure: the error's message, type and stack, and the run's own final failure. A classified error also carries a `NodeErrorEnvelope` in `details` and the original error in `cause`. + +### Secrets + +**Keep secret values out of everything this package hands Temporal. Only `emitEvent` payloads are redacted; nothing stops a secret anywhere else from reaching Event History.** + +Anyone who can open the Temporal UI can read the record. On Temporal Cloud, Temporal stores it on its own infrastructure. + +**You cannot edit a leaked key out of Event History. You can only delete the whole run. So rotate the key.** + +[`execution-core`](../execution-core/README.md#recorded-step-inputs-and-payload-redaction) redacts inside the workflow, before it calls the `emitEvent` activity, and matches by key name. Everything else in the list above stays as written. + +Two rules follow: + +- **Keep secrets in the worker's own environment.** The reference worker keeps its model key and its search key there. +- **If a secret must differ per run or per tenant, read it inside the executor.** Temporal never records what the executor does in its own body. It does record the return value and the message of any error, so keep the secret out of both, and out of the node's `config`, which travels in the workflow input. An executor is a closure: capture the secret in `executors` at worker start, as the reference worker does with its keys. + +A Payload Codec encrypts payloads before they leave the process. Pass one as `dataConverter` to `Worker.create` and to the `Client` you hand `TemporalWorkflowEngine`; nothing in this package stands in the way. The plugin ships no codec of its own (follow-up: temporal-payload-codec). + +### Payload size and the history budget + +Two [server defaults](https://docs.temporal.io/references/dynamic-configuration) bound a run. You can raise or lower them on a self-hosted deployment. On [Temporal Cloud](https://docs.temporal.io/cloud/limits) they are fixed. + +| What | Warns at | Fails at | Dynamic config | +| --------------------------------------------------------------------- | -------- | -------- | --------------------- | +| One payload set: an activity's input or result, or the workflow input | 512 KB | 2 MB | `limit.blobSize.*` | +| Event History size, per run | 10 MB | 50 MB | `limit.historySize.*` | + +A third default, `limit.historyCount.*`, caps a run at 10,240 events (warn) and 51,200 (fail). A chain costs about 18 events per node, so the size limits bind first unless node outputs are tiny. The continue-as-new suggestion Temporal raises at 4 MB or 4,096 events is inert here: this package never calls `continueAsNew`. + +Which limit you reach first depends on the size of your node outputs. `runGraph` gives every node the output of every node that finished before the current wave started, not only the output of its direct predecessors. So node 2 receives one output, node 3 receives two, and so on. + +Take a chain of N nodes each returning S bytes. The run writes about N(N+3)S ÷ 2 bytes into history. N(N−1)S ÷ 2 of that is `executeNode` arguments. The remaining 2NS is each output written twice more, once as the activity's result and once in its `node_completed` payload. Above about 45 KB per output the arguments reach 2 MB before history reaches 50 MB. Below that size, history goes first. Every call also carries `triggerPayload`, `variables`, `global` and the node itself regardless of S, and a large trigger payload is the usual reason that matters. + +At 100 KB per output, the `executeNode` arguments pass 512 KB at node 7. History passes 10 MB at node 13. The arguments pass 2 MB at node 22. + +What an oversize payload does depends on which one it is. The worker checks each outbound payload before sending it. Over the warn threshold it logs `[TMPRL1103]` at `WARN` and sends anyway. Over the error limit it logs at `ERROR` and fails the task instead. The error limit comes from the namespace. The warn threshold does not. It is the worker's own 512 KiB default, and only `NativeConnectionOptions.payloadLimits` moves it, not `limit.blobSize.warn`. That option is experimental, so expect it to change. + +- **`executeNode` arguments over the limit** fail the Workflow Task. Temporal retries that task, so the run hangs until you deploy a fix. +- **An `executeNode` return value over the limit** fails the activity attempt. The retry policy re-runs the node, the output is oversize again, and the node fails once the profile's attempts are spent. From there `errorPolicy` decides, as for any other node failure. + +`disablePayloadErrorLimit` on the worker skips the check and leaves the limit to the server. + +A large model answer is the usual oversize return value. Keep the bulk out of the node's output. Store it where your application already stores blobs, and return only an identifier for it, an S3 key or a row id. The next node then fetches the data itself. This is the claim-check pattern. This package ships no blob store and no fetch helper, so your executor has to do both sides. + ## Versioning and replay This package carries two contracts, not one. The API is the ordinary semver surface. The second is replay compatibility: a workflow can sit in Event History for days, and a new version of this package has to be able to replay a history that an older version recorded. diff --git a/packages/temporal/src/workflow/core-contract.ts b/packages/temporal/src/workflow/core-contract.ts index 0c6df939a..a62f6faa0 100644 --- a/packages/temporal/src/workflow/core-contract.ts +++ b/packages/temporal/src/workflow/core-contract.ts @@ -28,6 +28,10 @@ export type WorkflowExecutionInput = { executionId: string; definition: WorkflowDefinition; triggerPayload: Record; + /** + * Non-secret server-side run config. Everything here reaches Temporal Event History + * unredacted. See the `@workflowbuilder/temporal` README, "What Event History records". + */ variables: Record; global: Record; };