test_run_context.py

Claude / Hooks / test_run_context

Type: block (can deny or stop) · Events: SessionStart, UserPromptSubmit, PreToolUse · Registrations: 3

test_run_context.py tells the model what test runs are in flight on the box and denies CPU-heavy commands while an authoritative run is measuring. One script, three registrations: SessionStart and UserPromptSubmit print the context block, and PreToolUse(Bash) acts as the guard.

The scan walks /proc. A process counts as an authoritative run when one whole argv element (no whitespace) names a runner such as run_tests.py; it counts as a pytest process when an element is pytest, py.test, -m pytest or -mpytest. A quoted pattern or -c payload is one element with spaces and is skipped, which stops a pgrep watcher being counted as a run. Workers are folded into their owning runner by walking parent pids, and a run is marked foreign when its uid or checkout root differs from this session's (a CI job, a second worktree). The block lists each run with pid, project and elapsed time, the load average against the usable CPUs (idle, BUSY at 60 percent, SATURATED at 100 percent), the last result from reports/run-tests-latest.json, and the standing rules (cap interactive runs at 5 minutes, run targeted modules, observe live). On UserPromptSubmit it prints nothing unless a run is in flight. As a guard it denies a command that matches the heavy pattern when an authoritative run exists and the command is not purely observation (pytest-live, observe-pytest and similar, with segments split outside quotes and 2>&1 kept whole), unless it carries the override token.

Denial text names the run and offers two ways out: wait, or suspend your own job with kill -STOP <pid> and resume with kill -CONT; a genuinely non-competing command (one targeted module, a pinned single-core job) takes the override token plus a stated reason. Related: shell-guards section 2, observe-work, force-test-observation, RUL-032, RUL-009, RUL-040.

Earlier documentation

Source: claude/hooks/test_run_context.py (orchestrata-ultimate-architecture) · SessionStart + PreToolUse(Bash) · Python 3

Two jobs: report what test run is in flight, and refuse to start CPU-heavy work beside it.

Why it exists

A rule that was already known did not stop it being broken twice in two days: a «private-repo» full run was started beside orchestrata's 1631-module ship run, then a wiki RAG re-ingest beside the next one. The first corrupted a 360.72s-vs-369.75s comparison; the second ran the box oversubscribed while the gate was measuring. Port, database and CPU contention then fails runs for reasons that are not real defects.

What it reports (SessionStart)

• authoritative runs in flight, collapsed to the process tree's root — reporting each matching process separately once turned one ship into five "runs in flight";
• load against len(os.sched_getaffinity(0)), not cpu_count(), so a pinned box does not read as idle;
• the last authoritative result — and an aborted run is reported as ABORTED, never as "0 passed". Stating a number the run never produced is the one thing an always-on context block must not do.

What it denies (PreToolUse Bash)

A command matching _HEAVY_RE while an authoritative run is in flight is denied, with the owning run's pid in the reason. Override with ORCH_ALLOW_CONTENDING=1 and a stated reason.

Observation is never denied

_is_observation() allows a command whose every segment starts with a read-only tool (tail, head, grep, pgrep, cat, wc, ls, stat, echo, sleep, ps, uptime, observe-pytest, observe_pytest.py).

This has been the source of four false positives, all one bug: the matcher reads argument text as intent. Each is now a permanent test.

The third is the worst of them: the platform conventions require observing a run rather than waiting blind, and the shell spelling of that is an until/do/done loop — so the guard was rejecting the exact discipline it exists to enforce.

The skip still fails closed

Keywords are stepped over to find the command word, which is then still checked. A heavy command inside a loop body, or behind time, stays denied. A skip that stopped checking would make time <anything> a universal bypass — a far larger hole than the false positives it fixes — so the test suite carries a deny corpus alongside the allow corpus.

Design note

A guard that obstructs ordinary work gets disabled, and a disabled guard stops nothing. That is why every false positive above is treated as a defect in the guard rather than something for the user to work around.

Tests

