# Multi-step workflow

> Block a subagent from stopping until it has run every required step and emitted a deliberate completion marker.

A subagent must run tests, lint, and verify coverage before it finishes. It sometimes calls itself "done" after the implementation step and tries to stop early. You want a checklist guard that holds `SubagentStop` open until every step shows evidence in the transcript *and* the agent emits an explicit completion marker.

``` python
"""Block a subagent from stopping until every workflow step has run and the completion marker is emitted."""

from __future__ import annotations

from pydantic import BaseModel

from captain_hook import Artifact, Step, text_matches, workflow


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


workflow(
    label="VERIFY",
    marker="VERIFY COMPLETE",
    steps=[
        Step(
            name="run tests",
            check=text_matches(r"pytest.*passed"),
            message="Stop: tests not run. Run the test suite with pytest.",
        ),
        Step(
            name="run linter",
            check=text_matches(r"ruff check.*passed|no issues found"),
            message="Stop: linter not run. Run: ruff check .",
        ),
        Step(
            name="confirm coverage",
            check=text_matches(r"coverage:\s*\d+%"),
            message="Stop: coverage not checked. 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,
        ),
    ],
)
```

> **Note: Verified by replay**
>
> The guard's pass/fail decision is exercised by replaying a real subagent session against the transcript and on-disk artifacts, not by the inline `tests` dict -- [workflow](../../reference/state-sessions.md#captain_hook.workflow) here registers no inline tests.


# What it does

`workflow(label=..., marker=..., steps=[...], artifacts=[...])` registers a `SubagentStop` guard. On each stop attempt it scans the replayed session transcript:

- **Marker missing** -- blocks with the first step whose `check` has no transcript evidence, e.g. no `pytest.*passed` match means `"Stop: tests not run. Run the test suite with pytest."`
- **Marker present, artifact missing or invalid** -- blocks until `.reports/tests.json` exists and parses into `TestReport`; a non-zero `failed` count blocks with `"N tests failed"`.
- **Marker present, every step matched, artifact valid** -- allows the stop.

Each `Step.check` is a callable scanning the transcript for evidence the step ran; `text_matches(r"pytest.*passed")` is the canonical helper. An `Artifact(path, model, validate)` additionally requires a file that parses into your Pydantic model and clears the validator. The `marker` string prevents accidental passes from stale transcript text, so instruct the agent to emit it deliberately.

This page composes the [workflow](../../reference/state-sessions.md#captain_hook.workflow) primitive end to end: three transcript-scanning steps plus a validated artifact in one guard.


# Run it yourself

``` bash
uvx capt-hook --hooks docs/examples test
```


# See also

- [Workflows](../../docs/guide/state.md#enforce-a-multi-step-workflow)
- [Primitives reference](../../docs/reference/primitives.md)
- [Transcript](../../docs/guide/llm-hooks.md#query-the-transcript)
