Workflows v2 Migration

Workflows v2 moves durable workflow execution from the Python API service into api-rs. The workflow state machine is backed by Absurd queues and checkpoints, while Python workflow handlers run in their own sandbox through the Python workflow host.

This keeps the workflow programming model familiar, but changes what workflow files can assume about their runtime.

What changes

What keeps working

Most workflow handlers can keep the same shape:

from dataclasses import dataclass
from typing import Any
 
from api.workflow_engine import WorkflowContext
 
 
WORKFLOW_NAME = "nightly_report"
 
 
@dataclass
class Input:
    topic: str
 
 
async def handler(inp: Input, ctx: WorkflowContext) -> dict[str, Any]:
    facts = await ctx.step("collect", lambda: {"topic": inp.topic})
    result = await ctx.agent_turn(f"Summarize {facts['topic']}")
    return {"result": result}

Supported v2 primitives:

Required migrations

Keep imports narrow

Workflow files should import only the supported workflow-host API surface they need:

from api.workflow_engine import WorkflowContext
from api.runtime_control import ControlPlaneError

Supported workflow-host modules are api.workflow_engine, api.runtime_control, api.app, and api.metrics.

Do not import unrelated API-service internals or another workflow domain's local helpers. Domain-specific helpers should live next to the workflows that own them, for example workflows/slack/metrics.py.

If a workflow needs a helper, move it into the workflow file, a shared overlay module, or a supported workflow-host API module.

Put side effects behind steps

The handler may be replayed after a crash or retry. Any external write should be wrapped in ctx.step(...) so the result is checkpointed:

async def handler(inp: dict, ctx: WorkflowContext) -> dict:
    posted = await ctx.step(
        "post_summary",
        lambda: ctx.post_to_slack(inp["channel"], inp["summary"]),
    )
    return {"posted": posted}

Make agent turns explicit

Use ctx.agent_turn(...) when the workflow needs an agent sandbox:

result = await ctx.agent_turn(
    "Investigate this alert and return the next action.",
    thread_key=f"workflow:{ctx.run_id}:agent",
    harness="codex",
    model="gpt-5.2",
    reasoning="high",
    metadata={"workflow": WORKFLOW_NAME},
)

The workflow host sandbox is separate from the agent sandbox. The workflow handler coordinates the run; the agent turn runs through the normal Centaur session runtime.

Declare Workflow-Host Permissions

When a workflow calls tools directly from the workflow host with ctx.call_tool(...), declare the principal that should own those permissions:

WORKFLOW_NAME = "nightly_report"
WORKFLOW_PRINCIPAL = True

The API derives and registers the workflow-nightly-report principal in the Centaur Console and runs that workflow's host sandbox under it. Grant only the roles or secrets that workflow needs:

cargo run -p centaur-perms -- \
  principals grant workflow-nightly-report \
  --tool slack

The principal id is always workflow- plus the slugged WORKFLOW_NAME. Workflow code cannot choose another principal id, display name, or labels. WORKFLOW_PRINCIPAL = True requires apiRs.workflowHostSandbox=true, which renders WORKFLOW_HOST_SANDBOX=true; startup fails if a workflow declares a principal while workflow-host sandboxing is disabled.

Pick the model and reasoning effort

ctx.agent_turn(...) accepts optional model, provider, and reasoning kwargs. They ride the turn exactly like the Slack --model / --bedrock / -rsn flags: model selects the model within the harness, reasoning sets the codex reasoning effort (none/minimal/low/medium/high/xhigh/max), and provider selects the codex model provider. provider and reasoning only affect the codex harness; claude and amp ignore them. reasoning also accepts the reasoning_effort and effort aliases. When a kwarg is omitted the deployment/baked harness default stands — dispatched turns are no longer pinned to the deployment default.

To set a default for every turn in a workflow, declare a module-level AGENT_DEFAULTS dict. Explicit per-call kwargs override it key by key:

