scripts.check_kimi_agent_level

Reference › scripts › check_kimi_agent_level

The kimi agent-level gate, the mechanised form of RUL-018. It scans the tracked files of a repository for places where a kimi agent is started, and reports spawn lines that break the rule on agent level. health_check invokes it through main, and tests/scripts/test_check_kimi_agent_level.py imports it.

Invocation: main takes --repo-root (default the current directory) and --json. It returns 0 when the scan is clean or skipped, and 1 on any violation.

run_check takes the file list from tracked_corpus, which asks git for the tracked files and raises CorpusError if the corpus cannot be read, then calls _scan_one for each file. scan_text works line by line: it keeps shell-style assignments (_strip_quotes, _resolve, _bindings_on_line), so a nickname held in a variable still resolves, recognizes spawn lines (_is_spawn), works out the nickname being started (_spawn_nick) and its agent (_agent_of), and produces a Violation for either of two defects built by _d1 and _d2. The report is printed by _print_text_report.

At a glance

Class diagram

Dependency diagram

No internal imports and no importers.

API

Classes

CorpusError

Bases: Exception · Line 177

git ls-files could not be run — not a git repository, or git is unavailable.

Violation (dataclass)

Decorators: dataclass · Line 183

One RUL-018 finding: which detector fired, where, and the foreign nick that made it fire.

Functions

Source

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

