State & workflows

Typed session state, durable state, and multi-step workflow gates.

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 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 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.

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.

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 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.

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()

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:

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() 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.

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() too.

The framework reserves two slots of its own. 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 backs the LLM primitives. You read them at most, never write them.

Persist across sessions with DurableState

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, the cross-session twin of WorkflowState. Subclass it and declare a scope at subclass time.

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 subclass carries the same load(evt), save(evt), reset(evt), and mutate(evt) as WorkflowState. Here 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 on any write that accumulates into a collection; the same short-block and no-nesting rules apply.

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.

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 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.

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 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(). The label prefixes every block message, so uppercase names like "QA-GATE" stand out in the transcript.

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() signature, including the artifacts, post_complete, on_start, only_if, and skip_if parameters, see the primitives reference.

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”:

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 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:

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 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, a Pydantic model to parse the JSON against, and an optional validate callback that returns an error string to block or None to pass.

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 to block or warn, or None to allow the stop.

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 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 and parses against its model.

When to reach for state

Read and write a slot

Bundle one workflow with @workflow_state

Serialize racing writers with mutate()

Register models with @session_state

Persist across sessions with DurableState

Where state lives

Enforce a multi-step workflow

How the pieces fit together

Register a workflow

Write a step check

Verify on-disk artifacts

Run a callback after everything passes

Debug a workflow that blocks too much

Bundle one workflow with `@workflow_state`

Register models with `@session_state`