shell.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/tools/shell.py
 
"""
Sandboxed shell execution with a whole-argv command allowlist (M8.0 WU-3).
 
Purpose
-------
Provides :func:`run_command`, which parses a command string to argv,
rejects shell metacharacters outside quotes, and validates the WHOLE argv against the
locked Appendix A allowlist (:mod:`oc_chat.tools._argv_policy`) before
executing it ``shell=False`` with a scrubbed environment. There is no
``shell=True`` lane and no Unix→PowerShell conversion (F55).
 
**Tier 1 — allowed (read-only / no side-effects):** ``pip list``/``show``/
``freeze``, ``where``/``which``, ``dir``/``ls`` (a small option set + ≤1
in-sandbox path arg), ``echo`` (argv-only). Reads go through the filesystem
tools and git through ``oct_git_*`` — so ``git`` (freeform), ``cat``/``type``/
``grep``/``findstr``, ``curl``/``wget``, the shells, and the cmd.exe mutators
are **denied**.
 
**Tier 2 — human-confirm / grant-gated (argv-only):** ``python``/``pytest``/
``pip install``|``uninstall``/``oct``. M8.0-A (LD-4): a Tier-2 command is **not
executable by the agent** — a *model-supplied* flag never authorizes it. It
returns a ``requires_confirmation`` dict that no model-reachable path can satisfy
(fail closed). M8.0-C: the react-loop relays that signal to an interactive human
and, on 'y', re-authorizes the exact command via :func:`human_confirmed_command`
(a ContextVar the model cannot set); headless / sub-loop / batch fail closed.
One-use signed grants for headless Tier-2 are M8.1.
 
**The two timeouts (plan 283).** ``clamp_timeout`` bounds the ``timeout`` the
MODEL supplies — the M8 Appendix B ceiling, 300 s, with no out-of-band raise
path — and applies to every command. :func:`human_budget` is a second, separate
value carried the same way the approval is, read ONLY after the exact-string
approval matched, and it is the approving human's own patience: a number of
seconds, or ``None`` for "until it finishes". The ceiling is what the agent may
decide; the budget is what the human decided.
 
Responsibilities
----------------
- :func:`_classify` / :func:`_is_allowed` — back-compat wrappers over the
  whole-argv policy decision (``_argv_policy.classify_command``).
- :func:`run_command` — orchestrate validation, optional confirmation
  prompt, and ``subprocess.run`` execution with timeout and captured
  output.
- :func:`human_confirmed_command` / :func:`human_budget` — the two ContextVar
  seams the interactive react-loop sets after a human answers the Tier-2 ask:
  what was approved, and how long they will wait for it.
- :func:`_pip_index_pin` — the F78 index pin: an index-reaching ``pip``
  subcommand executes with ``--isolated --index-url <operator-trusted>``
  appended after the subcommand, and the ``python -m pip`` spelling of a
  command that names its own index is refused on the argv policy's terms.
- :func:`_python_m_pip_at` / :func:`_first_value_option` — CPython's own
  ``-m`` spellings, so the attached and clustered forms (``-mpip``, ``-qmpip``)
  reach the pin and ``-Xmpip`` (X's value) correctly does not.
- ``TOOLS`` and ``SCHEMAS`` for auto-registration.
 
Diagnostics
-----------
Domain: OC_CHAT.TOOLS
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- Commands not matching any whitelist are *blocked* — they return an
  error dict, never silently execute.
- Mutating ``git`` (add/commit/push) is intentionally *not* in either
  whitelist; agents must use ``oct_git_*`` instead.
- Subprocess calls always run with a finite timeout and captured
  stdout/stderr — no streaming output, no PTY.
- Tier-2 (code-exec / mutation) commands are not agent-runnable (M8.0-A,
  LD-4): a model-supplied flag never authorizes them; they return
  ``requires_confirmation`` and fail closed. Only the out-of-band human-confirm
  channel (M8.0-C, via :func:`human_confirmed_command`) or a one-use grant
  (M8.1) authorizes them.
- A granted ``pip install`` / ``pip download`` runs against the operator-trusted
  index and nothing else (M8 F78, SUPPLY-02 / plan 253). The pin is computed
  BEFORE the confirmation is offered and carried in the reason the human reads,
  so what is approved is what runs; it changes what EXECUTES and never what
  AUTHORIZES — the exact approved command string remains the only key.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.tool_results (CommandResult, ConfirmationRequired, ToolError - the run_command return types)
- oc_chat.tools._sandbox (scrubbed child environment, output secret redaction and the protected-read check)
- oc_chat.tools._argv_policy (classify_command, the pip index helpers and WINDOWS_BUILTINS - the whole-argv allowlist, the F78 index pin and refusal, and the built-ins run through cmd)
- oc_chat.tools.filesystem (lazy; _sandbox_root, _safe_path, _outside_sandbox_error - the sandbox root and search path checks)
- oc_chat.core.defaults (lazy; authority - the operator-trusted pip index URL for the F78 pin)
- oc_chat.core.resource_caps (lazy; clamp_timeout, clamp_results - the M8 ceilings on model-supplied arguments)
- oc_chat.tools._proc (lazy; run_cancellable - the cancellable subprocess runner)
- oc_chat.core.tool_activity (lazy; TOOL_ACTIVITY - the terminal activity record fed redacted output lines)
- oc_chat.core.cancellation (lazy; tool_cancel_requested - cooperative Ctrl+C polling in text_search)
- cmd (not imported; the Windows command interpreter, run as cmd /v:off /c for the built-ins dir and echo)
 
See Also
--------
oc_chat/docs/conventions/CONTRACTS.md §3 Tool registry contract; §4 Tool return-
value conventions (run_command → §4.1; confirmation gating → §4.5).
"""
import contextlib
import contextvars
import shlex
import subprocess
import os
import platform
from collections.abc import Callable, Iterator
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
 
