session_mode.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/session_mode.py
 
"""
The confirmation mode, the budget table, and what this session remembers (SHELL-04).
 
Purpose
-------
Plan 283. The Tier-2 ask used to have one useful answer: ``y``, meaning "run it
once, under the 300 s the model's clamp allows". That made a worker inside
oc_chat write helper scripts to shard a twelve-minute suite around the cap,
which is the cap's defect rather than the worker's. The operator ruled on
2026-09-15 (SHELL-04, amended v1.1 on 2026-09-17) for a wider set of answers and
for a MODE the human can switch on from the terminal at any moment. This module
holds the three pieces of state that ruling needs and nothing else:
 
- the **mode** (``ask`` or ``auto``), module state exactly like the plan-237
  checkpoint latches, settable only from the terminal;
- the **budget table**, which says how long each program is worth waiting for by
  default, and the one key rule that maps an argv onto it;
- the **remembered answers**, one per program, holding the ``a`` (always) and
  ``d`` (never) answers for the rest of this session.
 
What this module deliberately is NOT
------------------------------------
It is not authority. Nothing here decides whether a command may run — the argv
policy decides that, and :func:`oc_chat.tools.shell.human_confirmed_command`
remains the single authorisation seam. A remembered answer and the Auto mode
decide only whether the human is ASKED, exactly as the plan-235 directory grant
does, and both are consulted only after the react-loop's TTY check has already
established that a human is present.
 
It is also not a permission profile. ORCH-03 will add a tracked, persisted one;
until then :func:`remembered_answer` is the single function a profile would feed,
with the process-scoped store below as its only backend. That is why the lookup
is one function rather than a dict read scattered through the resolver.
 
Responsibilities
----------------
- :func:`current_mode` / :func:`set_mode` / :func:`auto_mode_active` /
  :func:`auto_mode_available` — the mode and the permissive-only rule.
- :func:`program_key` / :func:`table_budget_for` / :func:`default_budget_for` —
  the key rule and the budget table, written once.
- :func:`remembered_answer` / :func:`remember_answer` / :func:`forget_remembered`
  / :func:`remembered_programs` — the per-program memory.
- :func:`auto_answer_for` — what Auto mode answers for a program.
- :func:`ask_budget_extension` — the expiry prompt, with its grace window.
- :func:`read_typed_budget` — the ``t`` key's line read.
- :func:`reset_session_mode` — the session-change seam.
 
Contracts
---------
- The mode is settable ONLY through :func:`set_mode`, which the ``/mode``
  command, the ``--auto`` launch flag and the ``u`` key at the Tier-2 ask call.
  No tool, no model output, no settings file and no environment variable reaches
  it — unlike the risk profile, there is deliberately no env spelling at all.
- ``auto`` is inert outside the ``permissive`` risk profile, like the plan-235
  grant. The STATE is left alone there; only its effect is withdrawn, so
  returning to ``permissive`` does not silently resurrect a forgotten mode
  without the operator noticing it on the ``/mode`` output.
- State is process-lifetime and is NEVER persisted to disk or session metadata;
  a fresh process starts in ``ask`` with an empty memory. It is cleared at both
  seams where F67's ``reset_session_trust`` is called.
- An interrupt, a timeout or an unusable entry at any prompt here is the
  CONSERVATIVE answer — no extension, no budget, deny — never a silent continue.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.risk_profile (lazy; the permissive-only rule)
 
Diagnostics
-----------
Domain: OC_CHAT.CORE
Levels:
    L1 — never (a mode change is an operator decision, not a fault)
    L2 — mode / memory lifecycle events
    L3 — per-decision detail
    L4 — module load
"""
from __future__ import annotations
 
import os
import sys
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Sequence
 
from oc_diagnostics import _dbg
 
_dbg("OC_CHAT.CORE", "module", "loaded", 4)
 
 
#: The two modes. ``ask`` puts every confirmable call in front of the human;
#: ``auto`` answers the routine ones and still never answers the ruled set.
ASK = "ask"
AUTO = "auto"
 
