## State & Sessions


## HookState


Per-hook persistent state tracked across events in a session (`fire_count` for `max_fires`).


Usage

``` python
HookState()
```


## PrimitiveState


Per-primitive nudge/gate state shared across all hooks in a session.


Usage

``` python
PrimitiveState()
```


`last_fired_at` (the session-global turn-throttle read by every LLM hook and blocking gate) and the `echo_lemmas`/`echo_window_end` window (a global last-write-wins "don't re-trigger on a parrot of the last nudge" filter) are deliberately session-scoped, not per hook. `echo_verbatim` is the session-wide, cross-hook FIFO of fired-warn sentences that damps a later verbatim quote of any hook's warning, independent of the lemma window. `consumed` is a per-hook-name ledger of signal-text hashes each hook has already scored, so one hook's aggregate fire cannot mute another's signals -- access it through [consumed_for()](state-sessions.md#captain_hook.PrimitiveState.consumed_for).


#### Methods

| Name | Description |
|----|----|
| [consumed_for()](#captain_hook.PrimitiveState.consumed_for) | The mutable set of signal-text hashes [hook](registration.md#captain_hook.hook) has already consumed (created on first use). |
| [match_signals()](#captain_hook.PrimitiveState.match_signals) | Score [hook](registration.md#captain_hook.hook)'s non-consumed `texts` against `sig`; consume and return the |
| [strip_fired_output()](#captain_hook.PrimitiveState.strip_fired_output) | The candidate content after removing any seeded fired-warn sentences. |
| [unechoed_candidates()](#captain_hook.PrimitiveState.unechoed_candidates) | Map *texts* through fired-output stripping, dropping pure quotes -- the containment-only |


##### consumed_for()


The mutable set of signal-text hashes [hook](registration.md#captain_hook.hook) has already consumed (created on first use).


Usage

``` python
consumed_for(hook)
```


##### match_signals()


Score [hook](registration.md#captain_hook.hook)'s non-consumed `texts` against `sig`; consume and return the


Usage

``` python
match_signals(sig, texts, hook)
```


contributing texts (window order, deduped by text) on a fire, else `None`.

Under the default `scope="text"` an entry qualifies when the weights of its own matched signals reach `threshold`, and only qualifying entries are consumed -- sub-threshold matches stay live for a later pass. Under `scope="window"` a signal counts once toward `threshold` however many entries it matches (presence-union over the window), and every matching entry is consumed on a fire. Any veto matching any entry -- consumed or not -- suppresses the fire and consumes nothing under either scope. Consumption is scoped to [hook](registration.md#captain_hook.hook)'s own ledger.


##### strip_fired_output()


The candidate content after removing any seeded fired-warn sentences.


Usage

``` python
strip_fired_output(text)
```


Returns *text* unchanged when the ledger is empty or no seeded sentence is contained (Allow-side precision -- shared words alone never touch it). When a seeded sentence is contained it is deleted from the whitespace-normalized text and the remainder returned; an empty or sub-floor remainder means a pure quote of fired output, returned as `""` so callers drop it. A genuinely new violation riding alongside a quoted warn survives as the stripped remainder and is still scored.


##### unechoed_candidates()


Map *texts* through fired-output stripping, dropping pure quotes -- the containment-only


Usage

``` python
unechoed_candidates(texts)
```


candidate filter shared by the llm pre-gate and post-verdict consumption so the two agree on which texts are eligible (a divergence would let a quote veto or absorb a real violation).


## SourceEdits


Condition matching an [Edit](prompt-contexts.md#captain_hook.Edit)/`Write` of a non-test source file in one language.


Usage

``` python
SourceEdits(lang="py", include_tests=False, paths=None, project_only=True)
```


True when the current event edits a file matching the language's globs and -- unless `include_tests` -- is not a test file, optionally narrowed to `paths`.


#### Parameters


`lang: str = ``"py"`  
Language key into \[`LANG_GLOBS`\]\[captain_hook.langs.LANG_GLOBS\] (default `"py"`); unknown keys fall back to `*.<lang>`.

`include_tests: bool = ``False`  
Include test files too (default `False` excludes them).

`paths: str | Sequence[str] | None = None`  
One or more `fnmatch` globs the file must also match (e.g. `("src/**", "lib/**")`); `None` (default) imposes no path restriction.

`project_only: bool = ``True`  
Restrict matches to files inside the repository root (default `True`); pass `False` to also match external scratch files, attachments, or logs.


#### Example

``` python
>>> hook(Event.PostToolUse, only_if=[SourceEdits(lang="ts", paths=("src/**", "lib/**"))], message="...")
```


## WorkflowState


Base for a pydantic model that bundles one session workflow across several hooks.


Usage

``` python
WorkflowState()
```


Decorate the subclass with \[[workflow_state](state-sessions.md#captain_hook.workflow_state)\]\[captain_hook.workflow_state\] to register it; the subclass then carries three event-driven helpers. [load](configuration-prompts.md#captain_hook.Prompt.load) reads the stored state (defaulting to a fresh instance), `save` writes it, and `reset` deletes it.


#### Example

``` python
>>> @workflow_state("review")
... class ReviewState(WorkflowState):
...     intent: str | None = None
```


#### Methods

| Name | Description |
|----|----|
| [mutate()](#captain_hook.WorkflowState.mutate) | Yield the stored workflow state under an exclusive lock; persist it on clean exit. |


##### mutate()


Yield the stored workflow state under an exclusive lock; persist it on clean exit.


Usage

``` python
mutate(evt)
```


Reach for this over [load](configuration-prompts.md#captain_hook.Prompt.load)/`save` when several hooks race the same workflow record within a session and a lost update would corrupt it. Non-reentrant: nesting [mutate()](state-sessions.md#captain_hook.SessionSlot.mutate) on the same slot within one process deadlocks -- the file lock is held for the whole block and a second acquire of the same path (default `timeout=-1`) blocks forever.


## workflow_state()


Usage

``` python
workflow_state(name)
```


## SessionSlot


A typed slot for reading/writing a single Pydantic model in a session directory.


Usage

``` python
SessionSlot(session_dir, model)
```


#### Methods

| Name | Description |
|----|----|
| [mutate()](#captain_hook.SessionSlot.mutate) | Yield the loaded model under an exclusive file lock; persist it on clean exit. |


##### mutate()


Yield the loaded model under an exclusive file lock; persist it on clean exit.


Usage

``` python
mutate(*, timeout=-1)
```


The lock is held for the whole `with` block, so concurrent writers -- separate `capt-hook run` processes racing one session's state, or threads within a process -- serialize rather than clobber. The body must be short: no slow work under the lock. A null slot (no session directory) yields an in-memory model and persists nothing. `timeout` is the lock-acquire budget in seconds (`-1` waits indefinitely); pass `timeout=0` for a non-blocking acquire that raises `filelock.Timeout` on contention.


## SessionStore


Class-keyed store providing typed [SessionSlot](state-sessions.md#captain_hook.SessionSlot) access via `store[ModelClass]`.


Usage

``` python
SessionStore(session_dir)
```


#### Methods

| Name | Description |
|----|----|
| [load()](#captain_hook.SessionStore.load) | Read `model` from its session slot, defaulting to a fresh `model()`. |
| [once()](#captain_hook.SessionStore.once) | Return `True` the first time `(scope, key)` is seen this session, `False` thereafter. |
| [track()](#captain_hook.SessionStore.track) | Register `model` so it appears in [tracked_models()](state-sessions.md#captain_hook.SessionStore.tracked_models) and [tracked_paths()](state-sessions.md#captain_hook.SessionStore.tracked_paths). |
| [tracked_models()](#captain_hook.SessionStore.tracked_models) | Return the registered tracked-state models as an immutable tuple. |
| [tracked_paths()](#captain_hook.SessionStore.tracked_paths) | Return `{ModelClass.__name__: Path}` for every tracked model whose slot has a path. |
| [unseen()](#captain_hook.SessionStore.unseen) | Return the first-sight `keys` under `scope`, recording the whole fresh subset in one write. |


##### load()


Read `model` from its session slot, defaulting to a fresh `model()`.


Usage

``` python
load(model)
```


##### Parameters


`model: type[M]`  
The Pydantic model class to read.


##### Returns


`M`  
The persisted instance, or a newly constructed `model()` when no

stored state exists for this session.


##### once()


Return `True` the first time `(scope, key)` is seen this session, `False` thereafter.


Usage

``` python
once(key, *, scope=None)
```


Keyed, scoped dedup for hook authors -- the single-key case of [unseen()](state-sessions.md#captain_hook.SessionStore.unseen).


##### track()


Register `model` so it appears in [tracked_models()](state-sessions.md#captain_hook.SessionStore.tracked_models) and [tracked_paths()](state-sessions.md#captain_hook.SessionStore.tracked_paths).


Usage

``` python
track(model)
```


##### tracked_models()


Return the registered tracked-state models as an immutable tuple.


Usage

``` python
tracked_models()
```


##### tracked_paths()


Return `{ModelClass.__name__: Path}` for every tracked model whose slot has a path.


Usage

``` python
tracked_paths()
```


##### unseen()


Return the first-sight `keys` under `scope`, recording the whole fresh subset in one write.


Usage

``` python
unseen(keys, *, scope=None)
```


De-duplicates within the batch (order-preserving) and marks every returned key before any downstream filtering, so a batch is never partially recorded; no write when nothing is fresh. `scope` namespaces independent call sites on the shared session store.


## session_state()


Decorator that registers a Pydantic model for collective [SessionStore](state-sessions.md#captain_hook.SessionStore) introspection.


Usage

``` python
session_state(cls)
```


#### Example

``` python
>>> @session_state
... class Snapshot(BaseModel):
...     op_id: str
```


## DurableState


Base for a model persisted across sessions, scoped by the `scope` class keyword.


Usage

``` python
DurableState()
```


Declare the scope at subclass time -- `class Foo(DurableState, scope="global")`; it defaults to `"project"` (keyed by the repo root). The subclass carries [load](configuration-prompts.md#captain_hook.Prompt.load)/`save`/ `reset` plus a locked [mutate()](state-sessions.md#captain_hook.SessionSlot.mutate) context manager. Reach for it when state must outlive a single session; for within-session sharing use \[[workflow_state](state-sessions.md#captain_hook.workflow_state)\]\[captain_hook.workflow_state\] instead.


#### Example

``` python
>>> from captain_hook import Deque
>>> class JsonShapes(DurableState, scope="global"):
...     shapes: Deque[256]
```


## DurableSlot


A [SessionSlot](state-sessions.md#captain_hook.SessionSlot) rooted in a durable directory; inherits the locked [mutate()](state-sessions.md#captain_hook.SessionSlot.mutate).


Usage

``` python
DurableSlot(session_dir, model)
```


## DurableStore


Class-keyed durable store providing typed [DurableSlot](state-sessions.md#captain_hook.DurableSlot) access via `store[Model]`.


Usage

``` python
DurableStore(directory)
```


## Deque


A bounded `deque` field type whose `maxlen` survives JSON round-trips.


Usage

``` python
Deque(maxlen)
```


Subscript with the cap: `Deque[256]` is a `deque[str]` capped at 256, and `Deque[int, 256]` sets the element type. A bare [Deque](state-sessions.md#captain_hook.Deque) field (no `= Field(...)`) defaults to an empty bounded deque, the cap is re-applied on every load, and the deque auto-evicts its oldest item on append -- so a growing collection stays bounded across sessions. Unlike `annotated_types.MaxLen`, which only rejects over-long input, this caps the collection instead of raising.


#### Parameter Attributes


`maxlen: int`  


#### Example

``` python
>>> from captain_hook import DurableState, Deque
>>> class JsonShapes(DurableState, scope="global"):
...     shapes: Deque[256]
```


## Workflow


Usage

``` python
Workflow(
    *,
    label,
    marker,
    steps,
    artifacts=list(),
    post_complete=None,
    on_start=None
)
```


#### Parameter Attributes


`label: str`  

`marker: str`  

`steps: list[`<a href="state-sessions.html#captain_hook.Step" class="gdls-link gdls-code"><code>Step</code></a>`]`  

`artifacts: list[`<a href="state-sessions.html#captain_hook.Artifact" class="gdls-link gdls-code"><code>Artifact</code></a>`[BaseModel]] = list()`    

`post_complete: Callable[[`<a href="events-results.html#captain_hook.BaseHookEvent" class="gdls-link gdls-code"><code>BaseHookEvent</code></a>`], `<a href="events-results.html#captain_hook.HookResult" class="gdls-link gdls-code"><code>HookResult</code></a>` | None] | None = None`  

`on_start: Callable[[`<a href="events-results.html#captain_hook.BaseHookEvent" class="gdls-link gdls-code"><code>BaseHookEvent</code></a>`], `<a href="events-results.html#captain_hook.HookResult" class="gdls-link gdls-code"><code>HookResult</code></a>` | None] | None = None`  


#### Methods

| Name | Description |
|----|----|
| [setup()](#captain_hook.Workflow.setup) | Run the `on_start` callback when the workflow's subagent launches. |


##### setup()


Run the `on_start` callback when the workflow's subagent launches.


Usage

``` python
setup(evt)
```


## Step


One step of a [workflow()](state-sessions.md#captain_hook.workflow) guard: `check` gates progress, `message` is shown when it fails.


Usage

``` python
Step(*, check, message, name="")
```


`message` is the full "you are here, do this next" sentence surfaced on the blocking guard. [name](files-commands.md#captain_hook.Call.name) is optional and used only for readability at the call site.


#### Parameter Attributes


`check: Callable[[Session], bool]`  

`message: str`  

`name: str = ``""`  


#### Example

``` python
>>> Step(check=text_matches(r"pytest"), message="Step 1: run the tests, then fix failures.")
```


## Artifact


Usage

``` python
Artifact(*, path, model, validate=lambda _: None)
```


#### Parameter Attributes


`path: str`  

`model: type[M]`  

`validate: Callable[[M], str | None] = lambda _: None`    


## text_matches()


Usage

``` python
text_matches(pattern)
```


## workflow()


Usage

``` python
workflow(
    *,
    label,
    marker,
    steps,
    artifacts=None,
    post_complete=None,
    on_start=None,
    only_if=(),
    skip_if=(),
    tests=None
)
```