from oc_diagnostics import _dbg
 
from oc_chat.core.tool_results import (
    CommandResult,
    ConfirmationRequired,
    ToolError,
)
from oc_chat.tools._sandbox import (
    LineRedactor, redact_secrets, scrubbed_env)
from oc_chat.tools._argv_policy import (
    PIP_INDEX_REACHING_SUBCOMMANDS,
    WINDOWS_BUILTINS,
    classify_command,
    pip_index_denial,
    pip_index_option,
)
 
# ── Whole-argv command policy (M8.0 WU-3; Appendix A / LD-4) ─────────────────
#
# The legacy longest-prefix ``startswith`` whitelist + the Unix→PowerShell
# ``_get_cross_platform_command`` conversion (F55) are gone. Classification is
# now a deterministic WHOLE-argv match against the Appendix A table
# (:func:`oc_chat.tools._argv_policy.classify_command`); :func:`run_command`
# parses to argv and executes ``shell=False``. ``_classify`` / ``_is_allowed``
# are retained as back-compat wrappers (tests / analytics) over the new decision.
 
 
def _classify(command: str) -> str:
    """Return ``'safe'``, ``'confirm'``, or ``'denied'`` for *command*.
 
    Back-compat wrapper over the whole-argv policy (M8.0 WU-3): Appendix A
    ``allow`` → ``'safe'``, ``confirm`` → ``'confirm'``, ``deny`` → ``'denied'``.
    """
    from .filesystem import _sandbox_root
    decision = classify_command(command, _sandbox_root()).decision
    return {"allow": "safe", "confirm": "confirm", "deny": "denied"}[decision]
 
 
def _is_allowed(command: str) -> bool:
    """Return ``True`` if *command* is admitted by the whole-argv policy
    (Tier-1 ``allow`` or Tier-2 ``confirm``); ``False`` if denied."""
    return _classify(command) != "denied"
 
 
# ── Tier-2 out-of-band human-confirm channel (M8.0-C; LD-4) ──────────────────
#
# M8.0-A made a Tier-2 command non-executable by the agent: a *model-supplied*
# flag never authorizes it. M8.0-C adds the ONLY authorization path -- an
# interactive human 'y'. The authority is carried by a ``ContextVar`` rather than
# a parameter or an env var precisely because the model cannot reach it: the agent
# supplies JSON tool arguments, never Python execution context. The interactive
# react-loop sets it (via :func:`human_confirmed_command`) to the exact command a
# human approved, around the single re-dispatch of that command; ``run_command``
# honors it ONLY on an exact-string match and the token is reset immediately after,
# so the authorization never leaks to another command or outlives the one call.
# Headless / sub-loop / batch / piped stdin never set it -> Tier-2 fails closed.
_human_approved_command: contextvars.ContextVar[str | None] = contextvars.ContextVar(
    "oc_chat_human_approved_command", default=None
)
 
 
@contextlib.contextmanager
def human_confirmed_command(command: str) -> Iterator[None]:
    """Authorize exactly *command* for one Tier-2 :func:`run_command` execution.
 
    For use ONLY by the interactive react-loop after an out-of-band human 'y'
    (M8.0-C). Not authority the model can forge: it sets no env var and is not a
    tool argument. The token is reset on exit so the grant is scoped to the single
    re-dispatch this brackets.
    """
    token = _human_approved_command.set(command)
    try:
        yield
    finally:
        _human_approved_command.reset(token)
 
 