#!/usr/bin/env python3
"""Kimi agent-level gate — the mechanised form of RUL-018.

Ruling text, verbatim (this is the string ``scripts/rulings_lint.py``'s
``gate <script> <rule-id>`` grammar resolves against):

RUL-018 — "Kimi runs at the AGENT level, not the project level." The persistent kimi IRC presence must be the single systemd USER unit (kimi-irc-bot.service, nick `kimi`); a project-level kimi listener must never be started from this repo, and nicks are agent-specific (codewhale is the codewhale agent's identity, never assumed from a kimi session). Currently declared `none (judgment)` and therefore prose-only. Investigation proves it IS mechanizable as a check on repo contents, and that this repo ships TWO live violations right now: (1) `.kimi/hooks/session-start.sh` — tracked, executable, and live-wired as kimi's SessionStart hook via ~/.kimi/config.toml:92-96 — sets `IRC_NICK=kimi_agent` (line 124) and spawns `setsid nohup env IRC_NICK="$IRC_NICK" python3 "$PROJECT/scripts/_irc_watchdog.py" ... &` (line 136), i.e. a project-level kimi IRC listener started from this repo, exactly what SESSION_START.md forbids ("Do NOT start a project-level kimi listener from this repo — one agent-level daemon per agent"); (2) `scripts/_irc_listen.py:37` defaults to another agent's identity: `os.environ.get("IRC_NICK", "codewhale")`, though SESSION_START.md states this pair is claude's listener and "runs with IRC_NICK=claude". Runtime today is compliant (kimi-irc-bot.service active as `kimi`; repo listener PID 14191 up with IRC_NICK=claude), so violation (1) is LATENT — the hook only takes its spawn branch when claude's listener is absent. Shipped repo content is non-compliant even though the running fleet is fine; that gap is precisely what the gate closes. SEPARATE DEFECT SURFACED (not part of this mechanism): SESSION_START.md, the canonical IRC-discipline document, contained unresolved merge-conflict markers at lines 31/66/67 and 73/100/101 (`<<<<<<< Updated upstream` ... `>>>>>>> Stashed changes`). RE-CHECKED 2026-09-21: SESSION_START.md carries no conflict markers — the corruption was resolved in a later commit; this defect is closed. HONEST BLOCKER: this gate goes RED on today's tree; landing it requires fixing both violations in the same change (drop section 6 of the .kimi hook, reporting `systemctl --user is-active kimi-irc-bot.service` instead of spawning; flip _irc_listen.py's default to `claude`). Wiring it green by weakening it would manufacture the exact false confidence the task is trying to kill.

WHAT IS GRADED — repo CONTENTS, never runtime state. The corpus is
``git ls-files`` filtered to executable/config suffixes (.py .sh .bash .toml
.yaml .yml .service .json). Markdown is excluded ENTIRELY: SESSION_START.md,
the RUL-018 wiki page, ``plans/kimi-*.md`` and KIMI-PROCESS.md all quote
these nicks in prose and must never trip the gate. Untracked, gitignored
runtime state (``.kimi/irc-last-check``, ``.kimi/irc-log-offset``) is out of
scope — local runtime state is legitimately local. Whether the agent-level
daemon is up is separately observable via ``systemctl --user status
kimi-irc-bot.service``; this gate deliberately never shells out to it, so a
laptop with the unit stopped is not graded red for a content question.

Two detectors:

    D1  FOREIGN-IDENTITY BINDING. A line that BINDS the env var ``IRC_NICK``
        (word-anchored ``\\bIRC_NICK\\b``) to a string literal outside the
        repo-owned allowlist :data:`REPO_OWNED_NICKS`. Recognised binding
        forms: shell/env ``IRC_NICK=<lit>``, ``env IRC_NICK="<lit>"``,
        ``export IRC_NICK=<lit>``; systemd ``Environment=IRC_NICK=<lit>``;
        python ``os.environ.get("IRC_NICK", "<lit>")`` and the env-dict
        ``{"IRC_NICK": "<lit>"}``. Shell variable indirection
        (``IRC_NICK="$IRC_NICK"``) is resolved ONE hop within the same file
        when the right-hand side was assigned a literal earlier in that
        file; unresolvable indirection is SKIPPED rather than guessed
        (a false negative is preferred to a false positive).

    D2  PROJECT-LEVEL AGENT DAEMON SPAWN. A tracked file that BOTH (a)
        spawns ``scripts/_irc_listen.py`` or ``scripts/_irc_watchdog.py`` as
        a detached/background process (``setsid``, ``nohup``, a trailing
        ``&``, ``subprocess.Popen``, ``systemd-run``) AND (b) whose
        effective ``IRC_NICK`` for that spawn resolves per D1 to a
        non-repo-owned nick. Reported as the headline violation.

Both conditions are required for D2 — that two-factor rule is what keeps the
watchdog itself (``scripts/_irc_watchdog.py`` legitimately ``Popen``s the
listener, binds no nick, and therefore inherits claude's) and every prose
mention of the script names in tracked ``.json`` planning artifacts silent.

DELIBERATELY NOT FLAGGED (each verified against a real line in the live
tree; the gate must stay silent on all of these):

  * ``core/infra_config.py``, ``drivers/browser_providers.py``,
    ``drivers/driver_factory.py`` and their tests use ``"kimi"`` as the
    LLM/BROWSER PROVIDER name (Moonshot) — unrelated to IRC identity. No
    ``IRC_NICK`` token, so nothing matches. A naive grep for the nick would
    flood on these; keying on the BINDING SITE instead of the nick token is
    what makes this gate real rather than decorative.
  * ``scripts/_irc_conversation.py``'s ``PLATFORM_SLUG`` and
    ``PLATFORM_FOR_NICK`` map — the READ-SIDE conversation-capture
    attribution table. Knowing a nick in order to LABEL a captured turn is
    not assuming that identity; its dict keys are nicks, not ``IRC_NICK``.
  * ``scripts/swarm_supervisor.py``'s ``DEFAULT_IRC_NICK`` — silent twice
    over: ``\\bIRC_NICK`` cannot match inside ``DEFAULT_IRC_NICK`` (the
    preceding ``_`` is a word character), and ``swarm-supervisor`` is an
    allowlisted repo-owned role nick anyway.
  * ``scripts/_irc_send.py``'s ``os.environ.get("IRC_NICK", "claude")`` —
    repo-owned identity.
  * ``pgrep -f "_irc_listen.py"`` health probes and the
    ``sed -n 's/^IRC_NICK=//p'`` environ read in the kimi hook — neither is
    a spawn, and the ``sed`` right-hand side is not a nick-shaped literal.

No blanket ``tests/`` exemption is granted: a test that literally hardcoded
``IRC_NICK`` to a foreign agent and spawned the watchdog would collide with
that live agent's bot on the shared IRC channel and SHOULD fail. The only
exemptions are :data:`SELF_EXEMPT` — this gate and its own test module,
which must quote the violating lines verbatim in order to specify and to
prove the detector. A detector that flags its own specification is
degenerate.

Exit 0 clean (or skipped, with a printed reason — not a git repository or
git unavailable). Exit 1 on any violation, one line per finding.

Usage:
    python scripts/check_kimi_agent_level.py [--repo-root PATH] [--json]
"""

from __future__ import annotations

import argparse
import json
import re
import subprocess
import sys
from dataclasses import asdict, dataclass
from pathlib import Path

