"""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"(?