# ── The human's budget (plan 283 P1; SHELL-04) ──────────────────────────────
#
# ``clamp_timeout`` bounds what the MODEL may ask for and is untouched by this:
# the 300 s ceiling is the answer to "how long may the agent decide to wait".
# The budget below answers a different question — "how long is the HUMAN who
# just approved this command willing to wait" — and it is therefore read in
# exactly one place: inside ``run_command``, on the branch where
# ``_human_approved_command`` already matched. It travels by ContextVar for the
# same reason the approval does: the model supplies JSON tool arguments, never
# Python execution context, so there is no spelling of a tool call that reaches
# it. ``None`` is a real value here and means "no ceiling at all" — the operator
# answered "run until it finishes" — which is why the ContextVar holds a record
# rather than a bare int: "unset" and "unlimited" must not collapse into one.
 
 
@dataclass(frozen=True)
class HumanBudget:
    """How long the approving human will wait, and who to ask for more."""
 
    seconds: "int | None"
    """Whole seconds to allow, or ``None`` for no ceiling at all."""
    on_expired: "Callable[[], int | None] | None" = None
    """Called once at expiry to ask for an extension; ``None`` never asks."""
 
 
_human_budget: contextvars.ContextVar["HumanBudget | None"] = contextvars.ContextVar(
    "oc_chat_human_budget", default=None
)
 
 
@contextlib.contextmanager
def human_budget(
    seconds: "int | None", *, on_expired: "Callable[[], int | None] | None" = None,
) -> Iterator[None]:
    """Set the approving human's wait for one confirmed :func:`run_command`.
 
    For use ONLY by the interactive react-loop, alongside
    :func:`human_confirmed_command`, after a human answered the Tier-2 ask
    (plan 283). *seconds* is the budget they chose and ``None`` means they chose
    "until it finishes"; *on_expired* is the callable that will ask them for one
    extension when the budget runs out. Like the approval token, this is reset on
    exit, so a budget never leaks to another command or outlives the one call.
 
    It authorises NOTHING on its own: a budget without a matching
    :func:`human_confirmed_command` is never read, because the branch that reads
    it is only reached once the exact-string approval has already matched.
    """
    token = _human_budget.set(HumanBudget(seconds, on_expired))
    try:
        yield
    finally:
        _human_budget.reset(token)
 
 
def deny_unconfirmed_tier2(result: dict, *, note: str = "") -> dict:
    """Fail-closed converter for contexts with no interactive human.
 
    A Tier-2 ``requires_confirmation`` cannot be satisfied without a human, so in a
    sub-loop / batch / headless run it is converted to a terminal ``error`` the model
    can act on (stop asking) instead of being handed back the un-satisfiable signal.
    Any other result passes through unchanged (identity).
    """
    if isinstance(result, dict) and result.get("requires_confirmation"):
        msg = result.get("message") or (
            "Tier-2 command requires out-of-band human approval."
        )
        return {"error": (msg + " " + note).strip()}
    return result
 
 
# ── The F78 package-index pin (SUPPLY-02; plan 253) ─────────────────────────
#
# WU-3c closed the ENV half of F78: ``scrubbed_env`` drops ``PIP_*``, so a
# planted ``PIP_INDEX_URL`` cannot redirect a granted install. Two halves were
# left open. pip still reads its CONFIGURATION FILES -- the user profile's
# ``pip.ini`` and the virtual environment's own -- and an index could still be
# named in the approved command itself. ``--isolated`` closes the environment
# and the user profile; it does NOT close the site (venv) configuration file,
# which pip consults only when no index is given explicitly. That is why both
# the flag and an explicit ``--index-url`` are appended: either alone leaves one
# of the two paths open.
 