tests/claude/test_test_run_context.py — spawns real process trees and lets the hook read real /proc entries. A faked /proc would test the parser and prove nothing about the thing the hook exists to detect.

Registration

Commands (as registered, abridged):

• python3 "${CLAUDE_PLUGIN_ROOT}"/hooks/test_run_context.py

Location

• Script: plugins/orchestrata/hooks/test_run_context.py (repository path at 8a2cb42ad)
• Deployed: /home/mohamed/.claude/plugins/cache/orchestrata/orchestrata/1.5.0/hooks/test_run_context.py
• Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/8a2cb42ad/plugins/orchestrata/hooks/test_run_context.py

Source

test_run_context.py (deployed copy)

731 lines.

#!/usr/bin/env python3
"""Test-run context: what is running on this box, and what that forbids.

The user's instruction (2026-08-14): "I want a test run context to be injected
in future so you do not forget" — after I twice started a competing CPU job
beside a live authoritative suite:

* the «private-repo» full suite alongside orchestrata's 1631-module ship run, which
  made the 360.72s-vs-369.75s comparison worthless and contributed to the run
  the ship then failed on
* a wiki RAG re-ingest alongside the next ship's suite, pushing the 1-minute
  load past the logical-CPU count while the authoritative gate was measuring

Both times I knew the rule. Knowing it is not what stops it, because at the
moment of starting the second job the first one is invisible: it is a
background pid with no presence in the conversation. So this hook makes it
visible, and refuses the collision outright.

Two jobs, one file:

1. **SessionStart / UserPromptSubmit** — emit the live state: what test runs
   are in flight, how long they have been going, how loaded the box is, and
   what the last authoritative run reported.
2. **PreToolUse(Bash)** — DENY a heavy CPU command while an authoritative run
   is in flight, naming the pid and the remedy.

Job 2 exists because job 1 alone is the thing that already failed. A note in
context is advisory; the gate is not.

Cost: reads /proc and two small files, no subprocesses. Never blocks a session:
every non-PreToolUse path exits 0.
"""

from __future__ import annotations

import json
import os
import re
import sys
import time
from pathlib import Path

# ── what counts as a test run ────────────────────────────────────────────────
#
# Matched against the process cmdline. `run_tests.py` is the AUTHORITATIVE
# batched runner — a run of it is the thing nothing may compete with. A bare
# pytest is a test run too, but a targeted one, and blocking on it would make
# ordinary work impossible.
#: Scripts whose EXECUTION is an authoritative run. Matched per-argv-element
#: (see `_is_authoritative`), never as a substring of the whole command line.
_AUTHORITATIVE_SCRIPTS = frozenset(
    {"run_tests.py", "run_tests_streaming.py", "run_tests_threadpool.py"}
)
_AUTHORITATIVE_SUFFIXES = ("deploy/ship.sh",)
_PYTEST_RE = re.compile(r"\bpytest\b|-m\s+pytest")

# ── what must not start beside one ───────────────────────────────────────────
#
# Long, parallel, CPU-bound work. Deliberately NOT "anything python": a
# `python -c` one-liner costs nothing and blocking it would train me to work
# around this hook, which is how a guard becomes noise.
_HEAVY_RE = re.compile(
    r"""
    \bpytest\b
  | run_tests(_streaming|_threadpool)?\.py
  | \bship\.sh\b
  | rag-ingest | rag_ingest | \bingest\b
  | \buv\s+sync\b
  | \bcargo\s+build\b
  | \bmake\b(?!\s*-n)
  | benchmark
  | \bnproc\b\s*\)\s*-j          # explicit parallel builds
    """,
    re.VERBOSE,
)

_REPO_MARKERS = ("pyproject.toml", ".git")


def _argv(pid: str) -> list[str]:
    """The process's argv, as separate elements.

    `_cmdline` deliberately flattens the NUL separators so patterns can match
    across argument boundaries. That is right for *detecting* pytest and wrong
    for deciding whether a process IS a run: flattened, an argument that merely
    QUOTES a script path is indistinguishable from one that executes it.
    """
    try:
        raw = Path("/proc", pid, "cmdline").read_bytes()
    except OSError:
        return []
    return [part.decode("utf-8", "replace") for part in raw.split(b"\0") if part]