#: Nicks this repository owns and may legitimately bind ``IRC_NICK`` to.
#: ``claude`` is this checkout's listener identity (SESSION_START.md),
#: ``claude-out`` its outbound sender, ``swarm-supervisor`` the role nick of
#: ``scripts/swarm_supervisor.py``. Every other fleet nick belongs to another
#: agent: kimi's presence is the agent-level ``kimi-irc-bot.service``, and
#: codewhale is the codewhale agent's identity.
REPO_OWNED_NICKS = frozenset({"claude", "claude-out", "swarm-supervisor"})

#: Corpus filter. Markdown is absent ON PURPOSE (see module docstring).
CORPUS_SUFFIXES = frozenset(
    {".py", ".sh", ".bash", ".toml", ".yaml", ".yml", ".service", ".json"}
)

#: Generated DATA dumps the gate has no business regexing: AST
#: serializations, RTM mappings, and run reports are megabytes of
#: machine-written records — a shell/python IRC_NICK binding cannot live in
#: them, and every line scanned is pure cost (measured 2026-08-11: ~6M of
#: the corpus's lines were these files, ~13s of a full run_check; the
#: detector's _resolved_bindings/_spawn_nick ran 7.6M times). Real .json
#: config (settings, manifests) stays in the corpus.
CORPUS_EXCLUDED_PREFIXES = (
    "plans/ast-",
    "plans/reference-edges",
    "rtm/",
    "reports/",
)

#: The only exempt paths: this gate and its test module quote the violating
#: lines verbatim to specify and to prove the detector.
SELF_EXEMPT = frozenset(
    {
        "scripts/check_kimi_agent_level.py",
        "tests/scripts/test_check_kimi_agent_level.py",
        # context.json is the session CLI's work JOURNAL — narrative prose that
        # happens to be serialized as JSON. Its markdown twin, context.md, is
        # written from the same data and is already excluded wholesale by the
        # markdown rule; scanning one and not the other is an inconsistency in
        # this gate's own scope, not a real difference in risk. It binds
        # nothing and executes nothing. Measured 2026-08-05: logging the
        # RUL-018 FIX ("this repo bound IRC_NICK=kimi_agent") into the journal
        # made the gate flag the very repair it was reporting.
        "context.json",
    }
)

# ``\bIRC_NICK=`` — the word boundary is load-bearing: it makes
# ``DEFAULT_IRC_NICK`` unmatchable (the preceding ``_`` is a word char).
# The value alternation is ordered quoted / variable-reference / bare
# nick-shaped literal, and a bare value MUST start with a letter so that
# ``sed -n 's/^IRC_NICK=//p'`` yields no match at all.
_SHELL_BIND_RE = re.compile(
    r"\bIRC_NICK="
    r"(?P<val>\"[^\"\n]*\"|'[^'\n]*'"
    r"|\$\{?[A-Za-z_][A-Za-z0-9_]*\}?"
    r"|[A-Za-z][A-Za-z0-9_.-]*)"
)

# ``os.environ.get("IRC_NICK", "<lit>")`` and the env-dict
# ``{"IRC_NICK": "<lit>"}``. The quote immediately before ``IRC_NICK`` is
# what keeps ``"DEFAULT_IRC_NICK"`` out.
_PY_KV_RE = re.compile(
    r"[\"']IRC_NICK[\"']\s*[,:]\s*[\"'](?P<val>[^\"'\n]*)[\"']",
)

# A whole-line shell assignment with a LITERAL right-hand side. Only literal
# assignments enter the resolution table, which is what bounds indirection
# to exactly one hop.
_SHELL_ASSIGN_RE = re.compile(
    r"^\s*(?:export\s+)?(?P<name>[A-Za-z_][A-Za-z0-9_]*)="
    r"(?P<val>\"[^\"\n$]*\"|'[^'\n]*'|[A-Za-z][A-Za-z0-9_.-]*)\s*$"
)

_VAR_REF_RE = re.compile(r"^\$\{?(?P<name>[A-Za-z_][A-Za-z0-9_]*)\}?$")

#: This repo's IRC daemon pair — the processes a project-level listener
#: would be started from.
_LISTENER_RE = re.compile(r"_irc_(?:listen|watchdog)\.py")

_SPAWN_TOKEN_RE = re.compile(
    r"\b(?:setsid|nohup|systemd-run)\b|subprocess\.Popen|\bPopen\s*\("
)