#: Interpreters that can run pip as a module (``python -m pip install ...``).
#: That spelling classifies as a ``python`` Tier-2 command, so ``_classify_pip``
#: never sees it and the seam has to recognise it here.
_PYTHON_PROGRAMS: frozenset[str] = frozenset({"python", "python3", "py"})
 
#: CPython short options that TAKE A VALUE. Each takes it from the rest of its
#: own token when there is one (``-Xutf8``) and from the next argv element when
#: the token ends at the letter (``-X utf8``) — which is exactly why ``-Xmpip``
#: is X's value ``mpip`` and NOT ``-m pip``. ``-c`` and ``-m`` additionally
#: TERMINATE the option list: everything after them belongs to the code or the
#: module, never to the interpreter.
_PY_SHORT_VALUE_OPTIONS: frozenset[str] = frozenset({"c", "m", "W", "X"})
 
#: CPython short options that are pure flags, so an option cluster continues
#: past them: ``-qmpip`` is ``--quiet -m pip``. Probed one letter at a time
#: against CPython 3.14 rather than transcribed from ``--help``, which omits the
#: still-accepted legacy no-ops ``-t`` and ``-R``. ``-h`` / ``-?`` / ``-V`` are
#: flags too (``-Vzz`` answers "Unknown option: -z", so ``-V`` consumes nothing);
#: they make the interpreter print and exit, so nothing installs either way.
_PY_SHORT_FLAG_OPTIONS: frozenset[str] = frozenset("bdhiqstuvxBEIOPRSV?")
 
#: Sentinel for an option letter CPython does not accept. NUL can never be one
#: (the metacharacter gate denies it outright), so it cannot collide with ``-?``.
_PY_UNKNOWN_OPTION = "\0"
 
 
def _first_value_option(token: str) -> "tuple[str, str]":
    """The first value-taking option letter in a single-dash CPython *token*,
    and the value attached to it inside that same token.
 
    Returns ``("", "")`` for a cluster of pure flags and
    ``(_PY_UNKNOWN_OPTION, "")`` for a letter the interpreter refuses. A flag
    continues the cluster; the first value-taking letter ends it and takes the
    remainder of the token as its value — empty when the token ends there, in
    which case the value is the next argv element.
    """
    for pos, ch in enumerate(token[1:], start=1):
        if ch in _PY_SHORT_VALUE_OPTIONS:
            return ch, token[pos + 1:]
        if ch not in _PY_SHORT_FLAG_OPTIONS:
            return _PY_UNKNOWN_OPTION, ""
    return "", ""
 
 
def _python_m_pip_at(argv: list[str]) -> int:
    """Index of the token after ``-m pip`` in a *python* argv, or ``-1``.
 
    ``-m`` takes a value, and CPython's own parser accepts that value ATTACHED
    to the option token as readily as separated: ``-mpip``, and inside a flag
    cluster ``-qmpip`` / ``-Impip`` / ``-SEmpip``, are all ``-m pip``. Matching
    only a bare ``-m`` token let the attached spelling walk past both the
    refusal and the pin (gate finding, 2026-09-09) — the same class of gap the
    optparse abbreviations were for pip's own options.
 
    The scan is deliberately no WIDER than the interpreter. ``-W`` and ``-X``
    take their values the same way, so ``-Xmpip`` is X's value and not a module;
    ``-c`` terminates the option list, so ``-cmpip`` is code; and a token that
    is not an option at all is the script path, after which no module can
    follow. Each of those was verified by running CPython 3.14, because a scan
    that matched more than the interpreter does would deny commands that never
    reach pip at all.
    """
    i = 1
    while i < len(argv):
        token = argv[i]
        if not token.startswith("-") or token in ("-", "--"):
            return -1                       # the script path, or the end of options
        if token.startswith("--"):
            i += 1                          # a long option (--check-hash-based-pycs)
            continue
        option, attached = _first_value_option(token)
        if option == _PY_UNKNOWN_OPTION or option == "c":
            return -1                       # refused by the interpreter, or code
        if option == "":
            i += 1                          # a cluster of pure flags
            continue
        if option == "m":
            module = attached or (argv[i + 1] if i + 1 < len(argv) else "")
            if not module or Path(module).name.lower() != "pip":
                return -1
            return i + 1 if attached else i + 2
        i += 1 if attached else 2           # -W / -X consumed their value
    return -1
 
 