def _is_authoritative(argv: list[str]) -> bool:
    """True when this process EXECUTES a runner, not merely names one.

    The old test was a substring search over the whole command line, so any
    process that mentioned the filename counted as a run in flight: a
    `pgrep -f "…run_tests.py"` watcher, a grep, a `tail | grep` pipeline. On
    2026-08-17 a monitor watching a ship was itself reported as a second
    authoritative run, which is exactly what
    tests/claude/test_test_run_context.py asserts must not happen — the test
    was right and had been calling this out.

    An executed path is a WHOLE argv element and contains no whitespace; a
    quoted pattern or a `-c` payload is one element carrying spaces or
    newlines. That distinction is the entire fix, and it keeps the real
    runner matching: the runner's path is its own argv element.
    """
    for word in argv:
        if not word or any(character.isspace() for character in word):
            continue
        if word.rsplit("/", 1)[-1] in _AUTHORITATIVE_SCRIPTS:
            return True
        if any(word.endswith(suffix) for suffix in _AUTHORITATIVE_SUFFIXES):
            return True
    return False


def _is_pytest(argv: list[str]) -> bool:
    """True when this process EXECUTES pytest, not merely names it.

    §1.3 of the 2026-10-11 evidence: for 53 minutes the context block
    reported pid 930286 — an opencode dispatcher whose PROMPT (one argv
    element carrying whitespace) contained the word "pytest" — as
    "1 targeted pytest process(es) running". The check then was
    ``_PYTEST_RE.search`` over the flattened whole cmdline, which cannot
    distinguish executing pytest from quoting it: the same substring defect
    ``_is_authoritative`` had already fixed for the runners, fixed here with
    the same rule — an argv element that contains whitespace is quoted text,
    not a command, so it is skipped. Of the remaining elements the process
    executes pytest exactly when one's basename is ``pytest`` or ``py.test``,
    or an element equal to ``-m`` is immediately followed by an element
    equal to ``pytest``, or an element equals ``-mpytest``.
    """
    for index, word in enumerate(argv):
        if not word or any(character.isspace() for character in word):
            continue
        if word.rsplit("/", 1)[-1] in ("pytest", "py.test"):
            return True
        if word == "-m" and argv[index + 1 : index + 2] == ["pytest"]:
            return True
        if word == "-mpytest":
            return True
    return False


def _cmdline(pid: str) -> str:
    """The full cmdline, NUL-joined into spaces.

    /proc cmdline is NUL-separated; reading it as text and matching without
    translating the separators silently fails to match across argument
    boundaries — the exact false negative that once spawned a duplicate IRC
    listener.
    """
    try:
        raw = Path("/proc", pid, "cmdline").read_bytes()
    except OSError:
        return ""
    return raw.replace(b"\0", b" ").decode("utf-8", "replace").strip()


def _cwd(pid: str) -> str:
    try:
        return os.readlink(f"/proc/{pid}/cwd")
    except OSError:
        return ""


def _elapsed(pid: str) -> int:
    """Seconds since the process started, or -1 when unknowable."""
    try:
        start_ticks = int(
            Path("/proc", pid, "stat").read_text().rsplit(") ", 1)[1].split()[19]
        )
        uptime = float(Path("/proc/uptime").read_text().split()[0])
    except OSError, IndexError, ValueError:
        return -1
    return max(0, int(uptime - start_ticks / os.sysconf("SC_CLK_TCK")))


def _project_of(cwd: str) -> str:
    """The repo a run belongs to, so two projects' runs are distinguishable."""
    path = Path(cwd) if cwd else None
    while path and path != path.parent:
        if any((path / marker).exists() for marker in _REPO_MARKERS):
            return path.name
        path = path.parent
    return Path(cwd).name if cwd else "?"


