# State & workflows

Hooks run stateless by default. Each invocation starts fresh and forgets the last. Session state gives a hook memory that survives across invocations within one Claude Code session -- count events, deduplicate warnings, accumulate data. Workflows build on the same machinery to hold a subagent to an ordered checklist before it stops.


# When to reach for state

A hook decision draws on three sources. Pick the smallest one that answers your question.

| Source | Holds | Reach for it when |
|----|----|----|
| Handler logic | The current event only | The current tool call, file, or prompt decides the verdict |
| Conditions | The transcript, recomputed every event | The answer lives in tool uses and messages already on record |
| State | Custom Pydantic models written to disk | The answer is a computed value the transcript cannot reconstruct |

Conditions read what already happened. A [RanCommand](../../reference/conditions.md#captain_hook.RanCommand) condition tells you whether the agent ran tests, because the test command sits in the transcript. State holds values you compute and write yourself, like a running count, a cached score, or a set of paths you accumulated across edits. If the transcript already carries the answer, use a condition and skip state. Reserve state for values you cannot derive from the transcript alone.


# Read and write a slot

State lives in the [SessionStore](../../reference/state-sessions.md#captain_hook.SessionStore) on the hook context, reachable as `evt.ctx.session` or the shorthand `evt.ctx.s`. Index the store by a Pydantic model class to get a typed [SessionSlot](../../reference/state-sessions.md#captain_hook.SessionSlot).

``` python
from pydantic import BaseModel, Field
from captain_hook import on, Event

class EditCounter(BaseModel):
    count: int = 0
    files: list[str] = Field(default_factory=list)

@on(Event.PostToolUse)
def count_edits(evt):
    state = evt.ctx.s[EditCounter].get() or EditCounter()
    if evt.file:
        state.count += 1
        state.files.append(str(evt.file.path))
    evt.ctx.s[EditCounter].set(state)
```

A slot gives you three operations. `.get()` returns the stored instance or `None`, and passing a default returns that instead, as in `.get(EditCounter())`. `.set(instance)` writes the model to disk atomically. `.delete()` removes the stored file.

This read-update-write cycle is the common shape. `evt.ctx.s.load(EditCounter)` shortens the read leg, returning the stored instance or a fresh `EditCounter()` when nothing is stored yet.

> **Note: Mutable defaults**
>
> Give every list, set, or dict field a `Field(default_factory=...)`. A bare `list[str] = []` shares one list across instances.


# Bundle one workflow with `@workflow_state`

When a hook spans several events, you don't want each handler hand-rolling the same read-update-write. Subclass [WorkflowState](../../reference/state-sessions.md#captain_hook.WorkflowState) and decorate it with `@workflow_state("name")`; the model inherits three methods that take the event directly. `.load(evt)` reads and defaults to a fresh instance, `.save(evt)` writes, and `.reset(evt)` deletes.

This example threads one model through three events. A prompt seeds intent, edits accumulate as they happen, a test run flips a flag, and `Stop` gates the agent if it edited code but never ran tests.

``` python
from pydantic import Field
from captain_hook import Event, RanCommand, Tool, WorkflowState, on, workflow_state

@workflow_state("review")
class ReviewState(WorkflowState):
    intent: str | None = None
    edits: list[str] = Field(default_factory=list)
    ran_tests: bool = False

@on(Event.UserPromptSubmit)
def capture_intent(evt):
    state = ReviewState.load(evt)
    state.intent = (evt.user_prompt or "").strip()[:200]
    state.save(evt)

@on(Event.PostToolUse, only_if=[Tool("Edit|Write")])
def record_edits(evt):
    if evt.file:
        state = ReviewState.load(evt)
        state.edits.append(str(evt.file.path))
        state.save(evt)

@on(Event.PostToolUse, only_if=[RanCommand("pytest")])
def record_tests(evt):
    state = ReviewState.load(evt)
    state.ran_tests = True
    state.save(evt)

@on(Event.Stop)
def require_tests(evt):
    state = ReviewState.load(evt)
    if state.edits and not state.ran_tests:
        return evt.block(f"Edited {len(state.edits)} files but never ran pytest.")
```

Three handlers, one model, one stored file. Prefer `@workflow_state` over a loose bag of models whenever the data is conceptually one workflow.


## Serialize racing writers with [mutate()](../../reference/state-sessions.md#captain_hook.SessionSlot.mutate)

[load](../../reference/configuration-prompts.md#captain_hook.Prompt.load), modify, `save` is last-writer-wins. Within one session that bites when several hooks race the same record -- a batch of parallel tool calls can all read the same pre-increment value and clobber each other's writes. `WorkflowState.mutate(evt)` is a context manager that loads, yields, and saves under an exclusive file lock, so racing writers serialize instead:

``` python
with ReviewState.mutate(evt) as state:
    state.edits.append(str(evt.file.path))
```

Keep the block short -- the lock is held until it exits -- and never nest [mutate()](../../reference/state-sessions.md#captain_hook.SessionSlot.mutate) on the same slot within one process: the inner acquisition deadlocks on the outer lock.


# Register models with `@session_state`

`@session_state` registers a plain model for collective introspection without adding the workflow methods. Use it when you want several models to surface together, for example to list their on-disk paths in a subagent's setup context.

``` python
from pydantic import BaseModel, Field
from captain_hook import session_state

@session_state
class Snapshot(BaseModel):
    op_id: str

@session_state
class CleanupScope(BaseModel):
    files: list[str] = Field(default_factory=list)
```

The store then reports every registered model:

- `evt.ctx.s.tracked_models()` returns the registered classes.
- `evt.ctx.s.tracked_paths()` returns `{class_name: path}` for each tracked model whose slot has a path.

Registration does not change slot access. `evt.ctx.s[Snapshot].get()` still works the same way. Both `@workflow_state` and `@session_state` register the model, so a workflow model shows up in [tracked_models()](../../reference/state-sessions.md#captain_hook.SessionStore.tracked_models) too.

The framework reserves two slots of its own. [HookState](../../reference/state-sessions.md#captain_hook.HookState) counts how many times a hook fired and backs `max_fires` enforcement -- counted per agent context, so a subagent spends its own budget, not the session's. [PrimitiveState](../../reference/state-sessions.md#captain_hook.PrimitiveState) backs the LLM primitives. You read them at most, never write them.


# Persist across sessions with [DurableState](../../reference/state-sessions.md#captain_hook.DurableState)

[WorkflowState](../../reference/state-sessions.md#captain_hook.WorkflowState) and `@session_state` forget everything when the session ends; their files live under one session's directory. When a hook needs to remember a fact from one session and act on it in a later one, reach for [DurableState](../../reference/state-sessions.md#captain_hook.DurableState), the cross-session twin of [WorkflowState](../../reference/state-sessions.md#captain_hook.WorkflowState). Subclass it and declare a scope at subclass time.

``` python
from captain_hook import Deque, DurableState, Event, Tool, on

class FlaggedPaths(DurableState, scope="project"):
    paths: Deque[256]

@on(Event.PostToolUse, only_if=[Tool("Edit|Write")])
def remember(evt):
    if evt.file:
        with FlaggedPaths.mutate(evt) as state:
            state.paths.append(str(evt.file.path))
```

`scope` decides where the file lives. The default `"project"` gives each repo its own file, keyed by the repo root, so a learned fact stays scoped to the codebase it came from. `"global"` shares one file across every project on the machine. Choose it when the fact is repo-independent, like which shell commands emit JSON.

A [DurableState](../../reference/state-sessions.md#captain_hook.DurableState) subclass carries the same `load(evt)`, `save(evt)`, `reset(evt)`, and `mutate(evt)` as [WorkflowState](../../reference/state-sessions.md#captain_hook.WorkflowState). Here [mutate](../../reference/state-sessions.md#captain_hook.SessionSlot.mutate) matters even more: a durable file has concurrent writers *across sessions*, since two Claude Code sessions can fire the same hook at once, and a plain load-modify-save silently loses one session's write. Reach for [mutate](../../reference/state-sessions.md#captain_hook.SessionSlot.mutate) on any write that accumulates into a collection; the same short-block and no-nesting rules apply.

> **Note: Bounded collections**
>
> A growing durable collection needs a cap, or it grows without bound. `Deque[256]` is a `deque[str]` capped at 256 that auto-evicts its oldest item on append and keeps its bound across reloads. Use `Deque[item, n]` to set the element type.


# Where state lives

Each session gets a directory keyed by its Claude Code session id. A slot serializes its model to JSON at `~/.claude/state/hooks/sessions/<session_id>/<state_key>/<agent_id>/<model_key>.json` -- the agent segment is `main` for the top-level session and the agent id for a subagent, which is how per-agent budgets like `max_fires` stay separate. The model key is the snake-case class name, so `ReviewState` becomes `review_state.json`. When a session's transcript is deleted, its state directory is cleaned up on the next run.

Durable state lives outside any session directory, so it outlives them. A `"global"` model writes to `~/.claude/state/hooks/durable/global/<model_key>.json`, and a `"project"` model writes to `~/.claude/state/hooks/durable/projects/<project-key>/<model_key>.json`, where the project key is the repo basename plus a hash of its path. A project-scoped model needs a resolvable repo root; running outside any project directory or git repo, it keeps its data in memory for that call and persists nothing.

Set `CAPTAIN_HOOK_STATE_DIR` to relocate the root, which helps in CI sandboxes that forbid writes outside the workspace. The default is `~/.claude/state`. For every tunable setting, see [configuration](../../docs/guide/packs.md#configure-hook-settings).


# Enforce a multi-step workflow

Use a workflow when a subagent must finish an ordered checklist before it stops, such as running the linter, running the tests, then writing the report. A workflow blocks the stop and tells the agent which step to do next until every check passes. For a single yes-or-no gate, reach for [gate](../../reference/primitives.md#captain_hook.gate) instead -- workflows earn their weight only when you have several ordered steps or on-disk artifacts to verify.


## How the pieces fit together

A workflow has three interlinked parts.

The **marker** is a distinctive string the agent prints when it believes the whole checklist is done, such as `"QA COMPLETE"`. The workflow searches the transcript for it. While the marker is absent, the workflow walks the steps. Once the marker appears, the workflow stops checking steps and moves on to artifacts.

> **Warning: The marker is matched exactly**
>
> The workflow searches the transcript for the marker as a literal substring, including case. `"QA COMPLETE"` and `"qa complete"` are different strings -- a near-miss never satisfies the workflow, and the stop stays blocked. Choose a distinctive uppercase marker and print it verbatim.

A **Step** is one checkpoint. It pairs a `check` predicate with a `message` that says what is missing and what to run next. The first step whose `check` returns `False` blocks the stop. An optional [name](../../reference/files-commands.md#captain_hook.Call.name) labels the step at the call site.

An **Artifact** is a file on disk that must exist and parse against a Pydantic model. Artifacts run only after the marker is found, so they verify the output of a finished checklist, not its progress.

Here is the order in which a stop attempt flows through those parts:

    SubagentStop
         |
         v
      marker in transcript?
         |
         +-- no --> first Step whose check() is False
         |              |
         |              v
         |          BLOCK: "<LABEL> INCOMPLETE: <message>"
         |
         +-- yes --> each Artifact in order
                        |
                        +-- missing / bad JSON / validate() returns a string
                        |        |
                        |        v
                        |   BLOCK: "<LABEL> INCOMPLETE: <error>"
                        |
                        +-- all valid --> post_complete callback
                                              |
                                              +-- returns HookResult --> BLOCK or WARN
                                              +-- returns None        --> ALLOW stop

A workflow registers a blocking hook on `SubagentStop` with `max_fires=1`, so once it allows the stop it never fires again that session.


## Register a workflow

Pass a `label`, a `marker`, and the ordered `steps` to [workflow()](../../reference/state-sessions.md#captain_hook.workflow). The `label` prefixes every block message, so uppercase names like `"QA-GATE"` stand out in the transcript.

``` python
from captain_hook import workflow, Step, text_matches

workflow(
    label="QA-GATE",
    marker="QA COMPLETE",
    steps=[
        Step(
            name="run linter",
            check=text_matches(r"ruff check"),
            message="Linter has not been run. Run: ruff check src/",
        ),
        Step(
            name="run tests",
            check=text_matches(r"pytest.*passed"),
            message="Tests have not been run. Run: pytest tests/ -x",
        ),
    ],
)
```

When the agent tries to stop before running the linter, it sees the first step's messages joined under the label:

    QA-GATE INCOMPLETE: Linter has not been run. Run: ruff check src/

For the full [workflow()](../../reference/state-sessions.md#captain_hook.workflow) signature, including the `artifacts`, `post_complete`, `on_start`, `only_if`, and `skip_if` parameters, see the [primitives reference](../../docs/reference/primitives.md).


## Write a step check

A step's `check` receives the session transcript and returns a `bool`. You have two ways to write one.

`text_matches(pattern)` runs a regex over the transcript's *prose* -- assistant and user message text, never tool commands or their output. `text_matches(r"ruff check")` fires when the agent narrates "running `ruff check` now", not when it silently runs the command. That makes it right for checks on what the agent *said* (a self-review, a stated conclusion) and wrong for "did the agent run X":

> **Warning: [text_matches](../../reference/state-sessions.md#captain_hook.text_matches) scans prose, not commands**
>
> Tool invocations and their output never enter the scan. For "did the agent actually run it," write a custom check over the tool history -- `t.has_command("ruff", "check")` -- or gate with the [RanCommand](../../reference/conditions.md#captain_hook.RanCommand) condition instead of a workflow step.

Write a plain function when the check needs the tool history. The function receives the transcript and queries it -- the same API as [`evt.ctx.t`](../../docs/guide/llm-hooks.md#query-the-transcript):

``` python
from cc_transcript.query import Session

from captain_hook import Step


def wrote_three_tests(t: Session) -> bool:
    return t.tool_calls.named("Edit").touching("test_*.py").count() >= 3


Step(
    name="write tests",
    check=wrote_three_tests,
    message="Fewer than three test files were written. Write at least three test files before stopping.",
)
```


## Verify on-disk artifacts

When a step produces a file, add an [Artifact](../../reference/state-sessions.md#captain_hook.Artifact) to confirm the file exists and holds valid data. Artifacts run after the marker is found, so they check the finished output, not the steps. Each artifact names a [path](../../reference/files-commands.md#captain_hook.Target.path), a Pydantic `model` to parse the JSON against, and an optional `validate` callback that returns an error string to block or `None` to pass.

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

class LintReport(BaseModel):
    errors: int
    warnings: int

workflow(
    label="QUALITY",
    marker="QUALITY CHECK DONE",
    steps=[
        Step(
            name="run linter",
            check=text_matches(r"ruff check"),
            message="Run the linter. Run: ruff check src/ --output-format json > reports/lint.json",
        ),
    ],
    artifacts=[
        Artifact(
            path="reports/lint.json",
            model=LintReport,
            validate=lambda r: "Too many errors" if r.errors > 0 else None,
        ),
    ],
)
```

A missing file, JSON that fails to parse against the model, or a non-empty `validate` return each blocks the stop with a labeled message:

    QUALITY INCOMPLETE: reports/lint.json not found.


## Run a callback after everything passes

Pass `post_complete` to run one final check once the marker, steps, and artifacts all pass. The callback receives the event and returns a [HookResult](../../reference/events-results.md#captain_hook.HookResult) to block or warn, or `None` to allow the stop.

``` python
from captain_hook import workflow, Step, text_matches, HookResult, Action

def warn_on_large_changeset(evt):
    edits = evt.ctx.t.tool_calls.named("Edit").count()
    if edits > 20:
        return HookResult(
            action=Action.warn,
            message=f"Large changeset ({edits} edits). Consider splitting the PR.",
        )
    return None

workflow(
    label="REVIEW",
    marker="REVIEW COMPLETE",
    steps=[
        Step(
            name="self-review",
            check=text_matches(r"self.review|code review"),
            message="Perform a self-review. Review your changes against the style guide.",
        ),
    ],
    post_complete=warn_on_large_changeset,
)
```

A workflow is stateless across sessions. When a step needs to count events or remember earlier edits, read and write a Pydantic model through the session store above, then have the step's `check` consult it.


## Debug a workflow that blocks too much

When a workflow blocks where you expected it to pass, start with the marker. Its text must match the transcript exactly, including case, so `"QA COMPLETE"` and `"qa complete"` are different strings. Next, confirm the step's `check` matches the real transcript text -- remembering that [text_matches](../../reference/state-sessions.md#captain_hook.text_matches) sees prose only -- and loosen the regex to absorb formatting, preferring `text_matches(r"pytest.*passed")` over `text_matches(r"pytest passed")`. Finally, confirm the artifact file sits at the exact [path](../../reference/files-commands.md#captain_hook.Target.path) and parses against its `model`.