def _pip_subcommand_at(argv: list[str], program: str) -> int:
    """Position of an index-reaching pip subcommand in *argv*, or ``-1``.
 
    Both Tier-2 spellings are matched: a direct ``pip install ...`` and
    ``python -m pip install ...`` — the pin applies after ``pip`` wherever it
    sits in the argv, in whichever spelling the interpreter accepts
    (:func:`_python_m_pip_at`).
    """
    if program == "pip":
        at = 1
    elif program in _PYTHON_PROGRAMS:
        at = _python_m_pip_at(argv)
        if at < 0:
            return -1
    else:
        return -1
    return at if at < len(argv) and argv[at] in PIP_INDEX_REACHING_SUBCOMMANDS else -1
 
 
def _pip_index_pin(argv: list[str], program: str) -> "tuple[list[str] | None, str]":
    """Pin an index-reaching pip *argv* to the operator-trusted index (F78).
 
    Returns ``(executed_argv, denial)``. *executed_argv* is ``None`` when *argv*
    is not an index-reaching pip command, and the caller executes it unchanged.
    *denial* is non-empty when the command names an index or a trusted host
    itself: the argv policy already refuses that for the bare ``pip`` spelling,
    and this closes the ``python -m pip`` spelling on the same terms. Refusing
    rather than overriding is deliberate — appending a pin to a command that
    already carries ``--index-url`` would leave which index wins to pip's own
    last-one-wins parsing, and would leave the human approving a string whose
    index quietly did not apply.
    """
    at = _pip_subcommand_at(argv, program)
    if at < 0:
        return None, ""
    named = pip_index_option(argv[at + 1:])
    if named:
        return None, pip_index_denial(named)
    from oc_chat.core.defaults import authority
 
    pinned = list(argv)
    pinned[at + 1:at + 1] = ["--isolated", "--index-url", authority("pip_index_url")]
    return pinned, ""
 
 
# ── Public API ──────────────────────────────────────────────────────────────
 
