prompt_invariant_rules.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/linter/prompt_invariant_rules.py
 
"""
Purpose
-------
OC Chat prompt-invariant structural rules, lifted into the OCT linter from
``oc_chat/invariants/prompt_invariant_rules.py`` (Plan 140) at the Plan 151
monorepo migration. Provides local structural enforcement for the Plan 135
prompt invariants. The module exposes portable rule metadata, runtime path
scoping, and source checkers that the OCT strict profile hosts as the
``prompt_invariants`` rule (replacing the OC Chat pytest wrapper).
 
Responsibilities
----------------
- Define rule metadata and agent-legible remediation text.
- Keep runtime path scope centralized and repository-relative (scoped to
  the OC Chat runtime source tree; a no-op for every other project).
- Check a single source string or a set of filesystem paths for invariant
  violations.
- Keep detector logic independent of OC Chat runtime imports.
 
Diagnostics
-----------
Domain: OCT-LINTER
Levels:
    L2 - lifecycle events
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- Public checkers accept repository-relative paths for matching.
- Absolute filesystem paths are normalized only through ``check_paths`` with
  an explicit root.
- Detectors must not edit source or depend on live OC Chat runtime objects.
- Findings include remediation text from the rule registry.
 
Dependencies
------------
- ast (parses the source and walks it for the structural detectors)
- collections.abc (Iterable, Iterator, Mapping - type hints for paths and the rule table)
- dataclasses (dataclass - the frozen RuleMetadata and Finding records)
- io (StringIO - feeds the source text to tokenize)
- pathlib (Path - filesystem paths for the runtime scope walk)
- tokenize (NAME tokens for the retired composition symbol check)
- types (MappingProxyType - the read-only RULES table)
- oct.core.diagnostics (structured debug logging via _dbg)
"""
from __future__ import annotations
 
import ast
import io
import tokenize
from collections.abc import Iterable, Iterator, Mapping
from dataclasses import dataclass
from pathlib import Path
from types import MappingProxyType
 
from oct.core.diagnostics import _dbg
 
 
_dbg("OCT-LINTER", "module", "loaded", 4)
 
 
@dataclass(frozen=True)
class RuleMetadata:
    """Metadata for one portable invariant rule."""
 
    rule_id: str
    purpose: str
    remediation: str
    detector_kind: str
    spec_reference: str
 
 
@dataclass(frozen=True)
class Finding:
    """One invariant finding."""
 
    rule_id: str
    path: str
    line: int
    col: int
    message: str
    remediation: str
 
 
_PROMPT_REFS = (
    "oc_chat/plans/135-prompt-redesign-agnostic-core.md section 4.7; "
    "oc_chat/docs/architecture/inner-loop/inner-loop.md section 8"
)
 
