Multi-Step Workflow

Before a subagent finishes, it must run tests, lint, and verify coverage. The agent sometimes decides it's "done" after the implementation step and tries to stop early. You want a checklist guard that blocks SubagentStop until every step has run and the agent has emitted a deliberate completion marker.

from __future__ import annotations

from captain_hook import Artifact, Step, text_matches, workflow
from pydantic import BaseModel


class TestReport(BaseModel):
    passed: int
    failed: int


workflow(
    label="VERIFY",
    marker="VERIFY COMPLETE",
    steps=[
        Step(
            name="run tests",
            check=text_matches(r"pytest.*passed"),
            stopped_at="Stop: tests not run.",
            next_step="Run the test suite with pytest.",
        ),
        Step(
            name="run linter",
            check=text_matches(r"ruff check.*passed|no issues found"),
            stopped_at="Stop: linter not run.",
            next_step="Run: ruff check .",
        ),
        Step(
            name="confirm coverage",
            check=text_matches(r"coverage:\s*\d+%"),
            stopped_at="Stop: coverage not checked.",
            next_step="Check coverage and print `coverage: NN%`.",
        ),
    ],
    artifacts=[
        Artifact(
            path=".reports/tests.json",
            model=TestReport,
            validate=lambda r: f"{r.failed} tests failed" if r.failed else None,
        ),
    ],
)

What to learn: workflow(label=..., marker=..., steps=[...], artifacts=[...]) registers a SubagentStop guard. Each Step.check is a callable that scans the transcript for evidence the step ran — text_matches(r"…") is the canonical helper. Adding an Artifact(path, model, validate) requires the agent to also produce a file that parses into your Pydantic model and passes the validator. The marker string prevents accidental passes from stale transcript text — instruct the agent to emit it deliberately.