def run_command(
    command: str,
    timeout: int = 30,
    confirmed: bool = False,
) -> CommandResult | ConfirmationRequired | ToolError:
    """Execute a whitelisted shell command and return its output.
 
    Parameters
    ----------
    command : str
        Shell command string.
    timeout : int
        Maximum seconds to wait (default 30), clamped to the M8 ceiling of
        300 s. On the human-confirmed Tier-2 path an out-of-band
        :func:`human_budget` replaces it with the approving human's own wait —
        see the module docstring's "The two timeouts".
    confirmed : bool
        Retained for back-compat and **ignored** (M8.0-A): a model-supplied flag
        never authorizes a Tier-2 command. The only authorization is the out-of-band
        :func:`human_confirmed_command` ContextVar (interactive human 'y', M8.0-C) —
        unreachable by the model — or a one-use grant (M8.1).
 
    Returns
    -------
    dict
        ``{"stdout": str, "stderr": str, "returncode": int}`` on success, or
        ``{"requires_confirmation": True, "command": str, "reason": str, "message":
        str}`` for a Tier-2 command the agent cannot self-authorize (the react-loop
        relays it to a human via :func:`_resolve_tier2_confirmation`).
    """
    func = "run_command"
    from oc_chat.core.resource_caps import clamp_timeout
    timeout = clamp_timeout(timeout)  # M8.0 WU-9a: model arg cannot exceed 300 s
    _dbg("OC_CHAT.TOOLS", func, f"cmd={command!r} timeout={timeout} confirmed={confirmed}", 2)
    if not command or not command.strip():
        return {"error": "Empty command provided"}
 
    # M8.0 WU-3: deterministic whole-argv allowlist (Appendix A / LD-4). The
    # command is parsed to argv and validated as a whole; there is no shell=True
    # lane and no Unix→PowerShell conversion (F55).
    from .filesystem import _sandbox_root
    decision = classify_command(command, _sandbox_root())
    if decision.decision == "deny":
        _dbg("OC_CHAT.TOOLS", func, f"denied: {decision.reason}", 3)
        return {"error": f"Command denied: {decision.reason}"}
    pinned_argv: "list[str] | None" = None
    #: Only a human-set budget ever fills this in (plan 283); ``None`` keeps the
    #: timeout a hard verdict, which is what every Tier-1 command gets.
    on_budget_expired: "Callable[[], int | None] | None" = None
    reason = decision.reason
    if decision.decision == "confirm":
        # M8 F78 (plan 253): an index-reaching pip subcommand runs against the
        # operator-trusted index and nothing else. The pin is computed BEFORE
        # the confirmation is offered and carried in the reason, so the human
        # approves the argv that will actually run rather than a shorter string.
        pinned_argv, pin_denial = _pip_index_pin(decision.argv, decision.program)
        if pin_denial:
            _dbg("OC_CHAT.TOOLS", func, f"denied: {pin_denial}", 3)
            return {"error": f"Command denied: {pin_denial}"}
        if pinned_argv is not None:
            reason = f"{decision.reason}; runs as: {shlex.join(pinned_argv)}"
 
        # M8.0-A (LD-4): a model-supplied `confirmed=True` does NOT authorize a
        # Tier-2 (code-exec / package-mutation) command. M8.0-C: the ONLY
        # authorization is an out-of-band human 'y' relayed via the
        # `_human_approved_command` ContextVar (set by the interactive react-loop,
        # unreachable by the model), and only for the EXACT command approved.
        # Absent that, fail closed with a signal the react-loop can relay to a human.
        if _human_approved_command.get() != command:
            return {
                "requires_confirmation": True,
                "command": command,
                "reason": reason,
                "message": (
                    f"'{command}' is a Tier-2 command ({reason}) and cannot be "
                    f"run by the agent: Tier-2 requires out-of-band human approval that "
                    f"the agent cannot supply. Ask the user to run it, or use the "
                    f"dedicated tools (oct_* for git, the filesystem tools for files)."
                ),
            }
        _dbg("OC_CHAT.TOOLS", func,
             f"Tier-2 authorized by out-of-band human confirm: {command!r}", 2)
        # Plan 283 P1: and ONLY here, past the exact-string match, does the
        # approving human's own budget replace the model's clamped timeout.
        # Tier-1 never reaches this branch, so an allowed command keeps the
        # clamp whatever a stray budget in the context might say.
        _budget = _human_budget.get()
        if _budget is not None:
            timeout = _budget.seconds
            on_budget_expired = _budget.on_expired
            _dbg("OC_CHAT.TOOLS", func,
                 f"human budget applies: "
                 f"{'no ceiling' if timeout is None else str(timeout) + 's'} "
                 f"(extension {'offered' if on_budget_expired else 'not offered'})", 2)
    tier = "safe" if decision.decision == "allow" else "confirm"
 
    # Execute argv-only (shell=False). cmd.exe built-ins (dir/echo) have no
    # standalone executable on Windows, so they run via `cmd /c` — still
    # shell=False (the argv is passed as a list, never a shell string).
    try:
        # M8.0 WU-3c: minimal scrubbed env -- drop poisonable GIT_*/PIP_*/proxy/
        # cert/PYTHONPATH/... families that could redirect the child (F25/F72/F78/F79).
        env = scrubbed_env({"PYTHONIOENCODING": "utf-8"})
 
        from ._proc import run_cancellable
        from oc_chat.core.tool_activity import TOOL_ACTIVITY
 
        # The pinned argv when F78 applies, the classifier's argv otherwise.
        exec_argv = decision.argv if pinned_argv is None else pinned_argv
        executed_command = command if pinned_argv is None else shlex.join(pinned_argv)
        if platform.system() == "Windows" and decision.program in WINDOWS_BUILTINS:
            # /v:off disables cmd.exe delayed expansion (!VAR!); %VAR% expansion
            # is blocked upstream by the `%` metacharacter gate (WU-3a fold).
            exec_argv = ["cmd", "/v:off", "/c", *decision.argv]
 
        # Plan 279: each line goes to the terminal's activity record as the child
        # writes it, through the SAME redaction the result below goes through --
        # the status bar must never show what the result would have hidden. The
        # record is terminal-only: nothing here reaches the result, the events,
        # the session or the model.
        #
        # Plan 287: one LineRedactor per stream, built here and handed to each
        # reader by the runner. The filter remembers that it is inside a PEM
        # private-key block, which a per-line redaction of a whole-block pattern
        # otherwise cannot see; stdout and stderr are drained on separate threads,
        # so sharing one instance would interleave the two streams' armor states.
        def _note_lines_of_one_stream() -> "Callable[[str], None]":
            redactor = LineRedactor()
 
            def _note_line(line: str) -> None:
                TOOL_ACTIVITY.note_line(redactor.redact_line(line))
 
            return _note_line
 
        # Cancellable: a Ctrl+C terminates the child (a subprocess can be killed,
        # unlike a pure-Python worker thread). Returncode -2 == cancelled.
        result = run_cancellable(
            exec_argv,
            shell=False,
            timeout=timeout,
            env=env,
            cwd=_sandbox_root(),
            on_line_factory=_note_lines_of_one_stream,
            on_budget_expired=on_budget_expired,
        )
 
        # Secret-pattern redaction of all command output (M8 F46/F51; WU-2).
        stdout_content = redact_secrets(result.stdout or "")
        stderr_content = redact_secrets(result.stderr or "")
 
        response = {
            "stdout": stdout_content,
            "stderr": stderr_content,
            "returncode": result.returncode,
            "command_info": {
                "base_command": decision.program,
                # What RAN, which is the pinned argv when F78 applied.
                "full_command": executed_command,
                "tier": tier,
                "executed_at": datetime.now().isoformat(),
            },
        }
 
        warnings = []
        if not stdout_content.strip() and stderr_content.strip():
            warnings.append(
                "Empty stdout but stderr has content - check for output buffering issues"
            )
        if result.returncode != 0 and not stderr_content.strip():
            warnings.append(
                f"Non-zero return code ({result.returncode}) but no stderr output"
            )
        if warnings:
            response["warnings"] = warnings
 
        return response
 
    except subprocess.TimeoutExpired as timed_out:
        # Error output is redacted too -- it echoes the command, which may carry
        # a secret (e.g. an inline auth header) (M8 F46/F51; WU-2).
        # The exception's own timeout is quoted rather than the local, because a
        # human who extended the budget (plan 283) waited longer than the number
        # this function started with, and the line must say what actually elapsed.
        waited = timed_out.timeout if timed_out.timeout is not None else timeout
        return {
            "stdout": "",
            "stderr": redact_secrets(f"Command timed out after {waited:.0f}s: {command}"),
            "returncode": -1,
        }
    except FileNotFoundError as e:
        return {"stdout": "", "stderr": redact_secrets(str(e)), "returncode": -1}
    except Exception as e:
        return {"stdout": "", "stderr": redact_secrets(f"Execution error: {type(e).__name__}: {e}"), "returncode": -1}
 
 
