Files
FabledScribe/tests/test_scribe_marker.py
T
bvandeusenandClaude Opus 5 a49e7ed2af
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
fix(plugin): the hooks need no jq and no tac (#4107)
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
2026-09-20 12:20:45 -04:00

151 lines
6.7 KiB
Python

"""`.scribe` — the second key a directory can be scoped by (#4085).
Every hook in `plugin/hooks/` scopes its request to a project, and until this
existed all six turned on one key: `git remote get-url origin`. That key does
not exist outside a git repo, so a session in a plain directory was unscoped
everywhere at once — silently, because a missing remote is indistinguishable
from a remote nobody bound.
`scribe_scope_query` is the shared answer, so these run the real shell rather
than a reimplementation of it. What they pin:
* the marker is found from a subdirectory, the way git finds its root;
* a bare integer works, because that is what a person writes by hand;
* the marker BEATS a git remote, since someone put the file there on purpose;
* an `instance` naming a different Scribe is refused — the case the field
exists for. A project id means nothing on its own: id 2 is a different
project on every instance, so a marker that travels would otherwise scope
the session to the wrong project, confidently and without a word.
That last one is the reason the file is JSON and not a number, so it is tested
from both directions: the matching host is honoured, the mismatched one is not.
"""
from __future__ import annotations
import json
import os
import shutil
import subprocess
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[1]
DEFS = ROOT / "plugin" / "hooks" / "scribe_defs.sh"
INSTANCE = "https://scribe.example.com"
def _sh(script: str, cwd: Path, env_extra: dict | None = None) -> str:
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)}
env.update(env_extra or {})
body = f'set -uo pipefail\n. "{DEFS}"\nurl="{INSTANCE}"\n{script}\n'
out = subprocess.run(["bash", "-c", body], cwd=cwd, capture_output=True,
text=True, env=env, timeout=30)
assert out.returncode == 0, out.stderr
return out.stdout.strip()
def _marker(d: Path, **fields) -> None:
(d / ".scribe").write_text(json.dumps(fields))
def _git_repo(d: Path, remote: str) -> None:
env = {"PATH": os.environ["PATH"], "HOME": str(d)}
subprocess.run(["git", "init", "-q"], cwd=d, check=True, env=env)
subprocess.run(["git", "remote", "add", "origin", remote], cwd=d, check=True, env=env)
def test_a_bare_integer_is_a_valid_marker(tmp_path):
"""What someone writes by hand. It parses as JSON already, so one filter
reads it and the written form alike."""
(tmp_path / ".scribe").write_text("7\n")
assert _sh(f'scribe_scope_query "{tmp_path}"', tmp_path) == "project_id=7"
def test_the_marker_is_found_from_a_subdirectory(tmp_path):
_marker(tmp_path, instance=INSTANCE, project_id=2, project="FabledScribe")
deep = tmp_path / "src" / "scribe" / "services"
deep.mkdir(parents=True)
assert _sh(f'scribe_scope_query "{deep}"', tmp_path) == "project_id=2"
def test_a_matching_instance_is_honoured(tmp_path):
"""Host-only comparison, so http/https and a trailing slash are not a
mismatch — only a genuinely different Scribe is."""
_marker(tmp_path, instance="http://scribe.example.com/", project_id=2)
assert _sh(f'scribe_scope_query "{tmp_path}"', tmp_path) == "project_id=2"
def test_a_marker_for_another_instance_is_refused_with_a_reason(tmp_path):
"""The case the `instance` field exists for. No project beats the wrong
project, and the reason is carried so the session can say which it was."""
_marker(tmp_path, instance="https://someone-elses.example.org", project_id=2)
assert _sh(f'scribe_scope_query "{tmp_path}"', tmp_path) == ""
said = _sh(f'scribe_marker_read "{tmp_path}/.scribe"', tmp_path)
assert "someone-elses.example.org" in said and "scribe.example.com" in said
def test_the_marker_beats_a_git_remote(tmp_path):
"""Someone put the file there deliberately; a remote is only where the
code happens to be pushed. This is also how a directory overrides its
binding."""
_git_repo(tmp_path, "git@git.example.com:someone/thing.git")
assert _sh(f'scribe_scope_query "{tmp_path}"', tmp_path).startswith("repo=")
_marker(tmp_path, instance=INSTANCE, project_id=2)
assert _sh(f'scribe_scope_query "{tmp_path}"', tmp_path) == "project_id=2"
def test_without_a_marker_the_git_remote_still_answers(tmp_path):
"""The path every existing install is on — it must not have moved."""
_git_repo(tmp_path, "git@git.example.com:someone/thing.git")
got = _sh(f'scribe_scope_query "{tmp_path}"', tmp_path)
assert got.startswith("repo=") and "git.example.com" in got.replace("%2F", "/")
def test_a_plain_directory_with_no_marker_scopes_to_nothing(tmp_path):
"""Not an error — the caller sends no scope and the server says so."""
assert _sh(f'scribe_scope_query "{tmp_path}"', tmp_path) == ""
@pytest.mark.parametrize("body", ["", "not json at all", "{}", '{"project_id": 0}',
'{"project_id": "../../etc"}', "[1,2,3]"])
def test_an_unusable_marker_is_refused_rather_than_sent(tmp_path, body):
"""A marker is operator-written and can say anything. Nothing that is not
a positive integer may reach the query string."""
(tmp_path / ".scribe").write_text(body)
assert _sh(f'scribe_scope_query "{tmp_path}"', tmp_path) == ""
def test_the_reason_survives_the_command_substitution_that_fetches_it(tmp_path):
"""Regression on a bug this nearly shipped with.
The reason used to be a global the function set, which every caller reads
through `$( )` — a subshell, so the assignment died with it and the caller
saw an UNSET variable. The hooks all run `set -u`, where reading one aborts
the script: a directory with a misaddressed marker would have cost the
session its entire SessionStart context, to fetch a warning about a file.
So the reason comes back through stdout with the id, and this asserts on
the shape the caller actually uses: read in a subshell, split, and used
under `set -u` without the shell dying.
"""
_marker(tmp_path, instance="https://elsewhere.example.org", project_id=2)
got = _sh(
f'read_out=$(scribe_marker_read "$(scribe_marker_file "{tmp_path}")")\n'
'printf "id=[%s] why=[%s]" "${read_out%%$\'\\t\'*}" "${read_out#*$\'\\t\'}"',
tmp_path,
)
assert got.startswith("id=[]")
assert "elsewhere.example.org" in got
def test_the_walk_up_terminates_at_the_root(tmp_path):
"""No marker anywhere above: the loop must end rather than spin. tmp_path
is several levels down from /, so this really does walk."""
deep = tmp_path / "a" / "b" / "c"
deep.mkdir(parents=True)
assert _sh(f'scribe_scope_query "{deep}"', tmp_path) == ""