class CorpusError(Exception):
    """``git ls-files`` could not be run — not a git repository, or git is
    unavailable. The caller turns this into a SKIP, never a crash."""


@dataclass
class Violation:
    """One RUL-018 finding: which detector fired, where, and the foreign nick
    that made it fire."""

    detector: str
    path: str
    line: int
    nick: str
    message: str


def tracked_corpus(repo_root: Path) -> list[str]:
    """Repo-relative paths of every tracked file this gate grades:
    ``git ls-files`` narrowed to :data:`CORPUS_SUFFIXES`, minus
    :data:`SELF_EXEMPT`. Raises :class:`CorpusError` when git cannot answer."""
    try:
        result = subprocess.run(
            ["git", "ls-files", "-z"],
            cwd=repo_root,
            capture_output=True,
            text=True,
            timeout=30,
        )
    except (OSError, subprocess.TimeoutExpired) as exc:
        raise CorpusError(str(exc)) from exc
    if result.returncode != 0:
        raise CorpusError(result.stderr.strip() or "git ls-files failed")
    paths = [p for p in result.stdout.split("\0") if p]
    return [
        p
        for p in sorted(paths)
        if Path(p).suffix in CORPUS_SUFFIXES
        and p not in SELF_EXEMPT
        and not p.startswith(CORPUS_EXCLUDED_PREFIXES)
    ]


def _strip_quotes(value: str) -> str:
    if len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'":
        return value[1:-1]
    return value


def _resolve(value: str, assignments: dict[str, str]) -> str | None:
    """A binding's right-hand side -> the literal nick it means, or ``None``
    when the gate cannot know. One hop of shell indirection is resolved from
    *assignments* (literal assignments seen EARLIER in the same file);
    anything else — a command substitution, a positional parameter, an
    unassigned variable, an empty string — is deliberately unresolvable and
    therefore skipped rather than guessed."""
    literal = _strip_quotes(value).strip()
    if not literal:
        return None
    ref = _VAR_REF_RE.match(literal)
    if ref:
        return assignments.get(ref.group("name"))
    if literal.startswith("$"):
        return None
    return literal


def _bindings_on_line(line: str) -> list[str]:
    """Every raw ``IRC_NICK`` binding right-hand side on one line."""
    values = [m.group("val") for m in _SHELL_BIND_RE.finditer(line)]
    values.extend(m.group("val") for m in _PY_KV_RE.finditer(line))
    return values


def _is_spawn(line: str) -> bool:
    """Does this line launch a long-lived detached/background process?"""
    stripped = line.strip()
    if _SPAWN_TOKEN_RE.search(line):
        return True
    return stripped.endswith("&") and not stripped.endswith("&&")


def _agent_of(nick: str) -> str:
    """Display name of the agent a nick belongs to: ``kimi_agent`` -> ``kimi``."""
    for suffix in ("_agent", "-agent"):
        if nick.endswith(suffix) and len(nick) > len(suffix):
            return nick[: -len(suffix)]
    return nick


def _resolved_bindings(line: str, assignments: dict[str, str]) -> list[str]:
    """Every ``IRC_NICK`` binding on one line, resolved to a literal nick.
    Unresolvable right-hand sides are dropped, never guessed."""
    resolved = []
    for raw in _bindings_on_line(line):
        nick = _resolve(raw, assignments)
        if nick is not None:
            resolved.append(nick)
    return resolved


def _spawn_nick(line: str, effective: str | None) -> str | None:
    """D2's two-factor test: the foreign nick this line spawns the repo's IRC
    daemon under, or ``None`` when either factor is absent."""
    if not (_LISTENER_RE.search(line) and _is_spawn(line)):
        return None
    if effective is None or effective in REPO_OWNED_NICKS:
        return None
    return effective


def _d2(rel_path: str, lineno: int, nick: str) -> Violation:
    agent = _agent_of(nick)
    return Violation(
        "D2",
        rel_path,
        lineno,
        nick,
        f"project-level {agent} listener started from this repo "
        f"({rel_path}:{lineno} spawns the repo IRC daemon as "
        f"IRC_NICK={nick}). {agent.capitalize()}'s IRC presence is the "
        f"agent-level unit {agent}-irc-bot.service.",
    )


def _d1(rel_path: str, lineno: int, nick: str) -> Violation:
    return Violation(
        "D1",
        rel_path,
        lineno,
        nick,
        f"IRC_NICK bound to foreign agent identity {nick!r}; nicks are "
        f"agent-specific and this repo owns only {sorted(REPO_OWNED_NICKS)} "
        f"(expected 'claude').",
    )


