fix(plugin): the hooks need no jq and no tac (#4107)
CI & Build / Python lint (push) Successful in 2s
CI & Build / Plugin hooks (push) Successful in 9s
CI & Build / integration (push) Successful in 49s
CI & Build / TypeScript typecheck (push) Successful in 57s
CI & Build / Python tests (push) Failing after 1m7s
CI & Build / Build & push image (push) Skipped

Every hook opened `command -v jq >/dev/null 2>&1 || exit 0`, so on a machine
without jq the operator got no session context, no rules, no prior art and no
process sync — and not one word saying why, because `exit 0` is
indistinguishable from "ran fine, nothing to say". jq is absent by default on
macOS, on the Debian/Ubuntu slim images, on Alpine and in most CI containers.
That is not a prerequisite to document; it is the plugin handing its own
packaging problem to whoever installs it.

`tac` was worse: GNU-only, so the prior-art hook's enclosing-definition arm
did nothing at all on every Mac, silently, from the day it shipped. It is not
replaced but removed — scribe_defs judges each line independently, so
extracting forward and taking `tail -1` is the same answer as reversing and
taking the head, and it drops the early-exit `head` that #4042 was filed for.

No server contract changed, so a lagging plugin cache keeps working.

  scribe_json.awk   JSON -> IDX<TAB>PATH<TAB>VALUE. Two modes: `whole` for an
                    event or a response body, `lines` for a transcript, where
                    an unparseable record is dropped and the rest still read —
                    the `map(try fromjson catch empty)` the jq program opened
                    with. Arrays also report their LENGTH at `[#]`, which is
                    what keeps "zero notes" distinct from "no answer" (#2932).
  scribe_turn.awk   the turn-bounding program, replacing the thirty lines of
                    jq in the Stop hook.
  scribe_defs.sh    scribe_json_flat / _pick / _list / _len / _list_minus read,
                    scribe_json_out writes the envelope (five copies of one
                    shape, gone), scribe_urlenc replaces `jq -sRr '@uri'`.

Percent-encoding goes through `od -tu1` rather than an awk character loop on
purpose: awk's idea of a character follows the locale, so gawk reads an
accented letter as one and mawk as two, and an encoder built on substr() would
emit a different URL depending on which awk is installed. Encoding is defined
on bytes. Verified byte-identical to `jq -sRr '@uri'`.

Measured, not assumed. The per-event path costs 8ms against jq's 3ms. The
transcript path was 70x slower until two fixes: the Stop hook now finds where
the turn starts with a fixed-string grep before parsing (a needle carrying
unescaped quotes cannot occur inside a JSON string, so it matches only at a
record's top level — checked against a full JSON parse of a 27MB transcript:
152 prompt records, 152 matches, no misses, no extras), and the parser reads
each token out of a 1024-byte window instead of copying the rest of the buffer
per token, which was quadratic in line length on the 400KB tool results a
transcript carries.

Differential-tested against the jq program it replaces over 724 windows cut
from three real transcripts — 724 identical, 0 mismatched, 45 of them
exercising a real task close and a real reply. That sweep is what caught
`scribe_turn.awk` never setting FS, which truncated every multi-word reply at
its first space and was invisible to a test whose replies were all empty.

check_plugin.py's `jq -R` lint becomes a guard against either binary coming
back, and three smoke checks lose their `shutil.which("jq")` skip. jq is not
in `ci-python` either, so those three announced a skip on every CI run and had
never once run there: removing the dependency from the product also closed a
permanent hole in its verification. They pass now across all ten hooks.

tests/test_hook_json_reader.py is a differential against Python's `json` over
nested objects, arrays, unicode, escapes, control characters, empty cases and
a value longer than the token window, plus the envelope, the encoder and the
turn analyzer. 139 cases.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01821k5B3Ysecp9fNYs92Kuy
This commit is contained in:
2026-09-20 12:20:45 -04:00
co-authored by Claude Opus 5
parent 97867d47ff
commit a49e7ed2af
22 changed files with 1185 additions and 210 deletions
+1 -1
View File
@@ -21,7 +21,7 @@ HOOK = PLUGIN / "hooks" / "scribe_after_write.sh"
def _env(tmp_path, url="http://127.0.0.1:9"):
for tool in ("git", "jq", "curl", "bash"):
for tool in ("git", "curl", "bash"):
if shutil.which(tool) is None:
pytest.skip(f"hook runtime tool {tool!r} not installed")
return {"PATH": os.environ["PATH"], "SCRIBE_URL": url, "SCRIBE_TOKEN": "t",
+380
View File
@@ -0,0 +1,380 @@
"""JSON, read and written by the hooks without jq (#4107).
WHAT THIS REPLACED, AND WHY IT NEEDS TESTING AT ALL. Every hook used to open
`command -v jq >/dev/null 2>&1 || exit 0`, so on a machine without jq the
operator got no session context, no rules, no prior art and no process sync,
and not one word saying why — `exit 0` is indistinguishable from "ran fine,
nothing to say". jq is absent by default on macOS, on the Debian/Ubuntu slim
images, on Alpine and in most CI containers. The parser that replaced it is
`plugin/hooks/scribe_json.awk`, written in POSIX awk, and a parser is exactly
the kind of thing that works on the six documents you tried it on.
So the centre of this file is a DIFFERENTIAL against Python's `json`: the same
documents, flattened by both, compared. A reader that agrees with a real JSON
implementation on nested objects, arrays, escapes, unicode and the empty cases
is one the hooks can be handed an arbitrary event with.
The rest pins the three jobs around it — writing the hook envelope,
percent-encoding a URL, and bounding a turn in a transcript — and that each
guard here can fail (rule 167).
"""
from __future__ import annotations
import json
import re
import shutil
import subprocess
from pathlib import Path
from urllib.parse import quote
import pytest
HOOKS = Path(__file__).resolve().parents[1] / "plugin" / "hooks"
DEFS = HOOKS / "scribe_defs.sh"
PARSER = HOOKS / "scribe_json.awk"
TURN = HOOKS / "scribe_turn.awk"
def _need(*tools):
for t in tools:
if shutil.which(t) is None:
pytest.skip(f"hook runtime tool {t!r} not installed")
def sh(script: str, stdin: str = "") -> str:
"""Run a snippet with scribe_defs.sh sourced, under the hooks' own flags.
BYTES IN, BYTES OUT, decoded here — NOT `text=True`. Text mode turns on
universal newlines, which rewrites a carriage return in the output to a
newline before the assertion ever sees it. A test of an escaper cannot
quietly normalise the characters it exists to check: the first version of
this file did, and reported a round-trip failure that was entirely its own.
"""
_need("bash", "awk")
r = subprocess.run(
["bash", "-c", f'set -uo pipefail\n. "{DEFS}"\n{script}'],
input=stdin.encode(), capture_output=True,
)
assert r.returncode == 0, f"exit {r.returncode}: {r.stderr.decode()}"
return r.stdout.decode()
def flat(doc: str, mode: str = "whole") -> list[tuple[str, str, str]]:
_need("awk")
r = subprocess.run(["awk", "-v", f"mode={mode}", "-f", str(PARSER)],
input=doc.encode(), capture_output=True)
assert r.returncode == 0, r.stderr.decode()
return [tuple(line.split("\t", 2)) for line in r.stdout.decode().split("\n") if line]
# --------------------------------------------------------------------------
# The differential: this parser against Python's.
def reference_flatten(value, path="", out=None):
"""What scribe_json.awk is specified to emit, computed with `json`.
A string's VALUE is its RAW escaped body — the parser deliberately does not
decode, because a newline inside a value would break the line format the
shell reads it back with. Every array also reports its LENGTH at `[#]`,
including an empty one, and an empty object reports `{#}` 0 — that is what
lets a caller tell "the server answered with zero notes" from "the server
did not answer", which #2932 built an outage marker around.
"""
out = [] if out is None else out
if isinstance(value, dict):
if not value:
out.append((path + "{#}", "0"))
for k, v in value.items():
reference_flatten(v, f"{path}.{k}", out)
elif isinstance(value, list):
for i, v in enumerate(value):
reference_flatten(v, f"{path}[{i}]", out)
out.append((path + "[#]", str(len(value))))
elif isinstance(value, str):
out.append((path, json.dumps(value, ensure_ascii=False)[1:-1]))
elif value is None:
out.append((path, "null"))
elif isinstance(value, bool):
out.append((path, "true" if value else "false"))
else:
out.append((path, json.dumps(value)))
return out
NASTY = [
"plain",
"",
'has "quotes" inside',
"back \\ slash and \\\\ two",
"line one\nline two\nline three",
"tab\there and\ttwice",
"carriage\rreturn",
"héllo wörld — em dash",
"emoji \U0001f600 and \U0001f9ea and a ZWJ \U0001f469\U0001f4bb",
'mixed: "q" \\ \n \t é \U0001f600',
"a" * 3000, # longer than the parser's token WINDOW
'{"looks":"like json"}', # JSON inside a string must not be parsed
"trailing backslash \\",
" bell and  unit-sep",
]
DOCUMENTS = [
{"prompt": "write the tests first", "session_id": "abc-1", "cwd": "/x/y"},
{"tool_input": {"file_path": "/a/b.py", "content": "def f():\n pass\n"}},
{"note_ids": [1, 2, 3], "sync_note_ids": [], "rule_ids": [9]},
{"context": "", "n": None, "ok": True, "no": False, "num": -12.5, "exp": 1000.0},
{"processes": [{"name": "A", "slug": "a"}, {"name": "B", "slug": "b"}]},
{"deep": {"a": {"b": {"c": {"d": [{"e": "f"}]}}}}},
{"empty_obj": {}, "empty_arr": [], "nested_empty": {"x": []}},
{"message": {"role": "user", "content": [{"type": "text", "text": "hi"}]}},
2,
"bare string",
[1, [2, [3, []]]],
] + [{"v": s} for s in NASTY] + [{"k": {"inner": s, "arr": [s, s]}} for s in NASTY]
@pytest.mark.parametrize("doc", DOCUMENTS, ids=range(len(DOCUMENTS)))
def test_the_parser_agrees_with_a_real_json_implementation(doc):
encoded = json.dumps(doc, ensure_ascii=False)
got = [(p, v) for _idx, p, v in flat(encoded)]
assert got == reference_flatten(doc)
@pytest.mark.parametrize("doc", DOCUMENTS, ids=range(len(DOCUMENTS)))
def test_the_parser_reads_a_pretty_printed_document_the_same_way(doc):
"""Whole mode accumulates until the value is complete, so a document spread
over many lines reads identically to the compact form. Claude Code writes
compact events today; betting on that is how a reader breaks quietly."""
compact = [(p, v) for _i, p, v in flat(json.dumps(doc, ensure_ascii=False))]
pretty = [(p, v) for _i, p, v in flat(json.dumps(doc, indent=2, ensure_ascii=False))]
assert compact == pretty
@pytest.mark.parametrize("value", NASTY)
def test_a_value_survives_the_round_trip_the_hooks_actually_make(value):
"""Parse, pick, decode — what every hook does to read one field. The 3000
character entry matters most: it is longer than the parser's token window,
so it takes the wide-read fallback rather than the fast path."""
doc = json.dumps({"tool_input": {"content": value}}, ensure_ascii=False)
got = sh('flat=$(scribe_json_flat); scribe_json_pick "$flat" \'.tool_input.content\'',
stdin=doc)
assert got.rstrip("\n") == value.rstrip("\n")
# --------------------------------------------------------------------------
# Reading: the shell side.
def test_pick_list_and_len_read_what_the_server_sends():
body = json.dumps({"context": "some markdown", "note_ids": [11, 22, 33],
"sync_note_ids": [22], "derive_keys": ["k1", "k2"],
"rule_ids": [], "project": {"id": 7}})
out = sh(r"""
flat=$(scribe_json_flat)
echo "ctx=$(scribe_json_pick "$flat" '.context')"
echo "pid=$(scribe_json_pick "$flat" '.project.id')"
echo "notes=$(scribe_json_list "$flat" '.note_ids' | tr '\n' ',')"
echo "keys=$(scribe_json_list "$flat" '.derive_keys' | tr '\n' ',')"
echo "n=$(scribe_json_len "$flat" '.note_ids')"
echo "rules_n=$(scribe_json_len "$flat" '.rule_ids')"
echo "minus=$(scribe_json_list_minus "$flat" '.note_ids' '.sync_note_ids' | tr '\n' ',')"
""", stdin=body)
got = dict(line.split("=", 1) for line in out.splitlines())
assert got == {"ctx": "some markdown", "pid": "7", "notes": "11,22,33,",
"keys": "k1,k2,", "n": "3", "rules_n": "0", "minus": "11,33,"}
def test_an_absent_field_and_a_null_field_both_read_as_empty():
"""`// empty` is what every call site carried, and the shells downstream
test `[ -n "$x" ]`. A field the server left null and one it never sent mean
the same thing to all of them."""
out = sh(r"""
flat=$(scribe_json_flat)
printf '[%s][%s][%s]\n' "$(scribe_json_pick "$flat" '.nothing')" \
"$(scribe_json_pick "$flat" '.nulled')" "$(scribe_json_pick "$flat" '.real')"
""", stdin=json.dumps({"nulled": None, "real": "here"}))
assert out.strip() == "[][][here]"
def test_an_absent_array_length_is_empty_not_zero():
"""NOT the same answer as 0, and the distinction is load-bearing: the
record nudge fires when the server said "zero notes" and must not fire
when the server said nothing at all (#2932)."""
out = sh(r"""flat=$(scribe_json_flat)
printf '[%s][%s]\n' "$(scribe_json_len "$flat" '.note_ids')" \
"$(scribe_json_len "$flat" '.missing')" """,
stdin=json.dumps({"note_ids": []}))
assert out.strip() == "[0][]"
def test_a_document_that_does_not_parse_yields_nothing():
for broken in ['{"a":', '{"a" 1}', "not json at all", '{"a":1}{"b":2}', ""]:
assert flat(broken) == [], broken
def test_lines_mode_drops_only_the_record_that_did_not_parse():
"""`tail -n 3000` of a transcript cuts wherever it cuts, so the first line
is routinely half a record. This is the `map(try fromjson catch empty)`
the jq program it replaces opened with."""
doc = "\n".join(['ent":"truncated"}',
json.dumps({"type": "user", "n": 1}),
"{ also broken",
json.dumps({"type": "assistant", "n": 2})])
rows = flat(doc, mode="lines")
assert [r for r in rows if r[1] == ".n"] == [("1", ".n", "1"), ("2", ".n", "2")]
# --------------------------------------------------------------------------
# Writing, and encoding.
@pytest.mark.parametrize("value", NASTY)
def test_the_hook_envelope_is_valid_json_carrying_the_exact_context(value):
"""The one JSON document these hooks WRITE. Five hooks emitted it through
`jq -n --arg c`; getting the escaping wrong corrupts every injected context
that contains a Windows path, a regex or a quoted word."""
out = sh('scribe_json_out PreToolUse "$(cat)"', stdin=value)
parsed = json.loads(out)
assert parsed["hookSpecificOutput"]["hookEventName"] == "PreToolUse"
assert parsed["hookSpecificOutput"]["additionalContext"] == value.rstrip("\n")
@pytest.mark.parametrize("value", NASTY + ["a b&c=d", "~!*()", "/path/to?q=1#f", "100%"])
def test_urlenc_percent_encodes_exactly_the_unreserved_set(value):
"""Byte-exact, which is why it goes through `od` rather than an awk
character loop: awk's idea of a character follows the locale, so gawk reads
an accented letter as one and mawk as two, and an encoder built on substr()
would produce a different URL depending on which awk is installed.
Percent-encoding is defined on BYTES."""
got = sh("scribe_urlenc", stdin=value)
assert got == quote(value, safe="")
def test_urlenc_and_the_envelope_can_fail():
"""Rule 167 — a guard that cannot fail is decoration. If the escaper were a
no-op, this pair would be the assertion that noticed."""
assert sh("scribe_urlenc", stdin="a b") != "a b"
raw = sh('scribe_json_out PreToolUse "$(cat)"', stdin='he said "hi"')
assert '\\"hi\\"' in raw, raw
# --------------------------------------------------------------------------
# The transcript turn, which is the largest thing jq was doing here.
def _turn(records: list[dict]) -> dict:
_need("awk")
doc = "\n".join(json.dumps(r) for r in records) + "\n"
p1 = subprocess.run(["awk", "-v", "mode=lines", "-f", str(PARSER)],
input=doc, capture_output=True, text=True)
assert p1.returncode == 0, p1.stderr
p2 = subprocess.run(["awk", "-f", str(TURN)], input=p1.stdout,
capture_output=True, text=True)
assert p2.returncode == 0, p2.stderr
out = {}
for line in p2.stdout.splitlines():
k, _, v = line.partition("\t")
out[k] = v
return out
def _prompt(text="do the thing", **kw):
return {"type": "user", "message": {"role": "user", "content": text}, **kw}
def _close(tid="t1", task_id=41, status="done", name="mcp__x__update_task", **kw):
return {"type": "assistant", "message": {"content": [
{"type": "tool_use", "id": tid, "name": name,
"input": {"task_id": task_id, "status": status}}]}, **kw}
def _text(body, **kw):
return {"type": "assistant",
"message": {"content": [{"type": "text", "text": body}]}, **kw}
def _error(tid="t1"):
return {"type": "user", "message": {"content": [
{"type": "tool_result", "tool_use_id": tid, "is_error": True}]}}
def test_a_window_with_no_prompt_is_not_bounded():
assert _turn([_close(), _text("done")])["bounded"] == ""
def test_the_turn_starts_at_the_last_prompt():
facts = _turn([_prompt("first"), _close("t1", 11), _prompt("second"),
_close("t2", 22), _text("report")])
assert facts["bounded"] == "1"
assert facts["closed"] == "1"
assert facts["task_ids"] == "22"
def test_a_close_that_errored_is_not_a_close():
"""The tool_result carrying `is_error` arrives in a LATER record than the
tool_use it refutes, so a single forward pass would already have counted
it. Closing a task that failed to write must not count as closing it."""
assert _turn([_prompt(), _close("t1", 41), _error("t1"), _text("r")])["closed"] == "0"
assert _turn([_prompt(), _close("t1", 41), _error("t9"), _text("r")])["closed"] == "1"
def test_a_subagents_work_is_not_this_sessions():
"""Sidechain records interleave into the same file. A subagent closing a
task is not the operator's session closing one, and counting those made the
check fire on turns that closed nothing."""
assert _turn([_prompt(), _close("t1", 41, isSidechain=True), _text("r")])["closed"] == "0"
side = _turn([_prompt(), _prompt("subagent asked", isSidechain=True),
_close("t1", 41), _text("r")])
assert side["closed"] == "1", "a sidechain PROMPT must not re-cut the turn"
def test_a_meta_record_does_not_start_a_turn():
facts = _turn([_prompt("real"), _close("t1", 41),
_prompt("injected", isMeta=True), _text("report")])
assert facts["closed"] == "1"
def test_the_reply_is_the_text_after_the_last_action_and_keeps_its_newlines():
"""Text emitted BETWEEN two tool calls is narration mid-work, not a report;
holding it to the report shape would block turns that did report properly
at the end. The reply comes back still escaped and on one line, so the
format survives a multi-line answer."""
facts = _turn([_prompt(), _text("thinking out loud"), _close("t1", 41),
_text("line one\nline two"), _text("line three")])
assert facts["reply"] == "line one\\nline two\\nline three"
decoded = sh("scribe_json_unescape", stdin=facts["reply"])
assert decoded == "line one\nline two\nline three\n"
assert "thinking out loud" not in decoded
def test_a_tool_result_does_not_read_as_a_prompt():
"""A prompt's `message.content` is a STRING; a tool result carries an ARRAY
at the same path. Counting one as a prompt would cut the turn in the wrong
place — which is the whole bounding decision."""
facts = _turn([_prompt("real"), _close("t1", 41), _error("t9"), _text("report")])
assert facts["closed"] == "1" and facts["task_ids"] == "41"
def test_the_turn_guards_can_fail():
"""Rule 167. If the analyzer counted everything, or nothing, these are the
assertions that would notice."""
assert _turn([_prompt(), _close("t1", 41), _text("r")])["closed"] == "1"
assert _turn([_prompt(), _text("r")])["closed"] == "0"
# --------------------------------------------------------------------------
# And the dependency itself.
def test_no_hook_reaches_for_jq_or_tac():
"""The point of all of the above. A hook may use only what POSIX
guarantees: jq is not installed by default on macOS, the Debian/Ubuntu slim
images, Alpine or most CI containers, and tac is GNU-only — absent on macOS,
where the prior-art hook's enclosing-definition arm silently did nothing
from the day it shipped. Comment lines are exempt: several hooks now name
the old form so the next reader knows what changed."""
banned = re.compile(r"(?<![\w./-])(jq|tac)(?![\w./-])")
offenders = []
for script in sorted(HOOKS.glob("*.sh")) + sorted(HOOKS.glob("*.awk")):
for n, line in enumerate(script.read_text().splitlines(), 1):
if line.lstrip().startswith("#"):
continue
if banned.search(line):
offenders.append(f"{script.name}:{n}: {line.strip()}")
assert not offenders, "\n".join(offenders)
+7 -2
View File
@@ -282,6 +282,11 @@ def test_the_shipped_manifest_carries_a_minted_version():
def test_the_session_context_hook_still_reads_the_version_field():
"""The marker #2220 asked for. The value's SHAPE changed, not the field or
its reader — if this had to move, the derivation went somewhere it should
not have."""
not have.
The READER changed once, in #4107: `jq -r '.version'` became
`scribe_json_pick ... '.version'` when the hooks stopped depending on jq.
What this pins is unchanged — that the hook still reads `.version` out of
the manifest rather than deriving the string some other way."""
hook = (check_plugin.HOOKS_DIR / "scribe_session_context.sh").read_text()
assert re.search(r"jq\s+-r\s+'\.version", hook)
assert re.search(r"scribe_json_pick[^\n]*'\.version'", hook)
+1 -1
View File
@@ -28,7 +28,7 @@ REASON = "SERVER REASON: rewrite as a completion report"
def _env(tmp_path, url="http://127.0.0.1:9"):
for tool in ("jq", "curl", "bash"):
for tool in ("curl", "bash"):
if shutil.which(tool) is None:
pytest.skip(f"hook runtime tool {tool!r} not installed")
return {"PATH": os.environ["PATH"], "SCRIBE_URL": url, "SCRIBE_TOKEN": "t",
+1 -1
View File
@@ -92,7 +92,7 @@ def test_held_outranks_seen_regardless_of_kind():
# ── the recorder ───────────────────────────────────────────────────────────
def _run_recorder(event: dict, tmp: Path) -> Path:
for tool in ("bash", "jq"):
for tool in ("bash",):
if shutil.which(tool) is None:
pytest.skip(f"hook runtime tool {tool!r} not installed")
env = {"PATH": os.environ["PATH"], "HOME": str(tmp), "TMPDIR": str(tmp)}
+1 -1
View File
@@ -36,7 +36,7 @@ INSTANCE = "https://scribe.example.com"
def _sh(script: str, cwd: Path, env_extra: dict | None = None) -> str:
for tool in ("bash", "jq"):
for tool in ("bash",):
if shutil.which(tool) is None:
pytest.skip(f"hook runtime tool {tool!r} not installed")
env = {"PATH": os.environ["PATH"], "HOME": str(cwd)}
+1 -1
View File
@@ -51,7 +51,7 @@ def _env(tmp_path):
the clear happens with no credentials and no network at all — so sharing a
helper would mean testing this case in an environment that cannot show it.
"""
for tool in ("jq", "bash"):
for tool in ("bash",):
if shutil.which(tool) is None:
pytest.skip(f"hook runtime tool {tool!r} not installed")
# No SCRIBE_URL / SCRIBE_TOKEN on purpose — see the module docstring.
+1 -1
View File
@@ -77,7 +77,7 @@ def _swept_dirs() -> set[str]:
def _run_session_start(source: str, tmp: Path) -> Path:
"""Run the SessionStart hook for real, with the ledger directories filled."""
for tool in ("bash", "jq"):
for tool in ("bash",):
if shutil.which(tool) is None:
pytest.skip(f"hook runtime tool {tool!r} not installed")
+1 -1
View File
@@ -1108,7 +1108,7 @@ def _hook_runtime_env():
import os
import shutil
for tool in ("git", "jq", "curl", "bash"):
for tool in ("git", "curl", "bash"):
if shutil.which(tool) is None:
pytest.skip(f"hook runtime tool {tool!r} not installed")
return {"PATH": os.environ["PATH"],