def _repo_root_of(cwd: str) -> str:
    """The CHECKOUT a run belongs to — the path, not the name.

    ``_project_of`` answers "which project", and two checkouts of the SAME
    project answer it identically: ``/home/mohamed/workspace/orchestrata-
    ultimate-architecture`` and a CI runner's ``/builds/<user>/orchestrata-
    ultimate-architecture`` are both "orchestrata-ultimate-architecture". So a
    CI job was indistinguishable from the local run and was counted as one of
    THIS checkout's authoritative runs. Under ``ship`` that made two runs where
    the collapse invariant expects one, and failed the ship this hook exists to
    protect (measured 2026-08-26). The path is what identifies a checkout.

    Returns "" when no repo marker is found — an unknown root is never treated
    as a match, so an unidentifiable run stays visible rather than being
    silently folded into ours.
    """
    path = Path(cwd) if cwd else None
    while path and path != path.parent:
        if any((path / marker).exists() for marker in _REPO_MARKERS):
            return str(path)
        path = path.parent
    return ""


def _here_root() -> str:
    """This checkout's root — the one whose runs are OURS."""
    try:
        return _repo_root_of(os.getcwd())
    except OSError:
        return ""


def _uid_of(pid: str) -> str:
    """The real uid a process runs as, or "" when it cannot be read.

    The second half of "is this run OURS", and the half that actually catches
    the failing case. ``/proc/<pid>/cwd`` is a symlink only its OWNER may
    resolve, so for a CI job running as another user ``_cwd`` returns "" and
    the root comparison has nothing to compare — which is why the context
    block rendered that job as ``pid 1915816 · ?`` while still counting it as
    one of ours. ``/proc/<pid>/status`` is world-readable, so the uid survives
    exactly the boundary the cwd does not.

    A run under a different uid is by construction not this session's run.
    """
    try:
        status = Path(f"/proc/{pid}/status").read_text(encoding="utf-8")
    except OSError:
        return ""
    for line in status.splitlines():
        if line.startswith("Uid:"):
            fields = line.split()
            return fields[1] if len(fields) > 1 else ""
    return ""


def _ppid(pid: str) -> str:
    """The parent pid, or "" — read from stat, after the ")" that ends comm.

    Splitting on whitespace from the left breaks on any process whose comm
    contains a space or a paren; the field layout is only unambiguous after
    the final ")".
    """
    try:
        after = Path("/proc", pid, "stat").read_text().rsplit(") ", 1)[1].split()
    except OSError, IndexError:
        return ""
    return after[1] if len(after) > 1 else ""


def _is_foreign(record: dict, here: str, here_uid: str) -> bool:
    """Is this authoritative run somebody ELSE's?

    It competes for this box's CPU and ports exactly as much as ours does, so
    it stays visible either way — but it is not OUR run, and counting it as
    one is what broke the collapse invariant and failed a ship (2026-08-26).

    TWO independent grounds, because either signal alone leaves a hole:

    ``uid``
        catches the CI job, whose ``/proc/<pid>/cwd`` another user may not
        resolve — so its root reads as unknown and a root-only rule would
        never fire. This is the ground that catches the REAL failure.
    ``root``
        catches a second checkout run by ME (a worktree, a clone, a
        deployment), where the uid is identical and only the path differs.

    Each ground is consulted only when its input is KNOWN. An unreadable uid
    or an unidentifiable root is never assumed foreign: an unknown run stays
    visible as ours rather than being quietly reclassified into the quieter
    bucket, which is the failure mode that would hide a real contender.

    ``here``/``here_uid`` are parameters rather than reads of the ambient
    process because "foreign" is a question about a VIEWPOINT, and a rule you
    cannot ask from another viewpoint cannot be tested from one either.
    """
    if not record["authoritative"]:
        return False
    not_our_user = bool(here_uid and record["uid"] and record["uid"] != here_uid)
    not_our_tree = bool(here and record["root"] and record["root"] != here)
    return not_our_user or not_our_tree


