scripts.verify_package

Reference › scripts › verify_package

scripts/verify_package.py is a trust-but-sample verifier for a completed change: it runs a short, fail-fast list of checks over the files a change touched and prints a compact PASS or FAIL table. A human or an orchestrating session runs it once a change is complete. It has no internal imports; it calls other tools through _run, with a default timeout of 90 seconds.

Options: --paths (required; the files in scope), --symbols-gone (symbols that must no longer exist; check_symbols_gone uses the AST toolkit's refs command, and parse_refs_summary reads its definitions, imports and uses counts), --grep-gone (text patterns that must no longer appear; check_grep_gone), --touched (files for check_ruff), --import (modules that must still import; check_imports) and --full (test targets, each run in its own pytest subprocess in parallel by check_full). --slot-base (default 20) is the first ORCHESTRATA_WORKER_SLOT and Redis database index for the --full targets, kept clear of interactive development slots. run_checks runs the configured steps in order and stops at the first failure, and main exits 0 only when every step passed.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/verify_package.py

#!/usr/bin/env python3
"""verify_package.py — trust-but-sample verifier for a completed change.

A retro of swarm-orchestration cost centers found orchestrators re-deriving
"is this package actually done" by hand: re-grepping for a symbol that was
supposed to be removed, eyeballing ruff output, re-importing modules by
hand. This composes those checks into one fail-fast CLI with a compact
PASS/FAIL table, so an orchestrator can spot-check a builder's report in one
command instead of five.

Steps, run IN ORDER, fail-fast (a failing step stops the rest — later steps
are reported ``SKIP``). Target total runtime: under 60s WITHOUT ``--full``
(the caller is responsible for scoping ``--paths`` narrowly; this tool adds
generous but bounded per-step timeouts rather than a wall-clock budget of
its own):

1. ``--symbols-gone NAME...``  — each name must have ZERO refs (defs +
   imports + uses) under ``--paths`` (``serialize_codebase_ast.py refs``).
2. ``--grep-gone PATTERN...``  — each extended-regex pattern must have ZERO
   hits under ``--paths`` (``grep -rnE``).
3. ruff check on ``--touched`` (falls back to ``--paths`` when omitted) —
   always run.
4. ``--import MODULE...``      — each module must ``import`` cleanly, with
   every ``--paths`` entry added to ``PYTHONPATH`` for the check.
5. ``--full TARGET...``        — (only with ``--full``) each pytest target
   run in its OWN subprocess, in parallel, with the live-observer plugin.

Exit code is non-zero if any attempted step failed.

CLI:
    python scripts/verify_package.py --paths src tests \\
        --symbols-gone OldThing --grep-gone 'legacy_flag' \\
        --import orchestrata_ultimate.knowledge_graph.graph_client
"""

from __future__ import annotations

import argparse
import os
import queue
import re
import subprocess
import sys
from collections.abc import Callable, Sequence
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent
AST_TOOL = REPO_ROOT / "scripts" / "serialize_codebase_ast.py"
LIVE_OBSERVER_PLUGIN = "orchestrata_ultimate.core.threading.pytest_live_observer"

#: The `refs` subcommand's trailing summary line, e.g.
#: ``'GraphClient': 1 def, 13 import, 16 use(s).``
_REFS_SUMMARY_RE = re.compile(
    r":\s*(\d+)\s+def,\s*(\d+)\s+import,\s*(\d+)\s+use\(s\)\."
)

Row = tuple[str, str, str]  # (step name, "PASS"/"FAIL"/"SKIP", detail)


# ---------------------------------------------------------------------------
# Step 1 — symbols-gone (AST refs)
# ---------------------------------------------------------------------------


def _run(
    cmd: list[str],
    *,
    cwd: Path = REPO_ROOT,
    env: dict[str, str] | None = None,
    timeout: float = 90.0,
) -> subprocess.CompletedProcess[str]:
    try:
        return subprocess.run(
            cmd, cwd=str(cwd), env=env, capture_output=True, text=True, timeout=timeout
        )
    except subprocess.TimeoutExpired as exc:
        # `text=True` guarantees str at runtime, but TimeoutExpired.stdout is
        # typed loosely (bytes | str | None) -- normalize explicitly so the
        # return type actually matches the declared CompletedProcess[str].
        timeout_stdout = exc.stdout
        if isinstance(timeout_stdout, bytes):
            timeout_stdout = timeout_stdout.decode(errors="replace")
        return subprocess.CompletedProcess(
            cmd,
            returncode=124,
            stdout=timeout_stdout or "",
            stderr=f"timed out after {timeout}s",
        )


def parse_refs_summary(output: str) -> tuple[int, int, int]:
    """``(defs, imports, uses)`` from a ``refs`` invocation's summary line.

    ``(0, 0, 0)`` when the summary line is missing/unparseable — treated as
    "found nothing", the same verdict an actually-empty result gives, so a
    malformed AST-tool invocation reads as PASS-with-a-suspicious-detail
    rather than a crash (the caller still sees the raw output on FAIL).
    """
    m = _REFS_SUMMARY_RE.search(output)
    if not m:
        return (0, 0, 0)
    return (int(m.group(1)), int(m.group(2)), int(m.group(3)))


def check_symbols_gone(
    names: Sequence[str], paths: Sequence[str], repo_root: Path = REPO_ROOT
) -> tuple[bool, str]:
    source = ",".join(str(p) for p in paths)
    remaining: list[str] = []
    for name in names:
        cmd = [
            sys.executable,
            str(AST_TOOL),
            "refs",
            name,
            "--source",
            source,
        ]
        # Deliberately NOT --no-daemon: `query_ast_daemon` degrades to the
        # identical local computation on any miss (out-of-scope --source,
        # no resident daemon, ...), but when scripts/ast_daemon.py IS up
        # (the SessionStart hook keeps one warm for this repo) this turns a
        # ~11s cold reparse-plus-133MB-warmcache-load into a sub-second
        # socket round trip — real savings toward the "under 60s" budget
        # this tool documents for itself above.
        result = _run(cmd, cwd=repo_root)
        # An errored or timed-out `refs` invocation must never read as
        # "0 refs" (false confidence): mirror check_grep_gone's returncode
        # gate, and separately require the trailing summary line to be
        # present — its absence means the invocation didn't complete the
        # way we expect even when the process happened to exit 0/1.
        if result.returncode not in (0, 1):
            remaining.append(
                f"{name!r}: refs invocation error rc={result.returncode}: "
                f"{result.stderr.strip()[:200]}"
            )
            continue
        if not _REFS_SUMMARY_RE.search(result.stdout):
            remaining.append(
                f"{name!r}: refs invocation produced no summary line "
                f"(stdout: {result.stdout.strip()[:200]!r})"
            )
            continue
        defs, imports, uses = parse_refs_summary(result.stdout)
        total = defs + imports + uses
        if total:
            remaining.append(
                f"{name!r}: {total} ref(s) remain ({defs} def, {imports} import, {uses} use)"
            )
    if remaining:
        return False, "; ".join(remaining)
    return True, f"0 refs for {len(names)} symbol(s)"


# ---------------------------------------------------------------------------
# Step 2 — grep-gone
# ---------------------------------------------------------------------------


def check_grep_gone(
    patterns: Sequence[str], paths: Sequence[str], repo_root: Path = REPO_ROOT
) -> tuple[bool, str]:
    problems: list[str] = []
    for pattern in patterns:
        cmd = ["grep", "-rnE", pattern, *[str(p) for p in paths]]
        result = _run(cmd, cwd=repo_root, timeout=60.0)
        if result.returncode == 0:
            lines = result.stdout.strip().splitlines()
            first = lines[0][:200] if lines else ""
            problems.append(f"{pattern!r}: {len(lines)} hit(s) (first: {first})")
        elif result.returncode not in (0, 1):
            problems.append(
                f"{pattern!r}: grep error rc={result.returncode}: "
                f"{result.stderr.strip()[:200]}"
            )
    if problems:
        return False, "; ".join(problems)
    return True, f"0 hits for {len(patterns)} pattern(s)"


# ---------------------------------------------------------------------------
# Step 3 — ruff
# ---------------------------------------------------------------------------


def check_ruff(targets: Sequence[str], repo_root: Path = REPO_ROOT) -> tuple[bool, str]:
    cmd = [sys.executable, "-m", "ruff", "check", *[str(t) for t in targets]]
    result = _run(cmd, cwd=repo_root, timeout=60.0)
    if result.returncode != 0:
        tail = (result.stdout + result.stderr).strip()
        return False, tail[-800:]
    return True, f"clean ({len(targets)} target(s))"


# ---------------------------------------------------------------------------
# Step 4 — import check
# ---------------------------------------------------------------------------


def check_imports(
    modules: Sequence[str], paths: Sequence[str], repo_root: Path = REPO_ROOT
) -> tuple[bool, str]:
    env = os.environ.copy()
    extra = os.pathsep.join(str(Path(p).resolve()) for p in paths)
    existing = env.get("PYTHONPATH", "")
    env["PYTHONPATH"] = f"{extra}{os.pathsep}{existing}" if existing else extra
    failures: list[str] = []
    for module in modules:
        result = _run(
            [sys.executable, "-c", f"import {module}"],
            cwd=repo_root,
            env=env,
            timeout=30.0,
        )
        if result.returncode != 0:
            failures.append(f"{module}: {result.stderr.strip()[-300:]}")
    if failures:
        return False, "; ".join(failures)
    return True, f"{len(modules)} module(s) import cleanly"


# ---------------------------------------------------------------------------
# Step 5 — --full (parallel isolated pytest per target)
# ---------------------------------------------------------------------------


def _run_one_full_target(
    target: str, slot: int, repo_root: Path
) -> tuple[str, bool, str]:
    env = os.environ.copy()
    src_dir = str(repo_root / "src")
    existing = env.get("PYTHONPATH", "")
    env["PYTHONPATH"] = f"{src_dir}{os.pathsep}{existing}" if existing else src_dir
    env["ORCHESTRATA_WORKER_SLOT"] = f"slot{slot}"
    # pytest truncates every ``-rf`` short-summary line to the terminal width
    # (``shutil.get_terminal_size`` honours ``COLUMNS``). With no tty that is
    # 80 columns, and a tmp-path target alone exceeds it, so the assertion
    # message was cut off before it reached ``detail`` (found 2026-10-06 when
    # the promotion suite ran without a tty). Pin a width the summary fits in.
    env["COLUMNS"] = "400"
    # No REDIS_DB override: the suite isolates Redis state by test_id key
    # prefix, not by database index — and the old ``str(slot)`` was invalid
    # for any slot >= 16 (Redis ships 16 databases), which every unbounded
    # enumerate-slot beyond target 15 silently hit.
    cmd = [
        sys.executable,
        "-m",
        "pytest",
        target,
        "-p",
        LIVE_OBSERVER_PLUGIN,
        "-q",
        "--no-header",
        "-p",
        "no:cacheprovider",
    ]
    result = _run(cmd, cwd=repo_root, env=env, timeout=300.0)
    passed = result.returncode == 0
    detail = "" if passed else (result.stdout + result.stderr).strip()[-800:]
    return target, passed, detail


#: Bounded slot pool for ``check_full``: at most this many concurrent
#: targets, recycling slots ``slot_base .. slot_base+_FULL_SLOT_POOL-1``.
#: The old scheme (``slot_base + enumerate index``, one thread per target)
#: was UNBOUNDED: a 37-target run minted slots 20..56, and every child's
#: conftest auto-start idempotently CREATE DATABASE'd its ``orchkg_w<N>`` —
#: 30 stray databases on the test node (found 2026-08-05).
_FULL_SLOT_POOL = 8


def check_full(
    targets: Sequence[str], repo_root: Path = REPO_ROOT, *, slot_base: int = 20
) -> tuple[bool, str]:
    """Run each ``--full`` target in its own pytest subprocess, in parallel,
    on a BOUNDED, recycled slot band (``slot_base .. slot_base+7``) so
    targets never collide on a worker database — and never mint new ones.
    ``slot_base=20`` keeps the band above the batched gate's workers (0..15).
    """
    failures: list[str] = []
    pool_size = max(1, min(len(targets), _FULL_SLOT_POOL))
    slots: queue.Queue[int] = queue.Queue()
    for i in range(pool_size):
        slots.put(slot_base + i)

    def _run_with_recycled_slot(target: str) -> tuple[str, bool, str]:
        slot = slots.get()
        try:
            return _run_one_full_target(target, slot, repo_root)
        finally:
            slots.put(slot)

    with ThreadPoolExecutor(max_workers=pool_size) as pool:
        futures = [pool.submit(_run_with_recycled_slot, t) for t in targets]
        for future in futures:
            target, passed, detail = future.result()
            if not passed:
                failures.append(f"{target}: {detail}")
    if failures:
        return False, "; ".join(failures)
    return True, f"{len(targets)} target(s) passed"


# ---------------------------------------------------------------------------
# Orchestration + table rendering
# ---------------------------------------------------------------------------


def _print_table(rows: list[Row]) -> None:
    if not rows:
        print("(nothing to check)")
        return
    name_w = max(len("STEP"), *(len(r[0]) for r in rows))
    status_w = max(len("STATUS"), *(len(r[1]) for r in rows))
    print(f"{'STEP':<{name_w}}  {'STATUS':<{status_w}}  DETAIL")
    for name, status, detail in rows:
        print(f"{name:<{name_w}}  {status:<{status_w}}  {detail}")


def run_checks(
    *,
    symbols_gone: Sequence[str] = (),
    grep_gone: Sequence[str] = (),
    paths: Sequence[str],
    touched: Sequence[str] = (),
    imports: Sequence[str] = (),
    full: Sequence[str] = (),
    repo_root: Path = REPO_ROOT,
    slot_base: int = 20,
) -> tuple[bool, list[Row]]:
    """Run every configured step in order, fail-fast. Returns ``(all_ok, rows)``."""
    steps: list[tuple[str, Callable[[], tuple[bool, str]]]] = []
    if symbols_gone:
        steps.append(
            ("symbols-gone", lambda: check_symbols_gone(symbols_gone, paths, repo_root))
        )
    if grep_gone:
        steps.append(
            ("grep-gone", lambda: check_grep_gone(grep_gone, paths, repo_root))
        )
    ruff_targets = list(touched) or list(paths)
    # gate-ok: step label ("ruff", check_ruff), not an argv; check_ruff spawns sys.executable -m ruff
    steps.append(("ruff", lambda: check_ruff(ruff_targets, repo_root)))
    if imports:
        steps.append(("import", lambda: check_imports(imports, paths, repo_root)))
    if full:
        steps.append(
            ("full-tests", lambda: check_full(full, repo_root, slot_base=slot_base))
        )

    rows: list[Row] = []
    failed = False
    for name, fn in steps:
        if failed:
            rows.append((name, "SKIP", "not reached — fail-fast"))
            continue
        ok, detail = fn()
        rows.append((name, "PASS" if ok else "FAIL", detail))
        if not ok:
            failed = True
    return not failed, rows


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description=(
            "Trust-but-sample verifier: fail-fast checks (symbols-gone, "
            "grep-gone, ruff, import, optional --full pytest) with a "
            "compact PASS/FAIL table."
        )
    )
    parser.add_argument("--symbols-gone", nargs="+", default=[], dest="symbols_gone")
    parser.add_argument("--grep-gone", nargs="+", default=[], dest="grep_gone")
    parser.add_argument("--paths", nargs="+", required=True)
    parser.add_argument("--touched", nargs="+", default=[])
    parser.add_argument("--import", nargs="+", default=[], dest="import_")
    parser.add_argument("--full", nargs="+", default=[])
    parser.add_argument(
        "--slot-base",
        type=int,
        default=20,
        help="First ORCHESTRATA_WORKER_SLOT/REDIS_DB index for --full targets "
        "(default: 20, to stay clear of interactive dev slots).",
    )
    args = parser.parse_args(argv)

    ok, rows = run_checks(
        symbols_gone=args.symbols_gone,
        grep_gone=args.grep_gone,
        paths=args.paths,
        touched=args.touched,
        imports=args.import_,
        full=args.full,
        slot_base=args.slot_base,
    )
    _print_table(rows)
    return 0 if ok else 1


if __name__ == "__main__":
    sys.exit(main())