Claude Code Stop Hook: Recovering from Silent Tool Stalls
The first time it happened I lost forty minutes before I noticed. Claude Code had picked up a task, called an MCP tool, and then just... sat there. No error, no follow-up message, no exit — just a cursor that had forgotten what it was doing. I checked the transcript file, scrolled to the tail, and there it was: a tool_result block with an empty string payload, and nothing after it. The model had received a response so vacant it had nothing to say back, and the harness had no policy for that silence. I only caught it because I happened to glance at the terminal; another twenty minutes and I would have blamed the model for being lazy when the real culprit was a stalled tool call I never got told about.
This article walks through building a Stop hook in Python that catches those silent stalls and nudges the session back to life. By the end you will have a working hook script that reads session state from stdin, parses the transcript with a small AST-ish scanner, matches stall signatures (empty tool_result, hung MCP call, missing assistant follow-up), and re-injects a recovery prompt using hook exit code 2 — plus retry caps, a cooldown window, and a kill-switch file so a bad detector cannot flap the session into an infinite loop. We will use the standard library only, wire structured logging through the hook's stderr channel, and finish with an end-to-end smoke test that deliberately stalls a tool and watches the recovery fire. The rule underneath all of it: a hook that cannot decide when to stop trying is not a recovery mechanism, it is a new bug wearing the old bug's clothes.
If you already ship Claude Code hooks and have been bitten by the "why did the agent just give up" mystery, this is for you. You will finish with a hook you can drop into .claude/settings.json today and a mental model for writing future hooks that fail loudly instead of quietly.
Step 1: Reading a Silent Tool Stall Out of the Session Transcript
Claude Code writes every session to a JSONL transcript — one event per line, alternating assistant and user turns. When the model emits a tool_use block, the harness is supposed to answer it with a matching tool_result block on the very next user event. A silent tool stall is what happens when that pairing never closes: the model asked for a tool, nothing came back, and the loop quietly exits without a visible error.
Before we can teach a Stop hook to recover from that state, we need code that can see it. Step 1 builds a small stall_watch package that reads a transcript file, walks the content blocks, and reports every tool_use whose id was never claimed by a matching tool_result.tool_use_id. It is the sensor. The rest of the tutorial wires that sensor into the Stop hook contract, then into a recovery action.
Setup
We keep the surface area tiny — stdlib only for the runtime, pytest for tests. The scaffold from the previous commit already gave us pyproject.toml with the stall-watch package name and a dev extra pinned to pytest>=8.0. In this step we add three new files and update the package __init__.py to re-export the public API.
Project layout after this step:
codebase/
├── pyproject.toml
├── stall_watch/
│ ├── __init__.py # public re-exports
│ ├── transcript.py # the detector
│ └── simulate.py # fixture builders for tests
└── tests/
├── conftest.py # sys.path shim for src imports
└── test_transcript.py # four behavior tests
No new dependencies. If you are starting from the scaffold commit, install the dev extras once with pip install -e '.[dev]' and you are done.
Implementation
The detector lives in stall_watch/transcript.py. We model a stalled call as a frozen dataclass so downstream consumers — the Stop hook, log formatters — can pattern-match on it without worrying about mutation.
@dataclass(frozen=True)
class PendingToolCall:
tool_use_id: str
tool_name: str
line_number: int
line_number is the 1-indexed row inside the transcript where the unmatched tool_use was emitted. That column is what will let the hook print a useful jump-to-line pointer when it fires. Keeping the shape minimal now means we do not have to re-serialize it later.
Transcript reading is split into two tiny helpers so the top-level function stays flat — the codebase rule caps us at two levels of nested control flow, and a naive single-function walker blows past that immediately.
def _iter_events(transcript_path: Path) -> Iterator[tuple[int, dict]]:
with transcript_path.open("r", encoding="utf-8") as handle:
for index, raw_line in enumerate(handle, start=1):
stripped = raw_line.strip()
if not stripped:
continue
yield index, json.loads(stripped)
def _content_blocks(event: dict) -> Iterable[dict]:
payload = event.get("message") if isinstance(event.get("message"), dict) else event
content = payload.get("content")
if isinstance(content, list):
return content
return ()
_iter_events skips blank lines defensively — real transcripts sometimes end with a trailing newline, and we do not want a JSONDecodeError on the last iteration. _content_blocks handles both the wire shape Claude Code uses ({"message": {"content": [...]}}) and the older flat shape some replay tools emit, without branching inside the main loop.
The matching logic itself is a straight sweep: record every tool_use.id we see, delete it the moment we see a tool_result.tool_use_id claim it, and whatever is left at EOF is the stall set.
def find_pending_tool_calls(transcript_path: Path) -> list[PendingToolCall]:
pending: dict[str, PendingToolCall] = {}
for line_number, event in _iter_events(transcript_path):
for block in _content_blocks(event):
block_type = block.get("type")
if block_type == "tool_use":
_record_tool_use(pending, block, line_number)
elif block_type == "tool_result":
_clear_tool_result(pending, block)
return list(pending.values())
def has_silent_stall(transcript_path: Path) -> bool:
return bool(find_pending_tool_calls(transcript_path))
The record/clear helpers are one-liners that gate on the presence of the id — malformed events cannot poison the pending map. has_silent_stall is the boolean the Stop hook will call in step 2; it exists as a named function on purpose so the hook site reads as intent, not plumbing.
Tests need transcripts that look real without having to hand-craft JSONL. stall_watch/simulate.py builds three fixtures — a healthy session, a session that stalls after one tool_use, and a mixed session where an earlier call completes and a later one stalls. Each writer returns the tool ids it minted so the test can assert on identity, not just count.
def write_stalled_transcript(path: Path, tool_name: str = "Read") -> str:
tool_id = _fresh_tool_id()
events = [
_text_event("user", "Open the config file and tell me the port."),
_text_event("assistant", "Reading the config now."),
_tool_use_event(tool_name, {"file_path": "/tmp/config.yaml"}, tool_id),
]
_write_jsonl(path, events)
return tool_id
The stalled transcript deliberately ends on a tool_use event with no follow-up — that is the exact shape Claude Code leaves on disk when the harness times out mid-tool. The mixed variant proves the detector does not falsely blame the first, already-answered call when a later one goes silent.
Verification
Run the suite from inside codebase/:
python3 -m pytest
============================= test session starts ==============================
platform darwin -- Python 3.9.6, pytest-8.4.2, pluggy-1.6.0
rootdir: .../codebase
configfile: pyproject.toml
testpaths: tests
collected 4 items
tests/test_transcript.py .... [100%]
============================== 4 passed in 0.17s ===============================
Four green tests: healthy transcripts report zero pending calls, a stalled transcript flags exactly the unmatched tool_use with its line number, the mixed transcript blames only the unanswered call, and blank lines plus unknown event types do not crash the walker.
What we built
We now have a stateless function that turns a transcript path into a list of PendingToolCall records. That is the entire sensor layer — no hooks, no orchestration, no side effects. The Stop hook we bolt on next step will just call has_silent_stall(transcript_path) and act on the answer.
The detector holds two invariants we will lean on later. First, a tool_use id is only cleared by a matching tool_result.tool_use_id, so partial or out-of-order results cannot mask a real stall. Second, the walker never raises on malformed events — blank lines, unexpected type values, and missing content arrays are all treated as "not evidence of resolution," which is the safe default when the transcript itself is the thing we suspect is broken.
Because the fixtures live next to the detector, we can re-use them from the hook tests in step 2 without a second stub. That keeps the test data path honest: the same builder that verifies the detector also drives the hook, so if the transcript shape changes upstream we get one failure to fix, not two.
What this unlocks: step 2 wires this detector into a Claude Code Stop hook, so the moment a session exits with a pending tool_use, the hook can log the offending line and return an exit code the harness treats as "restart me — I have unfinished business."
Repository
The state of the code after this step: 589e21e
Step 2: Wiring the Stall Detector into a Claude Code Stop Hook
Step 1 gave us a pure function — hand it a transcript path, get back a list of PendingToolCall records. That is the sensor. What it is not yet is a hook: Claude Code will not open a Python file and call has_silent_stall for us. The harness fires a subprocess on the Stop event, hands it a JSON payload on stdin, and reads two things back — an exit code and whatever the subprocess wrote to stderr.
Step 2 builds the thinnest possible adapter around that contract. We add a hook.py module that reads the payload, refuses to re-enter when the harness has already flagged the session as in-recovery, forwards the transcript path to the detector, and exits with code 2 plus a human-readable stderr line when a stall is present. No recovery action yet — that lands in step 3. The job here is to make the plumbing so boring that later steps can trust it.
Setup
Everything stays inside the stall_watch package. We add one new module, one new test file, and update the package __init__.py to re-export the hook's public surface so callers can from stall_watch import run.
Project layout after this step:
codebase/
├── pyproject.toml
├── stall_watch/
│ ├── __init__.py # re-exports StopHookInput, parse_stop_hook_input, run
│ ├── transcript.py # (unchanged from step 1)
│ ├── simulate.py # (unchanged from step 1)
│ └── hook.py # NEW — the Stop hook adapter
└── tests/
├── conftest.py
├── test_transcript.py
└── test_hook.py # NEW — five behavior tests
No new dependencies. Python 3.9+ standard library only — json, sys, dataclasses, pathlib, typing.IO. The tests still run under the dev extra pinned in step 1, so python3 -m pytest continues to be the single entry point.
Implementation
We model the hook input as a frozen dataclass for the same reason we did in step 1: downstream code should pattern-match on it, not mutate it. The five fields mirror exactly what Claude Code documents for the Stop event payload.
@dataclass(frozen=True)
class StopHookInput:
session_id: str
transcript_path: Path
hook_event_name: str
stop_hook_active: bool
cwd: Path
The stop_hook_active field is the one that will save us from a lot of pain. Claude Code sets it to true when the harness is re-entering the Stop event because a previous Stop hook returned non-zero. If we ignored the flag and returned 2 again, we would build our own infinite loop — the exact failure mode this whole tutorial is meant to prevent.
Parsing is intentionally tolerant on optional fields and strict on the one field we truly need — the transcript path. Missing session_id or cwd is survivable; a missing transcript path is not.
def parse_stop_hook_input(raw: str) -> StopHookInput:
payload = json.loads(raw)
return StopHookInput(
session_id=str(payload.get("session_id", "")),
transcript_path=Path(payload["transcript_path"]),
hook_event_name=str(payload.get("hook_event_name", "Stop")),
stop_hook_active=bool(payload.get("stop_hook_active", False)),
cwd=Path(payload.get("cwd", ".")),
)
payload["transcript_path"] on a missing key raises KeyError on purpose. If the harness ever sends us a payload without one, we would rather crash loudly than silently rubber-stamp the session as healthy — a false negative is the worst outcome for a hook whose whole job is to catch silence.
The scanning helper is one line of real logic wrapped around an existence check. Claude Code sometimes fires Stop before it has flushed the transcript to disk on very short sessions, so we treat "file does not exist" as "no evidence of a stall" rather than an error.
def _scan_transcript(hook_input: StopHookInput) -> list[PendingToolCall]:
if not hook_input.transcript_path.exists():
return []
return find_pending_tool_calls(hook_input.transcript_path)
The reporter is separated from run so the control-flow function stays flat — the codebase rule caps nested if at two levels, and inlining formatting logic would push us right up against it. We print exactly two stderr lines: a count summary and a pointer at the first offender. Claude Code surfaces stderr from a non-zero Stop hook back into the session, so those two lines are what a human (or the next agent turn) will actually read.
def _report_stall(
hook_input: StopHookInput,
pending: list[PendingToolCall],
stderr: IO[str],
) -> None:
first = pending[0]
stderr.write(
f"stall_watch: {len(pending)} pending tool_use "
f"in {hook_input.transcript_path}\n"
)
stderr.write(
f"stall_watch: first stall = {first.tool_name} "
f"(id={first.tool_use_id}) at line {first.line_number}\n"
)
run is the whole hook contract in eight lines. It takes stdin and stderr as parameters — not as globals — so the test suite can drive it with io.StringIO instead of shelling out. The re-entry short-circuit sits at the top; every other branch is downstream of "we are allowed to act."
def run(stdin: IO[str], stderr: IO[str]) -> int:
hook_input = parse_stop_hook_input(stdin.read())
if hook_input.stop_hook_active:
return 0
pending = _scan_transcript(hook_input)
if not pending:
return 0
_report_stall(hook_input, pending, stderr)
return 2
Finally, a shell-level main wires the real sys.stdin and sys.stderr in, so Claude Code can run the module directly with python -m stall_watch.hook.
def main() -> int:
return run(sys.stdin, sys.stderr)
if __name__ == "__main__":
raise SystemExit(main())
The tests cover every branch of run plus the parser. We re-use write_healthy_transcript and write_stalled_transcript from step 1's simulate.py — the same builder that verifies the detector now drives the hook, so any upstream shape change breaks one place, not two.
def test_run_returns_two_and_reports_when_stalled(tmp_path: Path) -> None:
transcript = tmp_path / "stalled.jsonl"
stalled_id = write_stalled_transcript(transcript, tool_name="Read")
stdin = io.StringIO(_hook_stdin_payload(transcript))
stderr = io.StringIO()
exit_code = run(stdin, stderr)
assert exit_code == 2
message = stderr.getvalue()
assert "Read" in message
assert stalled_id in message
assert "line 3" in message
assert "1 pending tool_use" in message
The four sibling tests pin the other invariants: parser field extraction, healthy transcripts exit 0 with silent stderr, the stop_hook_active short-circuit refuses to escalate, and a missing transcript file is treated as no-stall instead of crashing the hook.
Verification
Run the full suite from inside codebase/ — the step 1 tests still run alongside the new ones:
python3 -m pytest
============================= test session starts ==============================
platform darwin -- Python 3.9.6, pytest-8.4.2, pluggy-1.6.0
rootdir: .../codebase
configfile: pyproject.toml
testpaths: tests
collected 9 items
tests/test_hook.py ..... [ 55%]
tests/test_transcript.py .... [100%]
============================== 9 passed in 0.15s ===============================
Nine green tests: the four transcript tests from step 1 still pass unchanged, and the five new hook tests cover the parser, the healthy path, the stalled path with the exact stderr contract, the re-entry short-circuit, and the missing-transcript tolerance.
What we built
We now have a script that satisfies the Claude Code Stop-hook contract end to end. It reads the documented payload from stdin, does one thing — ask the step 1 detector whether the transcript still has an open tool_use — and signals the answer through the exit code the harness already knows how to interpret.
Two invariants are locked in. First, the stop_hook_active flag is honored at the top of run, so no matter how noisy a stall gets, this hook cannot cause an infinite Stop-event bounce on its own. Second, the stderr report is deterministic: same session, same transcript, same two lines. That determinism is what will let step 4 add a cooldown window keyed on session_id without also needing to fingerprint output text.
The hook still has no recovery action — exit code 2 is currently just a loud shout. That is deliberate. Isolating the "detect + signal" surface here keeps the recovery dispatcher in step 3 orthogonal: it will subscribe to the same signal without having to re-implement any of this plumbing.
What this unlocks: step 3 layers a nudge-prompt builder on top of run, so instead of only writing to stderr, the hook can hand Claude Code a concrete instruction on how to unstick itself before returning 2.
Repository
The state of the code after this step: e25e4c0
Step 3: Classifying Empty, Hung, and Orphaned Stall Signatures
Step 2 left us with a hook that could catch exactly one shape of failure — a tool_use block that never received a tool_result. That covered the loudest case, but silent stalls come in more flavors than that. In real transcripts we also see tool_result blocks that arrived empty, MCP calls that hung inside the server without any error surfacing, and pairings that closed correctly but never got a follow-up assistant turn to interpret the output.
Step 3 keeps the hook shell from step 2 intact and pushes the intelligence into the detector. We introduce a StallSignature record with an explicit kind field, extend detect_stalls to emit up to four kinds, and let the reporter tally them so the stderr message tells you why the session went silent instead of just that it did. The Stop-hook contract does not change — same stdin payload, same exit code 2 — only the vocabulary the hook speaks gets richer.
Setup
No new packages are pulled in — the detector still runs on the standard library plus pytest from the dev extra. We add one new module surface (constants + StallSignature) inside stall_watch/transcript.py, three new fixture builders in stall_watch/simulate.py, one new label map inside stall_watch/hook.py, and a fresh test file dedicated to the classifier.
Project layout after this step:
codebase/
├── pyproject.toml
├── stall_watch/
│ ├── __init__.py # re-exports new kinds + StallSignature + detect_stalls
│ ├── transcript.py # extended: StallSignature, detect_stalls, KIND_* constants
│ ├── simulate.py # + write_empty_result_transcript,
│ │ # write_hung_mcp_transcript,
│ │ # write_missing_followup_transcript
│ └── hook.py # + KIND_LABEL map, per-kind stderr line
└── tests/
├── conftest.py
├── test_transcript.py # (unchanged)
├── test_hook.py # + 2 new tests for MCP + missing-follow-up labels
└── test_stall_signatures.py # NEW — six behavior tests for detect_stalls
Everything still runs under python3 -m pytest from codebase/. The step 1 and step 2 tests are treated as regression coverage — they must keep passing without modification.
Implementation
The heart of the change is a new record type. PendingToolCall from step 1 stays where it is (the older find_pending_tool_calls API still returns it), but the richer classifier speaks in StallSignature values. The extra kind and detail fields give the hook something to print without re-deriving the reason.
@dataclass(frozen=True)
class StallSignature:
kind: str
tool_use_id: str
tool_name: str
line_number: int
detail: str
Kinds are exposed as plain string constants at module top. That keeps them easy to compare in tests, easy to serialize into log lines, and easy to look up in the hook's label map — no enum boilerplate for what is essentially a tag.
KIND_PENDING_TOOL_USE = "pending_tool_use"
KIND_EMPTY_TOOL_RESULT = "empty_tool_result"
KIND_HUNG_MCP_CALL = "hung_mcp_call"
KIND_MISSING_FOLLOWUP = "missing_followup"
detect_stalls no longer walks the events inline; it first materializes them once, then delegates to _scan_tool_events which returns two maps — tool_id → (line, name) for uses and tool_id → (line, content) for results. Splitting the sweep from the classification keeps each function under the two-level nesting cap this project enforces.
def detect_stalls(transcript_path: Path) -> list[StallSignature]:
events = list(_iter_events(transcript_path))
tool_uses, tool_results = _scan_tool_events(events)
signatures: list[StallSignature] = []
for tool_id, (line_number, tool_name) in tool_uses.items():
if tool_id not in tool_results:
signatures.append(
_signature_for_unmatched(tool_id, line_number, tool_name)
)
continue
result_line, result_content = tool_results[tool_id]
if _is_empty_result_content(result_content):
signatures.append(
_signature_for_empty_result(tool_id, tool_name, result_line)
)
trailing = _missing_followup_signature(events, tool_uses)
if trailing is not None:
signatures.append(trailing)
signatures.sort(key=lambda sig: (sig.line_number, sig.kind))
return signatures
Unmatched tool_use blocks route through _classify_unmatched, which is the whole reason MCP calls get their own bucket. A stalled MCP call almost always means the remote server hung — the recovery playbook (restart the server, or fall back to a native tool) is completely different from a stall in a local built-in tool, so we split the label at the source rather than fixing it up at the report site.
def _classify_unmatched(tool_name: str) -> str:
if tool_name.startswith("mcp__"):
return KIND_HUNG_MCP_CALL
return KIND_PENDING_TOOL_USE
Empty results are subtle because "empty" is polymorphic. A content field can be None, an empty string, a whitespace string, or a list of blocks where every block's text and nested content are blank. _is_empty_result_content collapses all four into a single boolean without ever raising on a shape it does not recognize — anything unknown is treated as non-empty, so we prefer a missed alert over a false positive.
def _is_empty_result_content(content: Any) -> bool:
if content is None:
return True
if isinstance(content, str):
return content.strip() == ""
if isinstance(content, list):
return not any(_result_block_has_text(block) for block in content)
return False
The missing-follow-up detector looks at the last event only. If the transcript ends on a user event whose blocks are all tool_result, the model never got a chance to interpret the tool output before the harness cut the session. That is a distinct failure mode from the other three — the tool succeeded, but the loop still went silent — so it earns its own kind.
def _missing_followup_signature(
events: list[tuple[int, dict]], tool_uses: dict[str, tuple[int, str]]
) -> StallSignature | None:
if not events:
return None
last_line, last_event = events[-1]
is_trailing_result, tool_id = _event_is_only_tool_result(last_event)
if not is_trailing_result:
return None
tool_name = tool_uses.get(tool_id, (0, ""))[1] if tool_id else ""
detail = (
f"tool_result for {tool_name or 'tool'} at line {last_line} "
"had no assistant follow-up"
)
return StallSignature(
kind=KIND_MISSING_FOLLOWUP,
tool_use_id=tool_id,
tool_name=tool_name,
line_number=last_line,
detail=detail,
)
Over in hook.py, _report_stall now groups signatures by kind and prints one summary line per bucket, then a pointer at the first offender annotated with its kind label. Callers reading stderr get an at-a-glance breakdown — "1 empty tool_result, 1 hung mcp call" — followed by the exact line to jump to.
def _report_stall(
hook_input: StopHookInput,
signatures: list[StallSignature],
stderr: IO[str],
) -> None:
counts = Counter(sig.kind for sig in signatures)
for kind, count in counts.items():
stderr.write(
f"stall_watch: {count} {_label(kind)} "
f"in {hook_input.transcript_path}\n"
)
first = signatures[0]
stderr.write(
f"stall_watch: first stall = {first.tool_name} "
f"(id={first.tool_use_id}) at line {first.line_number} "
f"[{_label(first.kind)}]\n"
)
The simulate.py builders grow three siblings alongside write_stalled_transcript: one that terminates a Bash call with an empty tool_result, one that opens an unfinished mcp__example__slow_call, and one that closes the pair but stops the transcript there — no assistant follow-up. Every builder returns the tool ids it minted so tests assert on identity, not just count.
Verification
Run the full suite from inside codebase/ — the four transcript tests from step 1, the hook file (now seven tests: five from step 2 plus the two new label assertions), and the six new classifier tests should all light up:
python3 -m pytest
============================= test session starts ==============================
platform darwin -- Python 3.9.6, pytest-8.4.2, pluggy-1.6.0
rootdir: .../codebase
configfile: pyproject.toml
testpaths: tests
collected 17 items
tests/test_hook.py ....... [ 41%]
tests/test_stall_signatures.py ...... [ 76%]
tests/test_transcript.py .... [100%]
============================== 17 passed in 0.31s ==============================
Seventeen green tests split across three files. test_stall_signatures.py pins each new kind independently (pending tool_use, empty tool_result, hung MCP call, missing follow-up) plus a multi-kind case that appends an unmatched MCP call after an empty result. test_hook.py adds two label assertions — the stderr must actually contain the human phrase hung mcp call for MCP stalls and missing follow-up for orphaned results — so the classifier is visible from the outermost contract, not just internally.
What we built
The detector now speaks four distinct dialects instead of one. Whether the stall came from a tool that never returned, a tool that returned nothing useful, a remote MCP server that hung, or a completed call that never got interpreted, the signature carries a kind tag the rest of the system can branch on.
Two invariants harden the vocabulary. First, classification happens at emission time — the kind field is set once inside detect_stalls and never re-computed downstream, so the hook, the tests, and any future recovery dispatcher all agree on what a signature is. Second, the empty-result predicate is deliberately conservative: unknown shapes count as non-empty, so a novel content schema in a future Claude Code release cannot silently mass-flag healthy sessions.
The hook contract from step 2 is unchanged — same stdin payload, same exit code 2, same re-entry short-circuit. What did change is what it says on the way out. Instead of one generic "pending tool_use" line, the stderr now describes exactly which shape of silence the session died in.
What this unlocks: step 4 can key the recovery playbook on kind. A pending-tool-use stall wants a "retry the tool" nudge, an empty-result stall wants a "re-run with verbose flag" nudge, a hung MCP call wants a "restart the server or fall back" nudge, and a missing follow-up wants "please summarize the last tool_result." All four dispatch off the same signature stream this step produces.
Repository
The state of the code after this step: 88ad0aa
Step 4: Dispatching a Per-Kind Recovery Nudge Through Hook Exit Code 2
Step 3 taught the detector to speak four dialects — pending_tool_use, empty_tool_result, hung_mcp_call, missing_followup — but the hook still only did one thing with that vocabulary: shout it into stderr. A human reading the log could act on it, yet the agent that just tried to stop got no guidance whatsoever. That is exactly the moment we can close the loop, because Claude Code re-surfaces a non-zero Stop hook's stderr back into the model's next turn.
Step 4 introduces a recovery dispatcher that turns each StallSignature.kind into a targeted nudge, assembles them into a single re-injectable prompt with a header and a footer, and wires it into _report_stall so the prompt goes out on the same exit-code-2 event we already spend. No new failure modes are added, no new hook contract is negotiated — we just make the shout say something the model can act on.
Setup
The dispatcher is small enough to live in its own module without dragging in dependencies. We add stall_watch/recovery.py, extend stall_watch/hook.py to call it, add a fresh tests/test_recovery.py suite, and re-export the new public names from stall_watch/__init__.py. Nothing new lands in pyproject.toml — recovery is pure string composition over the existing dataclass.
Project layout after this step:
codebase/
├── pyproject.toml
├── stall_watch/
│ ├── __init__.py # + RECOVERY_HEADER/FOOTER,
│ │ # build_recovery_prompt, nudge_for_signature
│ ├── transcript.py # (unchanged from step 3)
│ ├── simulate.py # (unchanged from step 3)
│ ├── hook.py # + build_recovery_prompt call inside _report_stall
│ └── recovery.py # NEW — kind → nudge template + prompt assembler
└── tests/
├── conftest.py
├── test_transcript.py # (unchanged)
├── test_stall_signatures.py # (unchanged)
├── test_hook.py # (unchanged)
└── test_recovery.py # NEW — 11 behavior tests
Everything continues to run under python3 -m pytest from codebase/. All 17 tests from steps 1–3 stay green as regression coverage; the 11 new tests exercise both the string-composition layer and the hook wiring.
Implementation
The recovery module opens with two constants that bookend every prompt. The header primes the model to interpret what follows as a mandatory instruction rather than a suggestion, and the footer forces a concrete next action instead of another Stop attempt. Keeping them as module-level strings means tests can assert on identity without fuzzy-matching, and future re-tuning happens in exactly one place.
RECOVERY_HEADER = (
"stall_watch recovery: a silent tool stall was detected before you "
"tried to stop. Do NOT end the turn yet — act on the guidance below."
)
RECOVERY_FOOTER = (
"Take the smallest concrete action that unblocks the task, then produce "
"the assistant follow-up the user was waiting for. Only stop after that "
"follow-up is on the transcript."
)
Per-kind nudges live in a KIND_NUDGE dict keyed on the same string constants the detector emits. Each template uses named {tool_name}, {tool_use_id}, {line_number} slots so signatures fill in mechanically — no branching per kind at format time. This is the point where the classification work from step 3 actually pays out: the four kinds each get a distinct recovery playbook, and adding a fifth is a single dict entry rather than an if/elif chain.
KIND_NUDGE = {
KIND_PENDING_TOOL_USE: (
"Retry the {tool_name} call (id {tool_use_id}) or fall back to a "
"safer alternative — it started at line {line_number} and never "
"produced a tool_result."
),
KIND_EMPTY_TOOL_RESULT: (
"The {tool_name} call at line {line_number} returned an empty "
"tool_result. Re-run it with verbose output (or inspect stderr) "
"and report what actually came back."
),
KIND_HUNG_MCP_CALL: (
"The MCP tool {tool_name} (id {tool_use_id}) hung at line "
"{line_number}. Restart the MCP server or fall back to a native "
"tool before responding."
),
KIND_MISSING_FOLLOWUP: (
"The tool_result for {tool_name} at line {line_number} was never "
"interpreted. Summarize what the tool returned and continue the "
"task the user asked for."
),
}
DEFAULT_NUDGE = (
"A silent stall was detected on {tool_name} at line {line_number}. "
"Investigate and produce a follow-up before stopping."
)
nudge_for_signature is deliberately small and forgiving. Unknown kinds fall through to DEFAULT_NUDGE instead of raising — a future detector kind that ships before the recovery module is updated should still produce a usable prompt, not crash the hook. Empty tool_name and empty tool_use_id also degrade to readable placeholders so the prompt reads like English even when the transcript is missing metadata.
def nudge_for_signature(signature: StallSignature) -> str:
template = KIND_NUDGE.get(signature.kind, DEFAULT_NUDGE)
return template.format(
tool_name=signature.tool_name or "the pending tool",
tool_use_id=signature.tool_use_id or "<unknown>",
line_number=signature.line_number,
)
build_recovery_prompt composes the final block: header, one bullet per signature tagged with its kind, then footer. Returning the empty string when the signature list is empty is a deliberate no-op — it lets the caller inline the write without an extra if signatures branch on the hook side. The [kind] prefix on every bullet is the signal the model uses to distinguish "restart the MCP server" from "retry the tool" when a session stalls on more than one kind at once.
def build_recovery_prompt(signatures: list[StallSignature]) -> str:
if not signatures:
return ""
lines = [RECOVERY_HEADER]
for signature in signatures:
lines.append(f"- [{signature.kind}] {nudge_for_signature(signature)}")
lines.append(RECOVERY_FOOTER)
return "\n".join(lines) + "\n"
Over in hook.py, _report_stall gains a single new line — the recovery prompt is appended to stderr after the two summary lines from step 3. Order matters: the tally + first-offender pointer stay on top so an operator scanning logs sees the fingerprint immediately, and the recovery block below is what the Claude Code harness will re-inject on the next turn. Both audiences read the same stderr — one from top to bottom, the other as a single blob — so we keep the human-facing summary short and the model-facing prompt below it.
def _report_stall(
hook_input: StopHookInput,
signatures: list[StallSignature],
stderr: IO[str],
) -> None:
counts = Counter(sig.kind for sig in signatures)
for kind, count in counts.items():
stderr.write(
f"stall_watch: {count} {_label(kind)} "
f"in {hook_input.transcript_path}\n"
)
first = signatures[0]
stderr.write(
f"stall_watch: first stall = {first.tool_name} "
f"(id={first.tool_use_id}) at line {first.line_number} "
f"[{_label(first.kind)}]\n"
)
stderr.write(build_recovery_prompt(signatures))
The run function itself does not change. The stop_hook_active short-circuit from step 2 still gates the whole path, so re-injecting the recovery prompt cannot loop — on the next Stop event, either the model has produced a follow-up (transcript is now healthy, we return 0) or the harness is in re-entry mode with stop_hook_active=True and we bail out without re-shouting. That is the anti-infinite-loop invariant the whole tutorial has been building toward.
The test file exercises both layers. Five tests pin the pure string dispatcher: one per known kind confirms the templates format correctly, and one drives an unknown kind to prove the DEFAULT_NUDGE fallback fires. Two tests cover build_recovery_prompt — the empty-list no-op and a two-signature case that asserts header, footer, and one - [<kind>] … bullet per signature. The remaining four tests drive run end-to-end for each stall kind, asserting the exit code stays at 2 and that the exact recovery language for that kind appears in stderr.
Verification
Run the full suite from inside codebase/ — all four test files light up:
python3 -m pytest
============================= test session starts ==============================
platform darwin -- Python 3.9.6, pytest-8.4.2, pluggy-1.6.0
rootdir: .../codebase
configfile: pyproject.toml
testpaths: tests
collected 28 items
tests/test_hook.py ....... [ 25%]
tests/test_recovery.py ........... [ 64%]
tests/test_stall_signatures.py ...... [ 85%]
tests/test_transcript.py .... [100%]
============================== 28 passed in 0.92s ==============================
Twenty-eight green tests — the eleven new test_recovery.py cases stack cleanly on top of the seventeen carried over from steps 1–3. Because the hook wiring changes only inside _report_stall, the step 2 and step 3 hook tests keep passing without modification: they never asserted on the exact stderr length, only on the presence of the summary lines the recovery prompt now sits below.
What we built
The Stop hook now closes the loop. When a silent stall is detected, the hook returns exit code 2 as before, but its stderr carries a targeted, kind-specific instruction the Claude Code harness re-injects into the model on the next turn. A pending tool_use is asked to retry, an empty tool_result is asked to re-run with verbose output, a hung MCP call is asked to restart or fall back, and a missing follow-up is asked to summarize and continue.
Two invariants harden the recovery path. First, the dispatcher is total — every signature produces a nudge, including signatures with unknown kind values, because KIND_NUDGE.get falls back to DEFAULT_NUDGE. That means a detector upgrade never silently regresses the recovery experience. Second, the anti-loop guard from step 2 still holds: the recovery prompt only ships when stop_hook_active is False, so a single stall causes at most one nudge, not a runaway feedback cycle.
The user-facing shape is unchanged from step 3 — same exit code, same first two stderr lines — so any log dashboards or CI parsers that grew around the earlier steps keep working. What changed is that the third block of stderr is now actionable for the model itself, not just for a human sitting behind the terminal.
What this unlocks: everything downstream of "the model got a targeted instruction" — cooldown windows keyed on session_id, per-project overrides of the nudge templates, telemetry that counts which kind of recovery gets triggered most often. All of it hangs off build_recovery_prompt, so future work can wrap it, rate-limit it, or A/B test it without touching the detector or the hook contract.
Repository
The state of the code after this step: d536f15
Step 5: Fencing Recovery Behind Retry Caps, Cooldown Windows, and a Kill Switch
Step 4 finished the recovery loop: the Stop hook now shouts a kind-specific nudge into stderr, exits 2, and Claude Code re-injects that nudge on the next turn. The stop_hook_active short-circuit from step 2 protects against the immediate re-entry, but nothing yet protects against a tool that never recovers — an MCP server that hangs on every call, a model that keeps proposing the same broken sequence, an operator who wants to silence the hook without deleting it. Left unchecked, three of those in a row still ends in a nudge-storm that burns tokens and hides the underlying bug.
Step 5 wraps the dispatcher in three independent safety valves. Each stall gets a per-tool retry cap so the same tool_use_id cannot keep asking for help forever, the session gets a cooldown window so back-to-back Stop events do not stack multiple nudges, and the operator gets a kill switch — either an environment variable or a marker file — that disables recovery entirely without touching the hook script.
Setup
The guardrails land in a new module, stall_watch/guardrails.py, and the hook grows a thin orchestration seam that calls it before shipping any output. Per-session state persists as JSON on disk, so retry counters and cooldown timestamps survive across the many short-lived hook invocations a single conversation produces. All new public names are re-exported from stall_watch/__init__.py, and a dedicated tests/test_guardrails.py suite covers config parsing, state I/O, cap enforcement, cooldown timing, kill-switch precedence, and the wired-up hook behavior.
Project layout after this step:
codebase/
├── pyproject.toml
├── stall_watch/
│ ├── __init__.py # + GuardrailConfig, SessionState, ...
│ ├── transcript.py # (unchanged from step 3)
│ ├── simulate.py # (unchanged from step 3)
│ ├── recovery.py # (unchanged from step 4)
│ ├── hook.py # + _dispatch orchestrates guardrails
│ └── guardrails.py # NEW — caps, cooldown, kill switch, state
└── tests/
├── conftest.py
├── test_transcript.py # (unchanged)
├── test_stall_signatures.py # (unchanged)
├── test_hook.py # (unchanged)
├── test_recovery.py # (unchanged)
└── test_guardrails.py # NEW — 21 behavior tests
No new dependencies are needed — everything runs on stdlib json, os, pathlib, and time. State files live under <cwd>/.claude/stall_watch/<session_id>.json by default, next to the harness data the operator already accepts as project-local.
Implementation
GuardrailConfig is an immutable dataclass that holds every knob the hook needs at runtime: the state directory, the retry cap, the cooldown window in seconds, the name of the kill-switch env var, and an optional marker file path. Defaults live at module scope so the same constants can be asserted from tests and read from the ## Repository section without duplication. Isolating the config surface here keeps the hook code short and lets a future release ship a config file loader without changing the call sites.
DEFAULT_MAX_RETRIES = 3
DEFAULT_COOLDOWN_SECONDS = 0.0
DEFAULT_KILL_SWITCH_ENV = "STALL_WATCH_DISABLED"
DEFAULT_STATE_SUBDIR = (".claude", "stall_watch")
@dataclass(frozen=True)
class GuardrailConfig:
state_dir: Path
max_retries: int = DEFAULT_MAX_RETRIES
cooldown_seconds: float = DEFAULT_COOLDOWN_SECONDS
kill_switch_env: str = DEFAULT_KILL_SWITCH_ENV
kill_switch_file: Path | None = None
GuardrailConfig.from_env is the one place that reads environment variables. Every knob has a STALL_WATCH_* override, so an operator can lift the cap for a single session (STALL_WATCH_MAX_RETRIES=10), widen the cooldown for a noisy MCP server (STALL_WATCH_COOLDOWN_SECONDS=120), or point the kill switch at a custom sentinel file (STALL_WATCH_KILL_SWITCH_FILE=/tmp/stall_off). Reading os.environ behind a Mapping parameter makes the whole config trivially testable — every test passes a plain dict and no test ever mutates the process environment.
@classmethod
def from_env(
cls,
cwd: Path,
environ: Mapping[str, str] | None = None,
) -> "GuardrailConfig":
env = environ if environ is not None else os.environ
state_dir = Path(
env.get("STALL_WATCH_STATE_DIR", str(_default_state_dir(cwd)))
)
return cls(
state_dir=state_dir,
max_retries=_read_int(env, "STALL_WATCH_MAX_RETRIES", DEFAULT_MAX_RETRIES),
cooldown_seconds=_read_float(
env, "STALL_WATCH_COOLDOWN_SECONDS", DEFAULT_COOLDOWN_SECONDS
),
kill_switch_env=env.get(
"STALL_WATCH_KILL_SWITCH_ENV", DEFAULT_KILL_SWITCH_ENV
),
kill_switch_file=_optional_path(env.get("STALL_WATCH_KILL_SWITCH_FILE")),
)
SessionState holds the two things that must persist across Stop events: a retries dict keyed by tool_use_id and the wall-clock timestamp of the last recovery. Storing counters per tool_use_id — not per kind, not per session — is the key design choice: two different broken tools in the same session each get their own budget, and a healthy retry that spawns a fresh tool_use_id starts from zero. load_state swallows both missing files and corrupt JSON by returning an empty state, because a broken state file must never wedge the whole hook.
@dataclass
class SessionState:
retries: dict[str, int] = field(default_factory=dict)
last_recovery_at: float = 0.0
def load_state(config: GuardrailConfig, session_id: str) -> SessionState:
path = state_path(config, session_id)
if not path.exists():
return SessionState()
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError, ValueError):
return SessionState()
if not isinstance(data, dict):
return SessionState()
return SessionState.from_dict(data)
Session IDs come straight from the Claude Code payload, so _safe_session_id strips anything that is not alphanumeric or ._- before touching the filesystem. That closes the "malicious session id writes to /etc" path — even a value like abc/../etc collapses to abc_.._etc.json inside the configured state_dir.
The three guard predicates each stay tiny and side-effect-free. is_kill_switch_active checks the env var first, then the marker file — the env wins so an operator can override a stale file without deleting it. is_in_cooldown returns False whenever cooldown_seconds <= 0, which keeps the default behavior identical to step 4 and only turns the window on when an operator opts in. partition_signatures walks the incoming list once and splits it into allowed (below the cap) and capped (at or above the cap) buckets.
def partition_signatures(
signatures: Iterable[StallSignature],
state: SessionState,
config: GuardrailConfig,
) -> GuardrailDecision:
allowed: list[StallSignature] = []
capped: list[StallSignature] = []
for signature in signatures:
key = signature.tool_use_id or ""
count = state.retries.get(key, 0)
if count >= config.max_retries:
capped.append(signature)
continue
allowed.append(signature)
return GuardrailDecision(allowed=allowed, capped=capped)
Inside hook.py, the change is one new function — _dispatch — that runs the guardrails in the order the priorities dictate: kill switch first (an operator's manual off wins over everything), then cooldown (session-level throttle beats per-tool budget), then the retry cap partition. Only signatures that clear all three gates trigger _report_stall, and record_recovery writes the new counters and timestamp back to disk before returning exit code 2. Every suppressed path returns 0 with a targeted stderr line, so an operator scanning logs can always tell which gate fired.
def _dispatch(
hook_input: StopHookInput,
signatures: list[StallSignature],
config: GuardrailConfig,
environ: Mapping[str, str] | None,
now: float,
stderr: IO[str],
) -> int:
if is_kill_switch_active(config, environ):
_report_kill_switch(stderr, config)
return 0
state = load_state(config, hook_input.session_id)
if is_in_cooldown(state, config, now):
_report_cooldown(stderr, state, config, now)
return 0
decision = partition_signatures(signatures, state, config)
if not decision.allowed:
_report_exhaustion(decision.capped, stderr, config)
return 0
_report_stall(hook_input, decision.allowed, stderr)
new_state = record_recovery(state, decision.allowed, now)
save_state(config, hook_input.session_id, new_state)
return 2
run itself keeps the same shape it had in step 4 — parse input, honor stop_hook_active, scan the transcript — and then hands off to _dispatch with an effective config (env-derived if the caller did not inject one) and an effective clock (defaulting to time.time()). Threading config, environ, and now as parameters is what lets the test suite drive dozens of scenarios without a single monkeypatch — every side effect the guardrails care about is an argument.
Verification
Run the full suite from inside codebase/ — the new file joins the four from previous steps and everything stays green:
python3 -m pytest
============================= test session starts ==============================
platform darwin -- Python 3.9.6, pytest-8.4.2, pluggy-1.6.0
rootdir: .../codebase
configfile: pyproject.toml
testpaths: tests
collected 49 items
tests/test_guardrails.py ..................... [ 42%]
tests/test_hook.py ....... [ 57%]
tests/test_recovery.py ........... [ 79%]
tests/test_stall_signatures.py ...... [ 91%]
tests/test_transcript.py .... [100%]
============================== 49 passed in 1.35s ==============================
Forty-nine tests, twenty-one of them brand-new in test_guardrails.py. The new cases cover the defaults, env-driven overrides, session-id sanitization, load-corrupt-JSON recovery, round-tripped state, cooldown timing on both sides of the boundary, retry-cap exhaustion after N successes, kill-switch precedence (env vs. file, truthy vs. falsy), and the stop_hook_active short-circuit still winning over every guardrail. The seven step-4 hook tests keep passing unchanged, which is exactly the regression signal we want: the guardrails are additive, not a rewrite.
What we built
The Stop hook now has three independent brakes. The retry cap stops the same broken tool_use_id from soliciting help forever — three nudges by default, tunable per operator. The cooldown window stops back-to-back Stop events from stacking nudges when a model keeps trying to end the turn — zero by default (opt-in), but any positive value silences the second, third, fourth attempt until the window elapses. The kill switch — STALL_WATCH_DISABLED=1 or a marker file — lets an operator turn recovery off in a single command without editing the hook script or the settings JSON.
State persists per session in a small JSON file under .claude/stall_watch/, and every state read is defensive: missing file, corrupt JSON, or malformed dict all collapse to a fresh, empty state rather than crashing the hook. The recovery path still returns exit code 2 the way step 4 defined it — the guardrails only ever downgrade to exit 0 and a diagnostic stderr line, never upgrade to something else. That preserves the contract every earlier step relies on.
The gate order is deliberate and stable: kill switch → cooldown → retry cap → recovery. An operator override always wins over a session-level throttle, which always wins over a per-tool budget, which always wins over the default behavior. Each gate emits a distinct stderr message (kill switch active, cooldown active, retry cap reached), so the log tells you exactly which brake fired and why.
What this unlocks is operational trust. The tutorial started with "detect a silent stall" and now ends with "recover from it — safely, boundedly, and with an off switch." A future step could add per-project template overrides, telemetry counters, or a /stall_watch reset slash command, and all of them plug in through the same GuardrailConfig + SessionState seam.
Repository
The state of the code after this step: 08905a0
Step 6: Closing the Loop with a JSONL Event Log and an End-to-End Smoke Test
Step 5 left us with a Stop hook that detects a stall, dispatches a recovery nudge, and enforces three independent guardrails around it. The behavior is right, but the observability is thin: stderr messages help a human read a live tail, but they are terrible to grep across a week of sessions, and none of the earlier tests exercise the full stall-then-recover cycle end to end. If an operator opens an incident asking "why did the hook not fire on session X?", we currently have no durable trail to answer them.
Step 6 fixes both gaps. A tiny event_log module writes one JSON line per hook decision — stall_detected, recovery_dispatched, retry_cap_hit, cooldown_skipped, kill_switch_active, healthy_stop — into an operator-configured log file, and a new test_smoke_end_to_end.py suite drives the hook the way Claude Code drives it (JSON on stdin, exit code out) all the way through recovery and back to a quiet stop. Nothing about the recovery contract from step 4 or the guardrails from step 5 changes; we only add taps.
Setup
The new logger lives in stall_watch/event_log.py. It has no dependencies beyond json, os, and pathlib, and it exposes three tiny functions: resolve_log_path (env-var lookup with an explicit override), log_event (append-one-line-of-JSON, no-op when the path is None), and read_log (parse the file back for tests and post-mortems). Every event constant — EVENT_STALL_DETECTED, EVENT_RECOVERY_DISPATCHED, EVENT_RETRY_CAP_HIT, EVENT_COOLDOWN_SKIPPED, EVENT_KILL_SWITCH_ACTIVE, EVENT_HEALTHY_STOP — is a module-scope string so tests can assert on them without stringly-typed literals.
stall_watch/hook.py grows one new parameter on run (log_path: Path | None) and a matching parameter on _dispatch. The hook resolves the effective log path once from STALL_WATCH_LOG_FILE (or the explicit override) and then calls log_event at every branch point. A brand-new tests/test_smoke_end_to_end.py drives four scenarios: the round-trip stall-and-recover cycle, the kill-switch path, the retry-cap-plus-cooldown path, and a subprocess invocation via python -m stall_watch.hook to prove the hook works exactly as Claude Code will call it.
Project layout after this step:
codebase/
├── pyproject.toml
├── stall_watch/
│ ├── __init__.py # + re-exports event_log symbols
│ ├── transcript.py # (unchanged)
│ ├── simulate.py # (unchanged)
│ ├── recovery.py # (unchanged)
│ ├── guardrails.py # (unchanged)
│ ├── event_log.py # NEW — JSONL event sink
│ └── hook.py # + log_event calls at every decision
└── tests/
├── conftest.py
├── test_transcript.py
├── test_stall_signatures.py
├── test_hook.py
├── test_recovery.py
├── test_guardrails.py
└── test_smoke_end_to_end.py # NEW — 6 scenario tests
No new dependencies. The log file lives wherever the operator points STALL_WATCH_LOG_FILE, so nothing is written by default — logging is strictly opt-in.
Implementation
The event_log module keeps two things separate: choosing the log path and writing to it. resolve_log_path prefers an explicit Path argument (used by tests) over the STALL_WATCH_LOG_FILE env var, and returns None when neither is set. Returning None is the whole reason log_event guards on if log_path is None: return — the caller never has to branch, and disabled logging costs one attribute check per event.
DEFAULT_LOG_FILE_ENV = "STALL_WATCH_LOG_FILE"
EVENT_STALL_DETECTED = "stall_detected"
EVENT_RECOVERY_DISPATCHED = "recovery_dispatched"
EVENT_RETRY_CAP_HIT = "retry_cap_hit"
EVENT_COOLDOWN_SKIPPED = "cooldown_skipped"
EVENT_KILL_SWITCH_ACTIVE = "kill_switch_active"
EVENT_HEALTHY_STOP = "healthy_stop"
def resolve_log_path(
environ: Mapping[str, str] | None,
explicit: Path | None = None,
) -> Path | None:
if explicit is not None:
return explicit
env = environ if environ is not None else os.environ
raw = env.get(DEFAULT_LOG_FILE_ENV)
if not raw:
return None
return Path(raw)
log_event writes exactly one JSON object per line, sorted keys, with an arbitrary payload merged on top of a fixed header (ts, event, session_id). Sorting the keys is what makes the log grep-friendly and diff-friendly across sessions — two recovery_dispatched events for different sessions produce byte-comparable prefixes, which lands well in jq filters and diff outputs.
def log_event(
log_path: Path | None,
session_id: str,
event: str,
now: float,
**payload: Any,
) -> None:
if log_path is None:
return
record: dict[str, Any] = {
"ts": now,
"event": event,
"session_id": session_id,
}
record.update(payload)
log_path.parent.mkdir(parents=True, exist_ok=True)
with log_path.open("a", encoding="utf-8") as handle:
handle.write(json.dumps(record, sort_keys=True) + "\n")
Inside hook.py, _dispatch now emits one event at every gate it evaluates. EVENT_STALL_DETECTED fires first — before the kill switch — so we always have a record that a stall was noticed even when a downstream guardrail silences the recovery. Then each branch logs its own outcome (kill_switch_active, cooldown_skipped, retry_cap_hit, or recovery_dispatched) with the exact state that drove the decision. That gives an operator a straight-line answer to "why did the hook do X on session Y?" without needing to replay the transcript.
def _dispatch(
hook_input: StopHookInput,
signatures: list[StallSignature],
config: GuardrailConfig,
environ: Mapping[str, str] | None,
now: float,
stderr: IO[str],
log_path: "Path | None",
) -> int:
session_id = hook_input.session_id
log_event(
log_path,
session_id,
EVENT_STALL_DETECTED,
now,
signatures=_signature_summary(signatures),
transcript=str(hook_input.transcript_path),
)
if is_kill_switch_active(config, environ):
_report_kill_switch(stderr, config)
log_event(log_path, session_id, EVENT_KILL_SWITCH_ACTIVE, now,
env=config.kill_switch_env)
return 0
...
The healthy-stop path — a transcript with no stall signatures — used to return 0 silently. It now emits EVENT_HEALTHY_STOP before returning, which turns "the log is empty" from an ambiguous signal into a precise one: an empty log means the operator never enabled logging; a log with only healthy-stop entries means the hook is running but never had anything to do. Both are useful; conflating them is not.
def run(
stdin, stderr, config=None, environ=None, now=None, log_path=None,
) -> int:
hook_input = parse_stop_hook_input(stdin.read())
if hook_input.stop_hook_active:
return 0
signatures = _scan_transcript(hook_input)
effective_now = now if now is not None else time.time()
effective_log_path = resolve_log_path(environ, explicit=log_path)
if not signatures:
log_event(effective_log_path, hook_input.session_id,
EVENT_HEALTHY_STOP, effective_now,
transcript=str(hook_input.transcript_path))
return 0
...
The smoke test asserts on the exact event sequence for each scenario, not just on the exit codes. That is the whole point of a smoke test at this stage: a passing exit code tells you the hook returned; a matching event sequence tells you it made the correct decision for the right reason. Below is the full stall-and-recover cycle — one Stop dispatches recovery, the simulated agent appends the missing tool_result, the second Stop sees a healthy transcript, and the log tells the whole story.
def test_smoke_full_stall_and_recover_cycle(tmp_path: Path) -> None:
transcript = tmp_path / "session.jsonl"
tool_id = write_stalled_transcript(transcript, tool_name="Read")
log_path = tmp_path / "stall_watch.log"
stdin = io.StringIO(_payload(transcript))
stderr = io.StringIO()
first = run(stdin, stderr,
environ={"STALL_WATCH_STATE_DIR": str(tmp_path / "state")},
now=1000.0, log_path=log_path)
assert first == 2
append_recovery(transcript, tool_id)
second = run(io.StringIO(_payload(transcript)), io.StringIO(),
environ={"STALL_WATCH_STATE_DIR": str(tmp_path / "state")},
now=1001.0, log_path=log_path)
assert second == 0
events = _events(log_path)
assert events == [
EVENT_STALL_DETECTED,
EVENT_RECOVERY_DISPATCHED,
EVENT_HEALTHY_STOP,
]
The subprocess smoke test earns its keep by proving the module actually runs as python -m stall_watch.hook with a real pipe on stdin. Tests that call run(...) directly can hide packaging bugs — a missing entry in __init__.py, an import-time side effect that only triggers under a fresh interpreter, a sys.stdin assumption. Driving it through subprocess.run closes that gap.
def test_smoke_runs_from_shell_via_python_m(tmp_path: Path) -> None:
transcript = tmp_path / "session.jsonl"
write_stalled_transcript(transcript, tool_name="Read")
log_path = tmp_path / "stall_watch.log"
payload = _payload(transcript, session_id="subprocess-smoke")
env = {
"PATH": Path(sys.executable).parent.as_posix(),
"STALL_WATCH_STATE_DIR": str(tmp_path / "state"),
"STALL_WATCH_LOG_FILE": str(log_path),
"STALL_WATCH_DISABLED": "",
}
repo_root = Path(__file__).resolve().parent.parent
result = subprocess.run(
[sys.executable, "-m", "stall_watch.hook"],
input=payload, capture_output=True, text=True,
env=env, cwd=str(repo_root), timeout=15,
)
assert result.returncode == 2, result.stderr
assert "stall_watch:" in result.stderr
assert _events(log_path) == [
EVENT_STALL_DETECTED,
EVENT_RECOVERY_DISPATCHED,
]
Verification
Run the full suite from inside the codebase — six new smoke tests join the existing forty-nine and everything stays green:
python3 -m pytest
============================= test session starts ==============================
platform darwin -- Python 3.9.6, pytest-8.4.2, pluggy-1.6.0
rootdir: .../codebase
configfile: pyproject.toml
testpaths: tests
collected 55 items
tests/test_guardrails.py ..................... [ 38%]
tests/test_hook.py ....... [ 50%]
tests/test_recovery.py ........... [ 70%]
tests/test_smoke_end_to_end.py ...... [ 81%]
tests/test_stall_signatures.py ...... [ 92%]
tests/test_transcript.py .... [100%]
============================== 55 passed in 2.52s ==============================
Fifty-five tests, six of them brand-new in test_smoke_end_to_end.py. The prior forty-nine — transcript parsing, stall signatures, hook input, recovery dispatcher, and the guardrails — all still pass unchanged, which is exactly the invariant we want: adding an event tap must not alter any recovery behavior. The smoke suite alone asserts on four end-to-end sequences (recover, kill-switch, cap-plus-cooldown, subprocess) plus two unit-level checks that resolve_log_path picks the right source and that log_event is a safe no-op when logging is off.
What we built
The Stop hook now writes a structured JSONL event log at every decision point. Turning it on is one env var — STALL_WATCH_LOG_FILE=/path/to/stall_watch.log — and every subsequent Stop event appends one line. The line contains the timestamp, event name, session id, and enough payload to reconstruct the decision: the stall signatures for stall_detected, the retry counters for recovery_dispatched, the capped signatures for retry_cap_hit, and the remaining cooldown for cooldown_skipped.
The smoke suite proves the whole pipeline from step 1 to step 6 works end to end. test_smoke_full_stall_and_recover_cycle writes a stalled transcript, runs the hook, appends the missing tool_result and follow-up, runs the hook again, and asserts the log ends in stall_detected → recovery_dispatched → healthy_stop. test_smoke_kill_switch_logged_and_skips_recovery and test_smoke_retry_cap_and_cooldown_leave_a_trail cover the other two ways a Stop event can be silenced, and test_smoke_runs_from_shell_via_python_m proves the packaged module behaves the same when Claude Code calls it as a subprocess.
Between the guardrails from step 5 and the event log from step 6, an operator now has both the levers to shape recovery and the observability to audit it. A misconfigured MCP server that keeps stalling shows up as a repeating pattern in the log, the retry cap kicks in and stops the nudge storm, and the log records exactly which tool_use_id exhausted its budget — all without a live tail on stderr.
That completes the arc the article set out to walk: reproduce a silent tool stall in step 1, scaffold a Stop hook in step 2, classify the stall in step 3, dispatch a recovery nudge in step 4, fence the recovery behind guardrails in step 5, and log every decision plus verify the whole thing end to end in step 6. The hook is now shippable — small enough to audit, testable in isolation and as a subprocess, safe by default, and observable when an operator opts in.
Repository
The state of the code after this step: 48612e9
Repository
Full source at https://github.com/vytharion/claude-code-stop-hook-silent-tool-stall-recovery.
Walk the lessons by stepping through the git commits in the repo — each major step has its own commit you can git checkout and rerun.