# ── Tool registry entries ───────────────────────────────────────────────────
 
def text_search(
    pattern: str,
    search_dir: str = ".",
    *,
    file_path: str | None = None,
    case_sensitive: bool = False,
    max_results: int = 100,
) -> dict:
    """Search for a text pattern in files under *search_dir*.
 
    Phase P.10a: the historical parameter name was ``file_path``. That
    name was misleading (the argument is a *directory*, not a file path).
    The canonical name is now ``search_dir``. Callers passing
    ``file_path=`` still work for now but get a one-time
    :class:`DeprecationWarning`; on conflict (both keywords supplied),
    ``file_path`` wins for backwards compatibility.
 
    A *single file* is also a valid target (F26): ``search_dir`` may name a
    file, in which case only that file is searched; a directory walks the tree.
 
    Args:
        pattern: Regular-expression pattern to search for.
        search_dir: File or directory to search (default: project root).
        file_path: **Deprecated** alias for ``search_dir``. Keyword-only.
            Will be removed in a future release.
        case_sensitive: Match case when True (default False).
        max_results: Maximum number of matching files to return (default 100).
 
    Returns:
        Dict with ``matches`` list of ``{file, line_numbers, match_count}``,
        plus ``total_matches``, ``searched_files``, ``pattern``,
        and ``case_sensitive``.
    """
    import warnings
    from .filesystem import _outside_sandbox_error, _safe_path, _sandbox_root
    from ._sandbox import is_protected_read_path
    import re
 
    # P.10a deprecation alias: file_path wins on conflict for backward
    # compatibility. Emit a one-line DeprecationWarning per call.
    if file_path is not None:
        warnings.warn(
            "text_search(file_path=...) is deprecated; "
            "use text_search(search_dir=...) instead. The argument is a "
            "directory to search, not a file path — search_dir is the "
            "accurate name.",
            DeprecationWarning,
            stacklevel=2,
        )
        search_dir = file_path
 
    from oc_chat.core.resource_caps import clamp_results
    max_results = clamp_results(max_results)  # M8.0 WU-9a: cannot exceed 100
 
    # F26: accept a single FILE target as well as a directory. The 166 stream's
    # three failures were all `search_dir=<a file>` (with a `|` alternation
    # pattern, which is a red herring — the regex compiles fine); the historical
    # `is_dir()`-only guard rejected any file target as "not a directory". A file
    # target searches just that file; a directory target walks the tree.
    # F25: an out-of-sandbox search_dir is unreachable, not absent — name the
    # boundary (root + resolved path + existence) instead of raising or reading
    # as "not found".
    try:
        search_root = _safe_path(search_dir)
    except PermissionError:
        return {"error": _outside_sandbox_error(search_dir) or f"Search path not found: {search_dir}"}
    if search_root.is_file():
        search_candidates: "Iterator[object] | list" = [search_root]
    elif search_root.is_dir():
        search_candidates = search_root.rglob("*")
    elif not search_root.exists():
        return {"error": f"Search path not found: {search_dir}"}
    else:
        return {"error": f"Search path is neither a file nor a directory: {search_dir}"}
 
    flags = 0 if case_sensitive else re.IGNORECASE
    try:
        pattern_re = re.compile(pattern, flags)
    except re.error as e:
        return {"error": f"Invalid regex pattern: {e}"}
 
    matches = []
    files_searched = 0
 
    # Cooperative cancellation: a Ctrl+C sets a process-wide flag (a worker
    # thread running this search cannot be force-killed), polled once per file.
    from oc_chat.core.cancellation import tool_cancel_requested
 
    try:
        for fp in search_candidates:
            if tool_cancel_requested():
                return {
                    "matches": matches,
                    "total_matches": len(matches),
                    "searched_files": files_searched,
                    "pattern": pattern,
                    "case_sensitive": case_sensitive,
                    "cancelled": True,
                }
            if not fp.is_file():
                continue
            if is_protected_read_path(fp):
                continue  # do not search protected-read (secret) files (M8 F46b; WU-2)
            files_searched += 1
            try:
                with open(fp, "r", encoding="utf-8", errors="replace") as fh:
                    line_matches = [
                        ln for ln, line in enumerate(fh, start=1)
                        if pattern_re.search(line)
                    ]
                if line_matches:
                    matches.append({
                        "file": str(fp.relative_to(_sandbox_root())),
                        "line_numbers": line_matches,
                        "match_count": len(line_matches),
                    })
                    if len(matches) >= max_results:
                        break
            except (UnicodeDecodeError, OSError):
                continue
 
        return {
            "matches": matches,
            "total_matches": len(matches),
            "searched_files": files_searched,
            "pattern": pattern,
            "case_sensitive": case_sensitive,
        }
    except Exception as e:
        return {"error": f"Search error: {e}"}
 
 
