State & Sessions
HookState
Per-hook persistent state tracked across events in a session (fire_count for max_fires).
Usage
HookState()PrimitiveState
Per-primitive nudge/gate state shared across all hooks in a session.
Usage
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().
Methods
| Name | Description |
|---|---|
| consumed_for() | The mutable set of signal-text hashes hook has already consumed (created on first use). |
| match_signals() |
Score hook’s non-consumed texts against sig; consume and return the
|
| strip_fired_output() | The candidate content after removing any seeded fired-warn sentences. |
| unechoed_candidates() | Map texts through fired-output stripping, dropping pure quotes — the containment-only |
consumed_for()
The mutable set of signal-text hashes hook has already consumed (created on first use).
Usage
consumed_for(hook)match_signals()
Score hook’s non-consumed texts against sig; consume and return the
Usage
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’s own ledger.
strip_fired_output()
The candidate content after removing any seeded fired-warn sentences.
Usage
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
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/Write of a non-test source file in one language.
Usage
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
Falseexcludes them). paths: str | Sequence[str] | None = None-
One or more
fnmatchglobs 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); passFalseto also match external scratch files, attachments, or logs.
Example
>>> 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
WorkflowState()Decorate the subclass with [workflow_state][captain_hook.workflow_state] to register it; the subclass then carries three event-driven helpers. load reads the stored state (defaulting to a fresh instance), save writes it, and reset deletes it.
Example
>>> @workflow_state("review")
... class ReviewState(WorkflowState):
... intent: str | None = NoneMethods
| Name | Description |
|---|---|
| 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
mutate(evt)Reach for this over load/save when several hooks race the same workflow record within a session and a lost update would corrupt it. Non-reentrant: nesting 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
workflow_state(name)SessionSlot
A typed slot for reading/writing a single Pydantic model in a session directory.
Usage
SessionSlot(session_dir, model)Methods
| Name | Description |
|---|---|
| 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
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 access via store[ModelClass].
Usage
SessionStore(session_dir)Methods
| Name | Description |
|---|---|
| load() |
Read model from its session slot, defaulting to a fresh model().
|
| once() |
Return True the first time (scope, key) is seen this session, False thereafter.
|
| track() |
Register model so it appears in tracked_models() and tracked_paths().
|
| tracked_models() | Return the registered tracked-state models as an immutable tuple. |
| tracked_paths() |
Return {ModelClass.__name__: Path} for every tracked model whose slot has a path.
|
| 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
load(model)Parameters
model: type[M]- The Pydantic model class to read.
Returns
M-
The persisted instance, or a newly constructed
stored state exists for this session.model()when no
once()
Return True the first time (scope, key) is seen this session, False thereafter.
Usage
once(key, *, scope=None)Keyed, scoped dedup for hook authors — the single-key case of unseen().
track()
Register model so it appears in tracked_models() and tracked_paths().
Usage
track(model)tracked_models()
Return the registered tracked-state models as an immutable tuple.
Usage
tracked_models()tracked_paths()
Return {ModelClass.__name__: Path} for every tracked model whose slot has a path.
Usage
tracked_paths()unseen()
Return the first-sight keys under scope, recording the whole fresh subset in one write.
Usage
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 introspection.
Usage
session_state(cls)Example
>>> @session_state
... class Snapshot(BaseModel):
... op_id: strDurableState
Base for a model persisted across sessions, scoped by the scope class keyword.
Usage
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/save/ reset plus a locked mutate() context manager. Reach for it when state must outlive a single session; for within-session sharing use [workflow_state][captain_hook.workflow_state] instead.
Example
>>> from captain_hook import Deque
>>> class JsonShapes(DurableState, scope="global"):
... shapes: Deque[256]DurableSlot
A SessionSlot rooted in a durable directory; inherits the locked mutate().
Usage
DurableSlot(session_dir, model)DurableStore
Class-keyed durable store providing typed DurableSlot access via store[Model].
Usage
DurableStore(directory)Deque
A bounded deque field type whose maxlen survives JSON round-trips.
Usage
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 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
>>> from captain_hook import DurableState, Deque
>>> class JsonShapes(DurableState, scope="global"):
... shapes: Deque[256]Workflow
Usage
Workflow(
*,
label,
marker,
steps,
artifacts=list(),
post_complete=None,
on_start=None
)Parameter Attributes
label: strmarker: strsteps: list[Step]artifacts: list[Artifact[BaseModel]] = list()post_complete: Callable[[BaseHookEvent], HookResult | None] | None = Noneon_start: Callable[[BaseHookEvent], HookResult | None] | None = None
Methods
| Name | Description |
|---|---|
| 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
setup(evt)Step
One step of a workflow() guard: check gates progress, message is shown when it fails.
Usage
Step(*, check, message, name="")message is the full “you are here, do this next” sentence surfaced on the blocking guard. name is optional and used only for readability at the call site.
Parameter Attributes
check: Callable[[Session], bool]message: strname: str = ""
Example
>>> Step(check=text_matches(r"pytest"), message="Step 1: run the tests, then fix failures.")Artifact
Usage
Artifact(*, path, model, validate=lambda _: None)Parameter Attributes
path: strmodel: type[M]validate: Callable[[M], str | None] = lambda _: None
text_matches()
Usage
text_matches(pattern)workflow()
Usage
workflow(
*,
label,
marker,
steps,
artifacts=None,
post_complete=None,
on_start=None,
only_if=(),
skip_if=(),
tests=None
)