#: Every mode name :func:`set_mode` will accept.
VALID_MODES: frozenset[str] = frozenset({ASK, AUTO})
 
#: What a program is worth waiting for when nobody says otherwise: the M8
#: ceiling's own value, so an unlisted program behaves exactly as it did before
#: this plan existed.
DEFAULT_BUDGET_S: int = 300
 
#: The table. Keys are :func:`program_key` outputs; a key that is not here falls
#: back to its bare program word (so every ``oct`` subcommand inherits ``oct``),
#: and then to :data:`DEFAULT_BUDGET_S`. These four are the long-running
#: commands the ruling names: a suite and the toolchain that drives one.
DEFAULT_BUDGETS: dict[str, int] = {
    "pytest": 900,
    "python -m pytest": 900,
    "oct test": 900,
    "oct": 900,
}
 
#: How long the expiry prompt waits for a key before reading silence as "no".
#: A module constant so a test can shorten it and an operator can find it.
BUDGET_EXTENSION_GRACE_S: float = 30.0
 
#: Interpreter basenames normalised to ``python`` in a program key, so
#: ``py -m pytest`` and ``python3 -m pytest`` are the same program as
#: ``python -m pytest`` rather than three separate entries nobody would fill in.
PYTHON_PROGRAMS: frozenset[str] = frozenset({"python", "python3", "py"})
 
#: The mode for this process. Module state, not session state, for the same
#: reason the plan-237 latches are: it must not survive a save/load, and every
#: seam that drops session trust drops it.
_MODE: str = ASK
 
 
@dataclass(frozen=True)
class RememberedAnswer:
    """A standing answer for one program, for the rest of this session."""
 
    run: bool
    """``True`` for the ``a`` answer (always run), ``False`` for ``d`` (never)."""
    budget_s: "int | None" = None
    """The budget a remembered YES runs under; meaningless when *run* is False."""
 
 
#: What this session has been told to stop asking about, keyed by program.
_REMEMBERED: "dict[str, RememberedAnswer]" = {}
 
 
# ── The mode ────────────────────────────────────────────────────────────────
 
def current_mode() -> str:
    """The mode this session is in, regardless of whether it can take effect."""
    return _MODE
 
 
def set_mode(mode: str) -> None:
    """Move the session to *mode*; raises ``ValueError`` on anything else.
 
    Refusing an unknown name rather than falling back is deliberate: a typo at
    the ``/mode`` prompt must not leave the operator believing they switched
    something when they did not.
    """
    func = "set_mode"
    global _MODE
    if mode not in VALID_MODES:
        raise ValueError(
            f"unknown confirmation mode {mode!r}; valid modes are "
            f"{', '.join(sorted(VALID_MODES))}")
    _MODE = mode
    _dbg("OC_CHAT.CORE", func,
         f"confirmation mode is now {mode!r} — "
         f"{'the routine asks are answered without a keystroke' if mode == AUTO else 'every confirmable call reaches the human'}",
         2, override=True)
 
 
def auto_mode_available() -> bool:
    """Whether Auto mode could take effect here at all (``permissive`` only)."""
    func = "auto_mode_available"
    from oc_chat.core.risk_profile import resolve_risk_profile
    if resolve_risk_profile() != "permissive":
        _dbg("OC_CHAT.CORE", func, "not permissive — Auto mode is not available", 3)
        return False
    return True
 
 
def auto_mode_active() -> bool:
    """Whether Auto mode is on AND allowed to act."""
    return _MODE == AUTO and auto_mode_available()
 
 
