Layered defense against TUI pump-blocking hangs

[tony]

Summary

• Fix the two live CPU stalls that froze the Textual TUI on real stores: the find-in-detail re-highlight re-tokenized the whole JSON body on every keystroke/step, and a filter-widen re-rendered the entire results list on the pump.
• Add a runtime detection/prevention layer in ui/_runtime.py: a pump-thread sys.addaudithook that flags blocking-I/O initiation denylist-free, plus a log-only heartbeat watchdog that now defaults on for an interactive terminal.
• Widen the static non-blocking guard so it follows same-class helper calls, classifies @on/scheduler/call_from_thread/subscribe callees, and resolves import aliases against an expanded denylist.
• Add a reusable textual-non-blocking-pump skill plus AGENTS.md review rules and an ADR 0011 "coverage limits" record, so the methodology survives beyond this branch.
• Motivation: TUI hangs kept recurring. 100% static prevention is provably impossible (Rice's theorem — "blocks the pump" is a semantic, not syntactic, property), so this ships a two-gate defense — static analysis prevents the decidable subset at merge; a cause-agnostic runtime oracle detects the undecidable residue — and fixes the two hangs that were actually occurring.

Changes by area

The two live hangs

• src/agentgrep/ui/app_screen.py — Find-in-detail caches the syntax+search+filter base and copies it, overlaying only the find-match spans, so stepping matches (n/N) never re-tokenizes the body; the cache is invalidated on every detail re-render.
• src/agentgrep/ui/widgets/results.py — Results rows are memoized by record id, so a filter rebuild reuses renders instead of rebuilding every Text; invalidated on theme switch and cleared with the widget on a new search.

Runtime detection / prevention

• src/agentgrep/ui/_runtime.py — sys.addaudithook scoped to the pump thread flags blocking-I/O initiation (socket.connect, subprocess.Popen, sqlite3.connect, time.sleep, …) regardless of how the call is spelled; raises under pytest, logs interactively (opt-in via AGENTGREP_TUI_WATCHDOG). The heartbeat watchdog now defaults on for an interactive TTY (log-only), decoupled from the @pump_only/@offload asserts so the default can never crash a user on a latent violation.

Static guard

• tests/test_tui_non_blocking.py — Same-class interprocedural closure (follows self.helper()); @on and scheduler/call_from_thread/subscribe callee classification; import-aware denylist normalization plus conservative family expansion (network, fs-walk, json.load, input). The NB-9 JSON-confinement check stays intraprocedural by design.

Methodology

• .agents/skills/textual-non-blocking-pump/ — Pump-entrypoint catalog, the static-prevent vs runtime-detect split, per-change review rules, and a hang-audit procedure (.claude/skills symlinks to it).
• AGENTS.md, docs/dev/adr/0011-non-blocking-tui-invariants.md — Non-blocking review rules and the "Coverage limits and the runtime complement" record.

Design decisions

• Two-gate defense, not a "perfect" lint. Static analysis can only ever flag a syntactic proxy for a semantic property, so it prevents the enumerable, decidable subset at merge while a wall-clock watchdog catches the undecidable residue (CPU spin, native, data-dependent loops). The audit hook adds denylist-free prevention of I/O initiation specifically.
• Render cache, not an async restructure, for the filter fix. Profiling the rebuild (synthetic records, timings only) showed _render_record × N dominated (~75%) and was redundant — the rows were already rendered during streaming — so a synchronous id-keyed cache fixes it with zero concurrency risk. on_filter_completed is a pump-critical message handler, so awaiting a chunked apply would have blocked queued keystrokes anyway.
• Watchdog default-on, asserts opt-in. The watchdog is log-only and safe to default on; the @pump_only/@offload asserts can raise, so they stay gated on AGENTGREP_TUI_WATCHDOG and are deliberately decoupled from the watchdog predicate.
• NB-9 kept intraprocedural. The interprocedural closure runs only for the NB-1 blocking-call scan, so a pump method that calls the one sanctioned JSON builder is not falsely flagged.

Test plan

• test_pump_audit_hook_behavior (parametrized) — the hook flags blocking-I/O initiation only when armed and on the pump; a non-I/O event and a pure CPU spin never trip it
• test_runtime_enable_predicates (parametrized truth-table) — watchdog_enabled / guard-asserts / audit_hook_enabled decouple over env, pytest marker, and TTY
• test_forbidden_call_detector_resolves_aliases_and_families, test_classifier_sees_scheduled_callables_and_on_handlers, test_closure_follows_self_helpers — the guard catches alias/family evasion, @on/scheduler callees, and watch_x → _do_io → open()
• test_pump_methods_have_no_blocking_calls — the upgraded interprocedural scan passes clean on the real UI code
• test_detail_find_step_reuses_syntax_base — an n/N step reuses the cached base (no Syntax rebuild)
• test_filter_rebuild_reuses_cached_row_renders, test_new_search_clears_results_render_cache — a rebuild reuses cached rows; a fresh search releases them
• Full completion gate: ruff check, ruff format, ty check, py.test --reruns 0, just build-docs

[tony]

Code review

Found 1 issue:

1. _detail_find_base_for keys its find-overlay cache only on source, but bakes the live results-filter terms into the cached Text via _apply_filter_highlight (which reads self._filter_terms). The sole invalidation site, _present_detail, is skipped by on_filter_completed when the top filtered record is unchanged — so changing the results filter while find-in-detail is active returns stale filter highlights on every subsequent find keystroke/step. This is a regression from the pre-cache behavior that rebuilt the overlay each call. The sibling body cache _detail_cache_key already folds self._filter_terms into its key; this cache should too (or invalidate when the filter changes).

agentgrep/src/agentgrep/ui/app_screen.py

Lines 1996 to 2013 in 8a493e8

	cached = self._detail_find_base
	if cached is not None and self._detail_find_base_source == source:
	return cached
	if detect_content_format(source) == "json":
	text = _RichSyntax(source, "json", theme="ansi_dark", word_wrap=True).highlight(source)
	text.no_wrap = False
	self._apply_search_highlight(text)
	else:
	text = highlight_matches(
	source,
	list(self.query.terms),
	case_sensitive=self.query.case_sensitive,
	regex=self.query.regex,
	style=self._match_style("search"),
	)
	self._apply_filter_highlight(text)
	self._detail_find_base = text
	self._detail_find_base_source = source

🤖 Generated with Claude Code

_{- If this code review was useful, please react with 👍. Otherwise, react with 👎.}