def _runs() -> tuple[list[dict], list[dict]]:
    """(authoritative runs, other pytest processes) currently alive.

    A batched run is a process TREE — the runner, its eight workers, and every
    pytest subprocess under them. Reporting each match separately turned one
    ship into five "runs in flight" and made the block read as chaos. Only a
    process with no matched ancestor is a run; the rest are its workers.
    """
    me = os.getpid()
    matched: dict[str, dict] = {}
    for entry in Path("/proc").iterdir():
        if not entry.name.isdigit() or int(entry.name) == me:
            continue
        cmd = _cmdline(entry.name)
        if not cmd:
            continue
        argv = _argv(entry.name)
        is_auth = _is_authoritative(argv)
        if not is_auth and not _is_pytest(argv):
            continue
        cwd = _cwd(entry.name)
        matched[entry.name] = {
            "pid": entry.name,
            "project": _project_of(cwd),
            "root": _repo_root_of(cwd),
            "uid": _uid_of(entry.name),
            "elapsed": _elapsed(entry.name),
            "authoritative": is_auth,
            "foreign": False,
            "workers": 0,
        }

    authoritative: list[dict] = []
    others: list[dict] = []
    here = _here_root()
    here_uid = str(os.getuid())
    for pid, record in matched.items():
        # The nearest ancestor that OWNS this process. An authoritative run is
        # only ever a worker of another AUTHORITATIVE run (ship.sh -> run_tests
        # .py collapses to one line); it is never demoted under a plain pytest,
        # which would otherwise hide a real run started from a test — the case
        # tests/claude/test_test_run_context.py caught.
        owner = None
        walker, seen = pid, {pid}
        while True:
            parent = _ppid(walker)
            if not parent or parent in seen or parent == "1":
                break
            seen.add(parent)
            candidate = matched.get(parent)
            if candidate is not None and (
                candidate["authoritative"] or not record["authoritative"]
            ):
                owner = candidate
                break
            walker = parent
        if owner is not None:
            owner["workers"] += 1
            continue
        if _is_foreign(record, here, here_uid):
            record["foreign"] = True
            others.append(record)
            continue
        (authoritative if record["authoritative"] else others).append(record)
    return authoritative, others


def _load() -> tuple[float, int]:
    """1-minute load average and the CPUs this process may ACTUALLY use.

    ``sched_getaffinity`` rather than ``cpu_count``: the latter counts every
    logical CPU whether or not this process is allowed on it, so under a cpuset
    it overstates capacity and a pinned, saturated box reads as idle.
    """
    try:
        one = float(Path("/proc/loadavg").read_text().split()[0])
    except OSError, IndexError, ValueError:
        one = -1.0
    try:
        cores = len(os.sched_getaffinity(0))
    except AttributeError, OSError:
        cores = os.cpu_count() or 0
    return one, cores


def _last_run(cwd: str) -> str | None:
    """What the previous authoritative run reported, if this repo records it.

    An ABORTED run carries ``passed: 0, failed: 0`` and an ``aborted`` reason.
    Rendering that as "0 passed" states a result the run never produced — the
    first version of this hook did exactly that, and a fabricated number in an
    always-on context block is worse than no line at all.
    """
    report = Path(cwd) / "reports" / "run-tests-latest.json"
    try:
        data = json.loads(report.read_text())
    except OSError, ValueError:
        return None
    aborted = data.get("aborted")
    if aborted:
        return f"ABORTED ({aborted}) — no result was produced"
    passed = data.get("passed")
    failed = data.get("failed")
    if not passed and not failed:
        return None  # nothing was measured; say nothing
    duration = data.get("duration") or data.get("elapsed")
    if passed is None:
        return None
    parts = [f"{passed} passed"]
    if failed:
        parts.append(f"{failed} FAILED")
    if duration:
        parts.append(f"{float(duration):.0f}s")
    age = ""
    try:
        age = f", {int((time.time() - report.stat().st_mtime) / 60)}m ago"
    except OSError:
        pass
    return f"{' · '.join(parts)}{age}"