RULES: Mapping[str, RuleMetadata] = MappingProxyType({
    "prompt_no_raw_tool_schema_dump": RuleMetadata(
        rule_id="prompt_no_raw_tool_schema_dump",
        purpose="Prevent raw tool schema JSON from entering system-prompt construction.",
        remediation=(
            "Full schemas belong in the API tools parameter. messages[0] should "
            "contain only canonical policy plus compact tool map/reference so "
            "provider cache prefixes stay stable and portable."
        ),
        detector_kind="AST with anchored same-scope alias tracking",
        spec_reference=_PROMPT_REFS,
    ),
    "prompt_model_setter_chokepoint": RuleMetadata(
        rule_id="prompt_model_setter_chokepoint",
        purpose="Require runtime model switches to pass through ChatSession.set_model.",
        remediation=(
            "Raw writes skip the refresh contract and can leave cached context, "
            "prompt metadata, and presenter context-window gauges stale. Use "
            "session.set_model(new_model) for runtime switches."
        ),
        detector_kind="AST assignment checker",
        spec_reference=_PROMPT_REFS,
    ),
    "prompt_no_literal_mode_branch": RuleMetadata(
        rule_id="prompt_no_literal_mode_branch",
        purpose="Keep prompt composition mode behavior declarative.",
        remediation=(
            "Modes are declarative profiles. Add or modify profile data and "
            "shared resolver logic instead of hard-coding mode-specific branches."
        ),
        detector_kind="AST branch-condition checker",
        spec_reference=_PROMPT_REFS,
    ),
    "prompt_no_retired_composition_symbols": RuleMetadata(
        rule_id="prompt_no_retired_composition_symbols",
        purpose="Prevent retired prompt-composition API names from reappearing.",
        remediation=(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Rule IDs, mode literals, and remediation strings are invariant specifications, not secrets"
            "Use build_system_prompt, ensure_system_includes_tool_reference, "
            "and CANONICAL_POLICY. The retired names invite schema re-embedding "
            "and stale prompt ownership."
        ),
        detector_kind="Token checker with one explicit alias exception",
        spec_reference=_PROMPT_REFS,
    ),
    "prompt_cache_ctx_fields_distinct": RuleMetadata(
        rule_id="prompt_cache_ctx_fields_distinct",
        purpose="Keep cache-hit ratio and context saturation separate.",
        remediation=(
            "cache-hit is prompt-cache efficiency; ctx is context-window "
            "saturation from acc.max_prompt_in_turn / get_max_context_tokens(model). "
            "They answer different operational questions and must not share a "
            "computation."
        ),
        detector_kind="AST call keyword checker",
        spec_reference=_PROMPT_REFS,
    ),
})
 
_EXCLUDED_PARTS = {
    "__pycache__",
    "docs",
    "plans",
    "tests",
}
_MODE_LITERALS = {"standard", "option-c"}
_RETIRED_NAMES = {
    "build_system_prompt_with_tools",
    "ensure_system_includes_tool_schemas",
}
 
 
def is_runtime_path(path: str | Path) -> bool:
    """Return whether *path* is in the root-anchored runtime source scope."""
 
    parts = _path_parts(path)
    if not parts or not parts[-1].endswith(".py"):
        return False
    if any(part in _EXCLUDED_PARTS for part in parts):
        return False
    if parts == ("oc_chat", "main.py"):
        return True
    if len(parts) >= 3 and parts[0] == "oc_chat" and parts[1] == "core":
        return True
    if len(parts) >= 2 and parts[0] in {"providers", "oc_ui"}:
        return True
    return False
 
 
def check_source(path: str | Path, text: str) -> list[Finding]:
    """Check one repository-relative source path and return invariant findings."""
 
    normalized = _normalize_repo_path(path)
    if not is_runtime_path(normalized):
        return []
 
    tree = _parse_source(text)
    findings: list[Finding] = []
    if tree is not None:
        findings.extend(_check_raw_tool_schema_dump(normalized, tree))
        findings.extend(_check_model_setter_chokepoint(normalized, tree))
        findings.extend(_check_literal_mode_branch(normalized, tree))
        findings.extend(_check_cache_ctx_fields_distinct(normalized, tree))
    findings.extend(_check_retired_composition_symbols(normalized, text, tree))
    return sorted(findings, key=lambda item: (item.path, item.line, item.col, item.rule_id))
 
 
def check_prompt_invariants(path: str | Path, text: str) -> tuple[bool, str]:
    """OCT linter adapter: collapse :func:`check_source` findings to the
    ``(passed, message)`` contract the per-file linter dispatch expects.
 
    Returns ``(True, "")`` when the source is clean or out of the OC Chat
    runtime scope (a no-op for every non-OC-Chat project), else
    ``(False, "<one line per finding>")`` naming each violated rule with its
    location and remediation-bearing message.
    """
    findings = check_source(path, text)
    if not findings:
        return True, ""
    message = "; ".join(
        f"{f.rule_id} @ {f.line}:{f.col} - {f.message}" for f in findings
    )
    return False, message
 
 