def scan_text(rel_path: str, text: str) -> list[Violation]:
    """Run D1 + D2 over one file's contents.

    Lines are walked in order so that shell indirection resolves against
    assignments made ABOVE the binding, and so that a spawn inherits the
    file's most recently resolved ``IRC_NICK`` when the spawn line carries
    no binding of its own."""
    violations: list[Violation] = []
    assignments: dict[str, str] = {}
    current_nick: str | None = None

    for lineno, line in enumerate(text.splitlines(), start=1):
        resolved_here = _resolved_bindings(line, assignments)
        if resolved_here:
            current_nick = resolved_here[-1]
        effective = resolved_here[-1] if resolved_here else current_nick

        spawn = _spawn_nick(line, effective)
        if spawn is not None:
            # D2 is the headline finding for this line; a D1 on the same line
            # would be a duplicate report of the same defect.
            violations.append(_d2(rel_path, lineno, spawn))
        else:
            violations.extend(
                _d1(rel_path, lineno, nick)
                for nick in resolved_here
                if nick not in REPO_OWNED_NICKS
            )

        # Update the resolution table AFTER the line is graded: a literal
        # assignment can only feed indirection on LATER lines.
        assign = _SHELL_ASSIGN_RE.match(line)
        if assign:
            assignments[assign.group("name")] = _strip_quotes(assign.group("val"))

    return violations


def run_check(repo_root: Path) -> tuple[int, dict]:
    """Injection seam: tests pass a throwaway tmp_path git repo — this never
    reaches outside *repo_root*."""
    try:
        corpus = tracked_corpus(repo_root)
    except CorpusError as exc:
        return 0, {
            "status": "skipped",
            "reason": f"could not list tracked files at {repo_root}: {exc}",
        }

    def _scan_one(rel: str) -> list[Violation]:
        path = repo_root / rel
        try:
            text = path.read_text(encoding="utf-8", errors="replace")
        except OSError:
            return []
        return scan_text(rel, text)

    # Per-file scans are independent pure reads — thread them (free-threaded
    # build; the full-corpus scan was 21s serial, measured 2026-08-11).
    # executor.map preserves corpus order, so reports stay deterministic.
    if len(corpus) > 1:
        import os as _os
        from concurrent.futures import ThreadPoolExecutor

        with ThreadPoolExecutor(max_workers=_os.cpu_count() or 4) as executor:
            per_file = list(executor.map(_scan_one, corpus))
    else:
        per_file = [_scan_one(rel) for rel in corpus]
    violations: list[Violation] = [
        v for file_violations in per_file for v in file_violations
    ]

    report = {
        "status": "pass" if not violations else "fail",
        "repo_root": str(repo_root),
        "files_scanned": len(corpus),
        "violations": [asdict(v) for v in violations],
    }
    return (0 if not violations else 1), report


def _print_text_report(report: dict) -> None:
    if report["status"] == "skipped":
        print(f"[kimi-agent-level] SKIP — {report['reason']}", flush=True)
        return
    if report["status"] == "pass":
        print(
            f"[kimi-agent-level] OK — RUL-018 satisfied across "
            f"{report['files_scanned']} tracked file(s)",
            flush=True,
        )
        return
    print(
        f"[kimi-agent-level] FAIL — {len(report['violations'])} violation(s):",
        flush=True,
    )
    for v in report["violations"]:
        print(
            f"  RUL-018 [{v['detector']}] {v['path']}:{v['line']}: {v['message']}",
            flush=True,
        )


def main(argv: list[str] | None = None) -> int:
    """CLI entry point — the seam ``scripts/health_check.py`` invokes.
    Returns the process exit code: 0 clean or skipped, 1 on any violation."""
    parser = argparse.ArgumentParser(description="RUL-018 kimi agent-level gate")
    parser.add_argument(
        "--repo-root", default=None, help="Directory to gate (default: cwd)"
    )
    parser.add_argument("--json", action="store_true", help="Machine-readable output")
    args = parser.parse_args(argv)

    repo_root = Path(args.repo_root).resolve() if args.repo_root else Path.cwd()
    exit_code, report = run_check(repo_root)

    if args.json:
        print(json.dumps(report, indent=2, default=str))
    else:
        _print_text_report(report)
    return exit_code


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