def reset_session_mode() -> None:
    """Drop the mode and every remembered answer (the session-change seam).
 
    The state is cleared BEFORE the diagnostic, never after: ``_dbg`` raises
    under warnings-as-errors if diagnostics are not initialised yet, and a mode
    that survived its own reset because a log line failed would be exactly the
    kind of silent authority this module exists to prevent.
    """
    func = "reset_session_mode"
    global _MODE
    was, _MODE = _MODE, ASK
    dropped = len(_REMEMBERED)
    _REMEMBERED.clear()
    if was != ASK or dropped:
        _dbg("OC_CHAT.CORE", func,
             f"confirmation mode reset to {ASK!r} (was {was!r}); "
             f"{dropped} remembered answer(s) dropped", 2, override=True)
 
 
# ── The key rule and the budget table ───────────────────────────────────────
 
def _program_word(arg0: str) -> str:
    """*arg0*'s basename, lower-cased, ``.exe`` stripped, python family folded."""
    name = Path(arg0).name.lower()
    if name.endswith(".exe"):
        name = name[:-4]
    return "python" if name in PYTHON_PROGRAMS else name
 
 
def _module_after_dash_m(argv: "Sequence[str]") -> str:
    """The module named by a ``-m`` in a python *argv*, or ``""``.
 
    Deliberately narrower than CPython's own parser, which accepts the value
    attached to the option token and inside a flag cluster. A spelling this scan
    misses simply falls back to the bare ``python`` key and therefore to the
    SHORTER default budget, so the miss is always the conservative direction —
    unlike :func:`oc_chat.tools.shell._python_m_pip_at`, where a miss would have
    let a command past a guard and the wider scan is mandatory.
    """
    for pos, token in enumerate(argv[1:], start=1):
        if token == "-m":
            return argv[pos + 1] if pos + 1 < len(argv) else ""
        if token.startswith("-m") and len(token) > 2:
            return token[2:]
        if not token.startswith("-"):
            return ""                       # the script path; no module follows
    return ""
 
 
def program_key(argv: "Sequence[str]") -> str:
    """The identity a budget and a remembered answer are keyed by.
 
    THE key rule, written once (P2): the classifier's ``argv[0]``, plus the
    ``-m`` module or the ``oct`` subcommand where one is present. It is what the
    human sees named in the prompt, in the ``[remembered]`` line and in ``/mode``
    output, so it has to read as the thing they are answering about — ``oct
    test`` and ``oct lint`` are different questions, and ``python -m pytest`` is
    the same question as ``pytest`` even though the argv is not.
    """
    argv = list(argv)
    if not argv:
        return ""
    program = _program_word(argv[0])
    if program == "python":
        module = _module_after_dash_m(argv)
        return f"python -m {module}" if module else program
    if program == "oct" and len(argv) > 1 and not argv[1].startswith("-"):
        return f"oct {argv[1].lower()}"
    return program
 
 
def table_budget_for(program: str) -> "int | None":
    """The table's budget for *program*, or ``None`` when it names none.
 
    Falls back from a two-word key to its program word, which is what makes one
    ``"oct": 900`` entry cover every ``oct`` subcommand without listing them.
    """
    if not program:
        return None
    if program in DEFAULT_BUDGETS:
        return DEFAULT_BUDGETS[program]
    return DEFAULT_BUDGETS.get(program.split(" ", 1)[0])
 
 
def default_budget_for(program: str) -> int:
    """What the ask offers for *program*: the table, else the shipped default."""
    budget = table_budget_for(program)
    return DEFAULT_BUDGET_S if budget is None else budget
 
 
# ── The memory ──────────────────────────────────────────────────────────────
 
def remembered_answer(program: str) -> "RememberedAnswer | None":
    """This session's standing answer for *program*, or ``None``.
 
    The ONE function a tracked permission profile (ORCH-03) would later feed;
    today the process-scoped store below is its only backend.
    """
    return _REMEMBERED.get(program) if program else None
 
 
def remember_answer(program: str, answer: RememberedAnswer) -> None:
    """Record *answer* as this session's standing answer for *program*."""
    func = "remember_answer"
    if not program:
        return
    _REMEMBERED[program] = answer
    _dbg("OC_CHAT.CORE", func,
         f"session answer recorded for {program!r}: "
         f"{'run' if answer.run else 'deny'}"
         f"{f' under {answer.budget_s}s' if answer.run else ''} — "
         f"the human is not asked about it again until /mode forget", 2, override=True)
 
 