def iter_runtime_paths(root: Path) -> Iterator[Path]:
    """Yield existing Python files in OC Chat's runtime source scope."""
 
    root = root.resolve()
    candidates = [
        root / "oc_chat" / "core",
        root / "oc_chat" / "main.py",
        root / "providers",
        root / "oc_ui",
    ]
    seen: set[Path] = set()
    for candidate in candidates:
        if candidate.is_file():
            paths = [candidate]
        elif candidate.is_dir():
            paths = sorted(candidate.rglob("*.py"))
        else:
            paths = []
        for path in paths:
            resolved = path.resolve()
            if resolved in seen:
                continue
            seen.add(resolved)
            try:
                relative = resolved.relative_to(root)
            except ValueError:
                continue
            if is_runtime_path(relative):
                yield resolved
 
 
def check_paths(paths: Iterable[Path], *, root: Path) -> list[Finding]:
    """Check filesystem paths by converting them to repository-relative paths."""
 
    root = root.resolve()
    findings: list[Finding] = []
    for path in paths:
        resolved = path.resolve()
        try:
            relative = resolved.relative_to(root)
        except ValueError:
            continue
        if not is_runtime_path(relative):
            continue
        text = resolved.read_text(encoding="utf-8", errors="replace")
        findings.extend(check_source(relative, text))
    return sorted(findings, key=lambda item: (item.path, item.line, item.col, item.rule_id))
 
 
def _normalize_repo_path(path: str | Path) -> str:
    value = str(path).replace("\\", "/")
    while value.startswith("./"):
        value = value[2:]
    return value.strip("/")
 
 
def _path_parts(path: str | Path) -> tuple[str, ...]:
    normalized = _normalize_repo_path(path)
    return tuple(part for part in normalized.split("/") if part and part != ".")
 
 
def _parse_source(text: str) -> ast.Module | None:
    try:
        return ast.parse(text)
    except SyntaxError:
        return None
 
 
def _finding(rule_id: str, path: str, line: int, col: int, message: str) -> Finding:
    metadata = RULES[rule_id]
    return Finding(
        rule_id=rule_id,
        path=path,
        line=line,
        col=col,
        message=message,
        remediation=metadata.remediation,
    )
 
 
def _check_raw_tool_schema_dump(path: str, tree: ast.Module) -> list[Finding]:
    if _path_parts(path)[0] == "providers":
        return []
 
    findings: list[Finding] = []
    for scope in _iter_scopes(tree):
        nodes = list(_iter_scope_nodes(scope))
        aliases = _tool_schema_aliases(nodes)
        for node in nodes:
            if (
                isinstance(node, ast.Call)
                and _is_json_dumps(node.func)
                and node.args
                and isinstance(node.args[0], ast.Name)
                and node.args[0].id in aliases
            ):
                findings.append(_finding(
                    "prompt_no_raw_tool_schema_dump",
                    path,
                    node.lineno,
                    node.col_offset + 1,
                    "Raw json.dumps(tool_schemas) or same-scope alias in prompt construction.",
                ))
    return findings
 
 
def _iter_scopes(tree: ast.Module) -> Iterator[ast.AST]:
    yield tree
    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            yield node
 
 
def _iter_scope_nodes(scope: ast.AST) -> Iterator[ast.AST]:
    body = getattr(scope, "body", [])
    for child in body:
        yield from _iter_scope_child_nodes(child)
 
 
def _iter_scope_child_nodes(node: ast.AST) -> Iterator[ast.AST]:
    yield node
    if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Lambda)):
        return
    for child in ast.iter_child_nodes(node):
        yield from _iter_scope_child_nodes(child)
 
 
def _tool_schema_aliases(nodes: list[ast.AST]) -> set[str]:
    aliases = {"tool_schemas"}
    changed = True
    while changed:
        changed = False
        for node in nodes:
            target_names = _assignment_target_names(node)
            value_name = _assignment_value_name(node)
            if value_name in aliases:
                for name in target_names:
                    if name not in aliases:
                        aliases.add(name)
                        changed = True
    return aliases
 
 