TOOLS = {
    "run_command": run_command,
    "text_search": text_search,
}
 
SCHEMAS = [
    {
        "type": "function",
        "function": {
            "name": "run_command",
            "description": (
                "Run an allowlisted, read-only shell command (pip list/show/freeze, "
                "where/which, dir/ls, echo) and return stdout, stderr, and return "
                "code. Code-execution / mutation commands (python, pytest, pip "
                "install, ...) are NOT runnable here — use the dedicated tools "
                "(oct_* for git, the filesystem tools for files), or ask the user."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "command": {
                        "type": "string",
                        "description": "The shell command to run.",
                    },
                    "timeout": {
                        "type": "integer",
                        "description": "Timeout in seconds (default 30).",
                    },
                },
                "required": ["command"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "text_search",
            "description": (
                "Search for a regex pattern across files. `search_dir` may be a "
                "directory (walks the tree) OR a single file (searches just that "
                "file). Returns matching file paths, line numbers, and match "
                "counts. NOTE: the search-target parameter is named `search_dir`. "
                "The legacy alias `file_path` still works but emits a deprecation "
                "warning — use `search_dir` in new code."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "pattern": {
                        "type": "string",
                        "description": "Regular-expression pattern to search for.",
                    },
                    "search_dir": {
                        "type": "string",
                        "description": "File or directory to search (default: project root '.').",
                    },
                    "case_sensitive": {
                        "type": "boolean",
                        "description": "Match case when true (default false).",
                    },
                    "max_results": {
                        "type": "integer",
                        "description": "Maximum number of matching files to return (default 100).",
                    },
                },
                "required": ["pattern"],
            },
        },
    },
]