FabledScribe/src/scribe/services/system_rulings.py

"""An area's rulings, delivered when its files are touched (milestone 444).

A RULING is the operator's decision about how one area of the work must behave.
It lives in a `Rulings` section at the end of the area's System description
(writing-records.md says how one is written). The description already rides
every record filed under the System; this module is the other half: the first
command or edit in a session that touches the System's files
(`System.path_patterns`) shows its rulings, one line per System.

A LOOKUP, NOT A SEARCH. A shell command scores against prose decisions at noise
level, so nothing here is ranked: a path either falls under a System's
patterns or it does not. It takes no budget and no floor from the retrieval
arms beside it, and writes no retrieval_logs row — there is no score
distribution for it to join. Its surfacings go to `system_usage_events`.

ONCE IN FULL, THEN A REFERENCE (the #3750 shape). The hook keeps the Systems
already shown this session and passes them back; a repeat renders as one short
line rather than the rulings again, and is not counted as a surfacing.

Fails open everywhere: a delivery aid must never break the act it rides on.
"""
from __future__ import annotations

import logging
import posixpath
import re
import shlex

from scribe.services import systems as systems_svc
from scribe.services.system_usage import record_system_surfaced
from scribe.services.text import elide

logger = logging.getLogger(__name__)

# The section heading: `Rulings`, optionally as a markdown heading, bold, or
# with a trailing colon. The LAST one wins — the section sits at the end.
_HEADING = re.compile(r"^[ \t]*(?:#{1,6}[ \t]*)?\**Rulings\**[ \t]*:?[ \t]*$", re.I | re.M)
_BULLET = re.compile(r"^[ \t]*[-*][ \t]+(.*)$")

# Per line, so one System with a long list cannot crowd the others out.
RULING_CHARS = 300
RULINGS_PER_SYSTEM = 8

# A command is not a file list: these bound how much of it is read as paths.
_MAX_PATHS = 40
_LINE_SUFFIX = re.compile(r":\d+(?::\d+)?$")
_HAS_EXTENSION = re.compile(r"\.[A-Za-z0-9]{1,10}$")
_ASSIGNMENT = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=")


def parse_rulings(description: str | None) -> list[str]:
    """The rulings in a System description, one string per ruling.

    A bullet under the last `Rulings` heading is a ruling; an indented line
    after one continues it. The section ends at the first line that is
    neither — a heading or paragraph written after it is not a ruling.
    """
    text = description or ""
    found = list(_HEADING.finditer(text))
    if not found:
        return []
    out: list[str] = []
    for line in text[found[-1].end():].splitlines():
        if not line.strip():
            continue
        bullet = _BULLET.match(line)
        if bullet:
            if bullet.group(1).strip():
                out.append(bullet.group(1).strip())
        elif out and line[:1] in (" ", "\t"):
            out[-1] = f"{out[-1]} {line.strip()}"
        else:
            break
    return out


def _relative(token: str, root: str, cwd_rel: str) -> str:
    """`token` as a repo-relative path, or "" when it is not one."""
    if token.startswith("/"):
        if not root or not token.startswith(root.rstrip("/") + "/"):
            return ""
        token = token[len(root.rstrip("/")) + 1:]
    elif cwd_rel:
        token = f"{cwd_rel}/{token}"
    path = posixpath.normpath(token)
    if path in (".", "") or path == ".." or path.startswith("../"):
        return ""
    return path


def command_paths(command: str, *, root: str = "", cwd: str = "") -> list[str]:
    """The repo-relative paths a shell command names, in order, deduplicated.

    Reads and writes alike: a session forms its picture of an area by reading
    it, which is exactly when that area's rulings should reach it. A token is
    taken as a path when it has a `/` or a file extension; flags, URLs and
    globs are not paths. `root` is the repo's absolute root and `cwd` the
    command's working directory, both from the hook — an absolute path
    outside the root, or a relative one that climbs out of it, is dropped.
    """
    command = command or ""
    try:
        tokens = shlex.split(command, comments=False, posix=True)
    except ValueError:
        tokens = command.split()
    root = (root or "").rstrip("/")
    cwd_rel = ""
    if root and cwd:
        cwd = cwd.rstrip("/")
        if cwd.startswith(root + "/"):
            cwd_rel = cwd[len(root) + 1:]
    out: list[str] = []
    for raw in tokens:
        token = raw.strip().strip("'\"").rstrip(";,)|&")
        if token.startswith("-"):
            if "=" not in token:
                continue
            token = token.split("=", 1)[1]
        elif _ASSIGNMENT.match(token):
            token = token.split("=", 1)[1]
        if not token or "://" in token or any(c in token for c in "*?[]{}$`<>"):
            continue
        # `file.py:120` and a test id `file.py::test_x` name the file.
        token = _LINE_SUFFIX.sub("", token.split("::", 1)[0])
        if "/" not in token and not _HAS_EXTENSION.search(token):
            continue
        path = _relative(token, root, cwd_rel)
        if path and path not in out:
            out.append(path)
            if len(out) >= _MAX_PATHS:
                break
    return out


def _full_line(system, path: str, rulings: list[str]) -> str:
    shown = [elide(r, RULING_CHARS)[0] for r in rulings[:RULINGS_PER_SYSTEM]]
    more = len(rulings) - len(shown)
    listed = " ".join(f"({i}) {r}" for i, r in enumerate(shown, 1))
    if more > 0:
        listed += f" (+{more} more in `get_system({system.id})`)"
    return (
        f"> Rulings for `{path}` — the operator's decisions about "
        f"{system.name} (System {system.id}): {listed} "
        "Work here keeps to them; something that would depart from one is a "
        "question for the operator, not an option to choose. "
        "(Shown once per session.)"
    )


def _reference_line(system, path: str) -> str:
    return (
        f"> `{path}` is in {system.name} (System {system.id}) — its rulings "
        "were shown earlier this session and still apply."
    )


async def rulings_for_paths(
    user_id: int,
    project_id: int,
    paths: list[str],
    *,
    seen: list[int] | set[int] | None = None,
    source: str,
) -> dict:
    """Rulings lines for the Systems whose files `paths` touch.

    Returns {"lines": [...], "system_ids": [...]} — `system_ids` are the
    Systems shown IN FULL on this call, for the hook to add to the session's
    seen list and for the usage ledger; a System in `seen` gets a reference
    line and is in neither. A System with patterns but no rulings says
    nothing: there is no decision to deliver, and the charter already rides
    its records.
    """
    out: dict = {"lines": [], "system_ids": []}
    if not project_id or not paths:
        return out
    try:
        already = {int(s) for s in (seen or [])}
        for system, hit in await systems_svc.systems_for_paths(user_id, project_id, paths):
            rulings = parse_rulings(system.description)
            if not rulings:
                continue
            if system.id in already:
                out["lines"].append(_reference_line(system, hit[0]))
            else:
                out["lines"].append(_full_line(system, hit[0], rulings))
                out["system_ids"].append(system.id)
        if out["system_ids"]:
            record_system_surfaced(
                user_id=user_id, system_ids=out["system_ids"], source=source,
                project_id=project_id,
            )
    except Exception:
        logger.debug("rulings arm failed", exc_info=True)
        return {"lines": [], "system_ids": []}
    return out