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",
    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 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 * * 1-5",
    "timezone": "America/New_York",
    "input": {"profile": "default"},
}

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.