def _fmt(seconds: int) -> str:
    if seconds < 0:
        return "?"
    return f"{seconds // 60}m{seconds % 60:02d}s" if seconds >= 60 else f"{seconds}s"


def _context_block() -> str:
    authoritative, others = _runs()
    foreign = [r for r in others if r.get("foreign")]
    local_others = [r for r in others if not r.get("foreign")]
    load, cores = _load()
    lines = ["=== TEST RUN CONTEXT ==="]

    if authoritative:
        lines.append("  AUTHORITATIVE RUN IN FLIGHT — do not start competing CPU work:")
        for run in authoritative:
            lines.append(
                f"    pid {run['pid']} · {run['project']} · running {_fmt(run['elapsed'])}"
            )
        lines.append(
            "  Its numbers are only valid on an uncontended box. Starting a "
            "second suite, an ingest or a build now corrupts the measurement "
            "AND can fail the run outright (port/DB/CPU contention)."
        )
        lines.append(
            "  If you have work that must run: wait, or `kill -STOP <pid>` your "
            "own job and `kill -CONT` it after."
        )
    elif local_others:
        lines.append(f"  {len(local_others)} targeted pytest process(es) running:")
        for run in local_others[:4]:
            lines.append(
                f"    pid {run['pid']} · {run['project']} · running {_fmt(run['elapsed'])}"
            )
    elif not foreign:
        lines.append("  no test run in flight")

    if foreign:
        # Reported ALWAYS, alongside whatever this checkout is doing. It is not
        # our run — but a full suite in a CI checkout saturates the same CPUs
        # and binds the same ports, and a box that reads "idle" while one is in
        # flight is the lie this hook exists to prevent. Identified by ROOT or
        # UID, never by the project NAME, which is identical across checkouts.
        lines.append(
            f"  {len(foreign)} authoritative run(s) from ANOTHER checkout or "
            "user — not yours, but competing for this box:"
        )
        for run in foreign[:4]:
            # A CI job's root is unreadable across users, so fall back to the
            # uid: "pid 1915816 · ?" named nothing and was the line that let a
            # foreign run pass for ours.
            whose = run["root"] or f"uid {run['uid'] or '?'}"
            lines.append(
                f"    pid {run['pid']} · {whose} · running {_fmt(run['elapsed'])}"
            )

    if cores:
        # Logical CPUs include hyperthreads, which do NOT add a full core of
        # throughput, so parity with the core count already means the box is
        # oversubscribed for CPU-bound work — not "ok", which is what this
        # line said at load 9.42 while a suite was being starved.
        if load >= cores:
            state = "SATURATED"
        elif load >= cores * 0.6:
            state = "BUSY"
        else:
            state = "idle"
        lines.append(f"  load {load:.2f} on {cores} logical CPUs — {state}")

    last = _last_run(os.getcwd())
    if last:
        lines.append(f"  last authoritative run: {last}")

    lines.append(
        "  Standing rules: cap interactive runs at 5 min (`timeout 300`), run "
        "targeted modules not the full tree, observe live "
        "(`pytest-live` / `observe_pytest.py`) — never fire a batched run and "
        "wait blind."
    )
    lines.append("=== END TEST RUN CONTEXT ===")
    return "\n".join(lines)


# The deliberate way past this gate. A guard with no override is a trap: some
# work genuinely does not compete (a three-test module, a run pinned to one
# core), and with no way to say so the only remaining move is to disable the
# hook — which is how guards die. Requiring the prefix keeps the decision
# explicit and greppable in the transcript instead of silent.
_OVERRIDE = "ORCH_ALLOW_CONTENDING=1"