def remembered_programs() -> "dict[str, RememberedAnswer]":
    """A copy of every standing answer, for ``/mode`` to print."""
    return dict(_REMEMBERED)
 
 
def forget_remembered() -> int:
    """Drop every standing answer; returns how many were dropped."""
    func = "forget_remembered"
    dropped = len(_REMEMBERED)
    _REMEMBERED.clear()
    if dropped:
        _dbg("OC_CHAT.CORE", func,
             f"{dropped} remembered answer(s) dropped — the human is asked again", 2,
             override=True)
    return dropped
 
 
def auto_answer_for(program: str) -> RememberedAnswer:
    """What Auto mode answers for *program* (P3).
 
    A program the table names is answered the way ``a`` would answer it — run
    under its table budget, and remember it, so the rest of the session reads as
    one decision rather than a stream of identical ones. Anything else runs
    under the shipped default and is NOT remembered: the mode is a standing
    answer to the routine, not a standing answer to whatever turns up.
    """
    table = table_budget_for(program)
    if table is not None:
        answer = RememberedAnswer(run=True, budget_s=table)
        remember_answer(program, answer)
        return answer
    return RememberedAnswer(run=True, budget_s=DEFAULT_BUDGET_S)
 
 
# ── The prompts this module owns ────────────────────────────────────────────
 
def _read_line(prompt: str = "") -> str:
    """Read one line from the human (``""`` on interrupt or a closed stdin).
 
    A tiny seam so the ``t`` key is testable without faking ``sys.stdin``. The
    ask happens before the dispatch bracket opens, so no Live display is running
    over this and a plain ``input`` is the right primitive.
    """
    func = "_read_line"
    try:
        return input(prompt)
    except (EOFError, KeyboardInterrupt):
        _dbg("OC_CHAT.CORE", func, "interrupt at the budget entry -> no budget", 3)
        return ""
 
 
def read_typed_budget(presenter: Any) -> "int | None":
    """The ``t`` key: read a budget in whole seconds, or ``None`` to deny.
 
    An empty, unparseable or non-positive entry returns ``None``, which the
    caller reads as a denial rather than as the default budget. Guessing what a
    mistyped number meant is not consent, and re-asking in a loop is worse — the
    human can press the key again on the next ask.
    """
    func = "read_typed_budget"
    raw = _read_line("  Budget in seconds: ").strip()
    try:
        seconds = int(raw)
    except ValueError:
        presenter.show_info(
            f"  Not a whole number of seconds: {raw!r}. Denied — ask again to retry.")
        _dbg("OC_CHAT.CORE", func, f"unusable typed budget {raw!r} -> deny", 3)
        return None
    if seconds <= 0:
        presenter.show_info(
            f"  A budget must be a positive number of seconds, not {seconds}. Denied.")
        _dbg("OC_CHAT.CORE", func, f"non-positive typed budget {seconds} -> deny", 3)
        return None
    return seconds
 
 
def _read_key_within(prompt: str = "", *, valid: str = "",
                     deadline_s: float = 0.0) -> str:
    """Read one key, giving up after *deadline_s* seconds (``""`` on silence).
 
    :func:`oc_ui.terminal.input.read_single_key` waits forever, which is right
    at an ask the human is already looking at and wrong at an expiry prompt that
    arrives minutes into a run they may have walked away from. This is the same
    read with a deadline: a poll on Windows, a ``select`` elsewhere, and an
    immediate ``""`` wherever there is no TTY to read from.
 
    Returns the key lower-cased, or ``""`` for Enter, a timeout, a key outside
    *valid*, or any failure — every one of which the caller reads as "no".
    """
    func = "_read_key_within"
    if not (sys.stdin and getattr(sys.stdin, "isatty", lambda: False)()):
        return ""
    if prompt:
        sys.stdout.write(prompt)
        sys.stdout.flush()
    try:
        return _poll_for_key(valid, time.monotonic() + max(0.0, deadline_s))
    except (OSError, ValueError, ImportError) as exc:
        # EXTERNAL_ERROR (L2): the terminal is the operator's, and a console we
        # cannot read from is silence, which is already the conservative answer.
        _dbg("OC_CHAT.CORE", func,
             f"EXTERNAL_ERROR: timed key read unavailable "  # noqa: E501
             f"({type(exc).__name__}: {exc}); reading it as no answer", 2)
        return ""
    finally:
        if prompt:
            sys.stdout.write("\n")
            sys.stdout.flush()
 
 