def _assignment_target_names(node: ast.AST) -> list[str]:
    if isinstance(node, ast.Assign):
        return [
            target.id
            for target in node.targets
            if isinstance(target, ast.Name)
        ]
    if isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name):
        return [node.target.id]
    return []
 
 
def _assignment_value_name(node: ast.AST) -> str | None:
    value: ast.AST | None
    if isinstance(node, ast.Assign):
        value = node.value
    elif isinstance(node, ast.AnnAssign):
        value = node.value
    else:
        value = None
    return value.id if isinstance(value, ast.Name) else None
 
 
def _is_json_dumps(func: ast.AST) -> bool:
    return (
        isinstance(func, ast.Attribute)
        and func.attr == "dumps"
        and isinstance(func.value, ast.Name)
        and func.value.id == "json"
    )
 
 
def _check_model_setter_chokepoint(path: str, tree: ast.Module) -> list[Finding]:
    visitor = _ModelAssignmentVisitor(path)
    visitor.visit(tree)
    return visitor.findings
 
 
class _ModelAssignmentVisitor(ast.NodeVisitor):
    def __init__(self, path: str) -> None:
        self.path = path
        self.class_stack: list[str] = []
        self.function_stack: list[str] = []
        self.findings: list[Finding] = []
 
    def visit_ClassDef(self, node: ast.ClassDef) -> None:
        self.class_stack.append(node.name)
        self.generic_visit(node)
        self.class_stack.pop()
 
    def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
        self.function_stack.append(node.name)
        self.generic_visit(node)
        self.function_stack.pop()
 
    def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
        self.function_stack.append(node.name)
        self.generic_visit(node)
        self.function_stack.pop()
 
    def visit_Assign(self, node: ast.Assign) -> None:
        for target in node.targets:
            self._check_target(target, node.value)
        self.generic_visit(node)
 
    def visit_AnnAssign(self, node: ast.AnnAssign) -> None:
        self._check_target(node.target, node.value)
        self.generic_visit(node)
 
    def visit_AugAssign(self, node: ast.AugAssign) -> None:
        self._check_target(node.target, node.value)
        self.generic_visit(node)
 
    def _check_target(self, target: ast.AST, value: ast.AST | None) -> None:
        if not _is_model_write_target(target):
            return
        if self._is_allowed_chokepoint(target, value):
            return
        self.findings.append(_finding(
            "prompt_model_setter_chokepoint",
            self.path,
            getattr(target, "lineno", 1),
            getattr(target, "col_offset", 0) + 1,
            "Raw session.model/self.model assignment outside ChatSession.set_model.",
        ))
 
    def _is_allowed_chokepoint(self, target: ast.AST, value: ast.AST | None) -> bool:
        return (
            self.class_stack[-1:] == ["ChatSession"]
            and self.function_stack[-1:] == ["set_model"]
            and isinstance(target, ast.Attribute)
            and isinstance(target.value, ast.Name)
            and target.value.id == "self"
            and isinstance(value, ast.Name)
            and value.id == "new_model"
        )
 
 
def _is_model_write_target(target: ast.AST) -> bool:
    return (
        isinstance(target, ast.Attribute)
        and target.attr == "model"
        and isinstance(target.value, ast.Name)
        and target.value.id in {"self", "session"}
    )
 
 
def _check_literal_mode_branch(path: str, tree: ast.Module) -> list[Finding]:
    findings: list[Finding] = []
    for node in ast.walk(tree):
        test: ast.AST | None = None
        if isinstance(node, (ast.If, ast.IfExp, ast.While)):
            test = node.test
        if test is None:
            continue
        for candidate in ast.walk(test):
            if isinstance(candidate, ast.Compare) and _compare_has_mode_literal(candidate):
                findings.append(_finding(
                    "prompt_no_literal_mode_branch",
                    path,
                    candidate.lineno,
                    candidate.col_offset + 1,
                    "Literal mode branch on standard/option-c.",
                ))
    return findings
 
 
