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"],
            username="Summary Bot",
            icon_emoji=":memo:",
        ),
    )
    return {"posted": posted}

username and icon_emoji are optional. They customize the Slack app identity for that message and require the app installation to grant chat:write.customize. Omitting them preserves the app's default identity.

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.

ctx.agent_turn(...) resolves to the execution record, not to the answer. Read result_text for the agent's final text:

{
    "thread_key": "wf:<task-id>:agent:<workflow-name>",
    "execution_id": "...",
    "status": "completed",
    "output_lines": [...],
    "result_text": "the agent's final text",
}

Use ctx.run_agents(...) for independent work that should run concurrently:

reviews = await ctx.run_agents(
    [
        {"name": "correctness", "text": "Review the PR for correctness."},
        {"name": "security", "text": "Review the PR for security issues."},
    ],
    max_concurrency=2,
)

Each item uses a separate workflow-owned session. Results stay in input order. An individual agent failure produces an item with ok: false; it does not fail the whole batch or discard successful results.

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

To use an existing principal instead, declare its foreign ID or prn_-prefixed OID:

WORKFLOW_PRINCIPAL = "finance-automation"

The string form resolves the existing principal by foreign ID or OID and fails startup when it does not exist. It does not create or update the principal. The True form remains workflow- plus the slugged WORKFLOW_NAME. Any WORKFLOW_PRINCIPAL value 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.

Agent turns can also select an existing principal by foreign ID. This applies to ctx.agent_turn(...), its aliases, and each ctx.run_agents(...) item:

result = await ctx.agent_turn(
    "Prepare the finance report.",
    principal="finance-automation",
)

Principal lookup failures fail the turn. Put principal in AGENT_DEFAULTS when every agent turn in the workflow should use the same principal. An existing session bound to another principal returns a conflict and is not rebound.

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:

CENTAUR_API_TOKEN=$(kubectl exec -n centaur-system deploy/centaur-centaur-console -- \
  bin/rails runner 'print ApiServer::Jwt.encode_for_console_service')
 
curl -s "$CENTAUR_API_URL/api/workflows/runs" \
  -H "Authorization: Bearer $CENTAUR_API_TOKEN" \
  -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.