WORKFLOW_NAME = "nightly_report"
AGENT_DEFAULTS = {"harness": "codex", "model": "gpt-5.2", "reasoning": "high"}
 
async def handler(inp: Input, ctx: WorkflowContext) -> dict[str, Any]:
    await ctx.agent_turn("Draft the report.")            # gpt-5.2 @ high
    await ctx.agent_turn("Tidy formatting.", reasoning="low")  # gpt-5.2 @ low

Keep harness and model together — a model is only meaningful within its harness, and because kwargs override AGENT_DEFAULTS key by key, overriding one without the other can strand a model on the wrong harness.

Declare webhook metadata in the workflow

Expose a workflow through WEBHOOKS:

WORKFLOW_NAME = "github_issue_triage"
 
WEBHOOKS = [
    {
        "slug": "github-issue-triage",
        "provider": "github",
        "auth": {"type": "github_hmac", "secret_ref": "GITHUB_WEBHOOK_SECRET"},
        "trigger_key": {"type": "header", "name": "X-GitHub-Delivery"},
    }
]

The v2 webhook endpoint is:

POST /api/webhooks/{slug}

Webhook delivery is idempotent when trigger_key resolves to the same value. Sensitive headers are redacted before the webhook envelope is persisted.

Move schedules into workflow metadata

Schedules can live beside the handler:

SCHEDULE = {
    "type": "cron",
    "cron": "0 9 * * MON-FRI",
    "timezone": "America/New_York",
    "input": {"profile": "default"},
}

Write day-of-week as names (MON-FRI), not numbers: the parser is Quartz-style (1 = Sunday), so a Unix-style 1-5 fires Sunday–Thursday. See Schedule a workflow for details.

api-rs reconciles enabled schedule metadata into Absurd schedule tasks. ETL workflows can be routed to a separate queue so long-running sync jobs do not block normal workflow runs.

Audit direct database access

The middle migration path allows workflows to use the main database through ctx._pool. That keeps existing DB-heavy workflows moving, but it is not a hard isolation boundary.

Use this only for workflows that already own their tables or are explicitly part of the platform data path. Prefer explicit tool calls or narrowly scoped SQL helpers for new workflows.

Compatibility checklist

For each existing workflow:

1. Confirm the file exports WORKFLOW_NAME.
2. Confirm imports do not require the old Python API package, except api.workflow_engine.WorkflowContext.
3. Confirm third-party Python packages are installed in the workflow-host sandbox image.
4. Wrap Slack posts, database writes, external HTTP calls, and tool calls in ctx.step(...) when they must not repeat.
5. Replace direct agent-control-plane calls with ctx.agent_turn(...).
6. If the workflow uses ctx._pool, confirm the workflow-host sandbox receives DATABASE_URL.
7. If the workflow is scheduled, add SCHEDULE metadata and verify the schedule queue has a sleeping tick task.
8. If the workflow is webhook-triggered, add WEBHOOKS metadata and verify repeated deliveries return the same run id.

Known gaps

The v2 workflow host intentionally exposes a narrow Python API package. Workflows that import unrelated API-service internals should move that behavior into the workflow-host API surface or a small local helper owned by the workflow domain before they are v2-ready.

ctx.call_tool(...) is a compatibility surface in the Python workflow host. It uses the generated centaur-tools call bridge against the installed tool package; agent sandboxes should use direct tool CLIs instead of deprecated /tools/... HTTP routes.

Verify a migration

Start with an import and discovery check in the same image that production will run:

WORKFLOW_DIRS=/opt/centaur/workflows python3 /usr/local/bin/workflow-host <<'EOF'
{"type":"discover"}
EOF

Then create a real run:

curl -s "$CENTAUR_API_URL/api/workflows/runs" \
  -H "Content-Type: application/json" \
  -d '{
    "workflow_name": "nightly_report",
    "input": {"topic": "open incidents"}
  }' | jq

Inspect the run, checkpoints, and sandbox state. A migrated workflow is not done until it has completed through the api-rs runtime in the same sandbox image and database configuration used by the deployment.