def _poll_for_key(valid: str, end: float) -> str:
    """Poll the TTY for a key in *valid* until *end*; ``""`` when none arrives."""
    if os.name == "nt":
        import msvcrt
 
        while time.monotonic() < end:
            if not msvcrt.kbhit():
                time.sleep(0.05)
                continue
            ch = msvcrt.getwch()
            if ch in ("\x00", "\xe0"):          # function-key prefix
                msvcrt.getwch()
                continue
            if ch in ("\r", "\n"):
                return ""
            ch = ch.lower()
            if not valid or ch in valid:
                return ch
        return ""
 
    import select
    import termios
    import tty
 
    fd = sys.stdin.fileno()
    old = termios.tcgetattr(fd)
    try:
        tty.setcbreak(fd)
        while True:
            remaining = end - time.monotonic()
            if remaining <= 0 or not select.select([sys.stdin], [], [], remaining)[0]:
                return ""
            ch = sys.stdin.read(1)
            if ch in ("\r", "\n"):
                return ""
            ch = ch.lower()
            if not valid or ch in valid:
                return ch
    finally:
        termios.tcsetattr(fd, termios.TCSADRAIN, old)
 
 
def ask_budget_extension(presenter: Any, program: str, seconds: int) -> "int | None":
    """Ask the human for one more budget-width; ``None`` declines.
 
    The hook :func:`oc_chat.tools._proc.run_cancellable` calls when a human's
    budget elapses. The child is still alive while this waits, so the answer is
    genuinely "keep going" rather than "start again", and the window is bounded:
    an operator who has walked away is not a yes, and the run they left behind
    must not outlive their attention by more than the grace window.
 
    The grace constant is read HERE, at call time, rather than captured — so
    monkeypatching :data:`BUDGET_EXTENSION_GRACE_S` shortens a real window and a
    later fix to it needs no second edit.
    """
    func = "ask_budget_extension"
    answer = _read_key_within(
        f"Budget of {seconds} s reached for {program}; extend by {seconds} s? (y/N) ",
        valid="yn", deadline_s=BUDGET_EXTENSION_GRACE_S)
    if answer == "y":
        _dbg("OC_CHAT.CORE", func,
             f"human extended {program!r} by {seconds}s at the budget prompt", 2,
             override=True)
        return seconds
    presenter.show_info(
        f"  No extension for {program}; the command was stopped at "
        f"{seconds} s.")
    _dbg("OC_CHAT.CORE", func,
         f"no extension for {program!r} within the "
         f"{BUDGET_EXTENSION_GRACE_S:.0f}s window -> the child is terminated", 2,
         override=True)
    return None
 
 
__all__ = [
    "ASK",
    "AUTO",
    "BUDGET_EXTENSION_GRACE_S",
    "DEFAULT_BUDGETS",
    "DEFAULT_BUDGET_S",
    "PYTHON_PROGRAMS",
    "RememberedAnswer",
    "VALID_MODES",
    "ask_budget_extension",
    "auto_answer_for",
    "auto_mode_active",
    "auto_mode_available",
    "current_mode",
    "default_budget_for",
    "forget_remembered",
    "program_key",
    "read_typed_budget",
    "remember_answer",
    "remembered_answer",
    "remembered_programs",
    "reset_session_mode",
    "set_mode",
    "table_budget_for",
]