def _compare_has_mode_literal(node: ast.Compare) -> bool:
    left = node.left
    for operator, right in zip(node.ops, node.comparators):
        if isinstance(operator, (ast.Eq, ast.NotEq)):
            if (
                (_is_mode_expr(left) and _is_mode_literal(right))
                or (_is_mode_literal(left) and _is_mode_expr(right))
            ):
                return True
        left = right
    return False
 
 
def _is_mode_expr(node: ast.AST) -> bool:
    return (
        (isinstance(node, ast.Name) and node.id == "mode")
        or (isinstance(node, ast.Attribute) and node.attr == "mode")
    )
 
 
def _is_mode_literal(node: ast.AST) -> bool:
    return isinstance(node, ast.Constant) and node.value in _MODE_LITERALS
 
 
def _check_retired_composition_symbols(
    path: str,
    text: str,
    tree: ast.Module | None,
) -> list[Finding]:
    allowed_default_system = _allowed_default_system_positions(path, tree)
    findings: list[Finding] = []
    try:
        tokens = tokenize.generate_tokens(io.StringIO(text).readline)
        for token in tokens:
            if token.type != tokenize.NAME:
                continue
            line, col = token.start
            if token.string in _RETIRED_NAMES:
                findings.append(_finding(
                    "prompt_no_retired_composition_symbols",
                    path,
                    line,
                    col + 1,
                    f"Retired prompt composition symbol {token.string!r}.",
                ))
            elif token.string == "_DEFAULT_SYSTEM" and (line, col) not in allowed_default_system:
                findings.append(_finding(
                    "prompt_no_retired_composition_symbols",
                    path,
                    line,
                    col + 1,
                    # OCT-LINT: disable=no_hardcoded_secrets reason="Rule IDs, mode literals, and remediation strings are invariant specifications, not secrets"
                    "Unauthorized _DEFAULT_SYSTEM name token.",
                ))
    except tokenize.TokenError:
        return findings
    return findings
 
 
def _allowed_default_system_positions(
    path: str,
    tree: ast.Module | None,
) -> set[tuple[int, int]]:
    if path != "oc_chat/core/session.py" or tree is None:
        return set()
    allowed: set[tuple[int, int]] = set()
    for node in ast.walk(tree):
        if not isinstance(node, ast.Assign):
            continue
        if not (
            isinstance(node.value, ast.Name)
            and node.value.id == "CANONICAL_POLICY"
        ):
            continue
        for target in node.targets:
            if isinstance(target, ast.Name) and target.id == "_DEFAULT_SYSTEM":
                allowed.add((target.lineno, target.col_offset))
    return allowed
 
 
def _check_cache_ctx_fields_distinct(path: str, tree: ast.Module) -> list[Finding]:
    findings: list[Finding] = []
    for node in ast.walk(tree):
        if not isinstance(node, ast.Call) or not _is_show_turn_summary_call(node):
            continue
        keyword_values = {
            keyword.arg: keyword.value
            for keyword in node.keywords
            if keyword.arg in {"cache_hit_pct", "context_used_pct"}
        }
        cache_value = keyword_values.get("cache_hit_pct")
        context_value = keyword_values.get("context_used_pct")
        if cache_value is None or context_value is None:
            continue
        if isinstance(cache_value, ast.Constant) and isinstance(context_value, ast.Constant):
            continue
        if ast.dump(cache_value, include_attributes=False) != ast.dump(
            context_value,
            include_attributes=False,
        ):
            continue
        findings.append(_finding(
            "prompt_cache_ctx_fields_distinct",
            path,
            node.lineno,
            node.col_offset + 1,
            "cache_hit_pct and context_used_pct share the same computation.",
        ))
    return findings
 
 
def _is_show_turn_summary_call(node: ast.Call) -> bool:
    return (
        (isinstance(node.func, ast.Attribute) and node.func.attr == "show_turn_summary")
        or (isinstance(node.func, ast.Name) and node.func.id == "show_turn_summary")
    )