#: Commands that only LOOK at things. A pipeline built purely from these can
#: mention run_tests.py or ship.sh freely — as a tail'd log name, a grep
#: pattern, a pgrep argument — without BEING one. The heavy matcher reads
#: argument text as intent, and its first live false positive (2026-08-14)
#: denied `tail ship.log; pgrep -f 'deploy/ship.sh'` — an OBSERVATION of the
#: run this hook protects. A guard that blocks watching the very work it
#: guards trains its user to override reflexively, which is how the override
#: stops meaning anything.
_OBSERVATION_CMDS = frozenset(
    {
        "tail",
        "head",
        "grep",
        "pgrep",
        "cat",
        "wc",
        "ls",
        "stat",
        "echo",
        "sleep",
        # `ps -o etime= -p <pid>` — asking how long the guarded run has been
        # going is observation by definition, and it was part of the polling
        # this guard denied on 2026-08-19.
        "ps",
        "uptime",
        # The deployed observer itself. `observe-pytest` matches _HEAVY_RE via
        # \bpytest\b -- the hyphen is a word boundary -- so the guard denied
        # `observe-pytest list` WHILE a run was in flight (measured
        # 2026-08-19), which is the only time that command has any purpose. It
        # reads the session-registry JSON and nothing else: no CPU, no ports,
        # no database. This is the second instance of the false positive the
        # comment above describes, and the same conclusion applies -- the
        # standing rule is to observe a run rather than wait blind, so the
        # guard must never be what stops it.
        "observe-pytest",
        "observe_pytest.py",
        # Read-only text/path tools of the same class as grep/head/tail/cat/wc
        # above: they transform bytes that are already on disk and start no
        # work of their own. `sed` is the load-bearing one — CLAUDE.md mandates
        # ranged reads (`sed -n '400,455p' file`) over whole-file reads, and
        # leaving it out denied that exact command for naming a heavy FILE
        # while a run was in flight (measured 2026-08-21, fifth instance of the
        # family). Omitting them made the guard forbid the reading discipline
        # the project requires.
        "sed",
        "awk",
        "cut",
        "tr",
        "sort",
        "uniq",
        "diff",
        "jq",
        "basename",
        "dirname",
        "realpath",
        "readlink",
        "printf",
        "date",
        "column",
    }
)


#: Shell syntax that can begin a segment. These are KEYWORDS, not commands, so
#: the command word is whatever follows them — exactly as with the VAR=value
#: prefixes below. Measured 2026-08-19: `until ! pgrep -f 'deploy/ship.sh';
#: do sleep 5; done` was denied because the first word was `until` and the
#: `do sleep 5` segment began with `do`. That is the sanctioned way to wait for
#: a run instead of waiting blind — the platform conventions REQUIRE it — so
#: the guard was denying the very pattern it exists to push people toward.
_SHELL_KEYWORDS = frozenset(
    {"!", "until", "while", "do", "done", "if", "then", "elif", "else", "fi", "time"}
)


def _ampersand_is_redirection(buf: list[str], command: str, index: int) -> bool:
    """True when the ``&`` at *index* belongs to a REDIRECTION, not sequencing.

    ``2>&1`` is one token, not two commands. Splitting on its ``&`` left a
    fragment beginning ``1``, which is not an observation tool, so the whole
    command failed closed. Measured 2026-08-21: ``observe-pytest list 2>&1 |
    head`` was denied WHILE a run was in flight — the fourth instance of the
    false-positive family the module docstring above describes, and the one
    that denies the standing "observe, never wait blind" rule outright.

    Two shapes, both one token: ``N>&M`` / ``>&M`` / ``<&M`` (the char before
    is a redirection operator) and ``&>file`` / ``&>>file`` (the char after is
    ``>``). Everything else — ``&&`` and a trailing background ``&`` — is
    sequencing and still splits.
    """
    if "".join(buf).rstrip().endswith((">", "<")):
        return True
    return command[index + 1 : index + 2] == ">"


def _segments(command: str) -> list[str]:
    """Split on shell sequencing that is OUTSIDE quotes.

    A plain ``re.split(r"[;|&\\n]+", ...)`` also cuts inside quoted arguments.
    Measured 2026-08-19: ``pgrep -af 'ship.sh|run_tests'`` split into two, the
    second starting with ``run_tests'``, which is not an observation tool — so
    observing the run was denied because the PATTERN NAMING IT contained a
    pipe. Third instance of this false-positive family, and the same lesson:
    the matcher reads argument text as intent.

    Backslash escapes are not modelled. An unterminated quote leaves the rest
    of the string in one segment, which can only ever make the guard stricter.
    """
    out: list[str] = []
    buf: list[str] = []
    quote = ""
    for index, ch in enumerate(command):
        if quote:
            buf.append(ch)
            if ch == quote:
                quote = ""
        elif ch in "'\"":
            quote = ch
            buf.append(ch)
        elif ch == "&" and _ampersand_is_redirection(buf, command, index):
            buf.append(ch)
        elif ch in ";|&\n":
            out.append("".join(buf))
            buf = []
        else:
            buf.append(ch)
    out.append("".join(buf))
    return out


def _is_observation(command: str) -> bool:
    """True when EVERY segment of *command* starts with an observation tool.

    Segments split on shell sequencing/pipes outside quotes; leading VAR=value
    assignments and shell keywords are skipped the way a shell skips them. One
    non-observation segment makes the whole command subject to the heavy
    matcher again — fail closed.
    """
    seen_any = False
    for seg in _segments(command):
        words = seg.strip().split()
        while words and (
            words[0] in _SHELL_KEYWORDS
            or re.fullmatch(r"[A-Za-z_][A-Za-z_0-9]*=\S*", words[0])
        ):
            words = words[1:]
        if not words:
            continue
        seen_any = True
        if Path(words[0]).name not in _OBSERVATION_CMDS:
            return False
    return seen_any


def _pretooluse(payload: dict) -> int:
    """Deny a heavy command while an authoritative run is in flight."""
    command = (payload.get("tool_input") or {}).get("command") or ""
    if _OVERRIDE in command:
        return 0
    if not _HEAVY_RE.search(command):
        return 0
    if _is_observation(command):
        return 0

    authoritative, _ = _runs()
    # A run's own internal invocations are children of the run; only refuse when
    # the caller is a shell the agent is driving, which is what this hook sees.
    if not authoritative:
        return 0

    run = authoritative[0]
    print(
        json.dumps(
            {
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": (
                        f"An AUTHORITATIVE test run is in flight: pid {run['pid']} "
                        f"({run['project']}), running {_fmt(run['elapsed'])}.\n"
                        "This command is CPU-heavy and would contend with it. That "
                        "has already cost two measurements and one failed ship "
                        "(2026-08-13/14): a competing suite made a 360.72s-vs-369.75s "
                        "comparison meaningless, and an oversubscribed box fails "
                        "runs on port and timing contention that is NOT a real "
                        "defect.\n"
                        "Wait for the run to finish, or suspend your job "
                        "(`kill -STOP <pid>` / `kill -CONT <pid>`).\n"
                        f"If this genuinely does not compete — a targeted module, "
                        f"a pinned single-core job — prefix it with {_OVERRIDE} "
                        "and state why."
                    ),
                }
            }
        )
    )
    return 0


def main() -> int:
    try:
        payload = json.load(sys.stdin) if not sys.stdin.isatty() else {}
    except ValueError, OSError:
        payload = {}

    event = payload.get("hook_event_name") or ""

    if event == "PreToolUse":
        return _pretooluse(payload)

    block = _context_block()
    # Only speak when there is something to say on a prompt; a session start
    # always reports, so the baseline is established once per session.
    if event == "UserPromptSubmit" and "no test run in flight" in block:
        return 0
    print(block)
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except Exception as exc:  # never break a session over context
        print(f"[test_run_context] skipped: {exc}", file=sys.stderr)
        sys.exit(0)

Related

• Claude hooks index
• Claude skills index
• Claude commands index
• Rulings
• Orchestrata Ultimate workflow
• Orchestrata Ultimate gates
• Orchestrata Ultimate briefs
• Orchestrata Ultimate
• Codetools