|
| None | _terminate ("subprocess.Popen" proc) |
| "Callable[[str], None] | None" | _observer_for_one_pipe ("Callable[[str], None] | None" on_line, "Callable[[], Callable[[str], None]] | None" on_line_factory) |
| None | _read_pipe (pipe, "list[str]" sink, "Callable[[str], None] | None" on_line) |
| "subprocess.CompletedProcess" | run_cancellable ("str | list[str]" cmd, *, bool shell=False, "float | None" timeout=None, "dict | None" env=None, "str | None" cwd=None, float poll_interval=0.1, "Callable[[str], None] | None" on_line=None, "Callable[[], Callable[[str], None]] | None" on_line_factory=None, "Callable[[], int | None] | None" on_budget_expired=None) |
Cancellable subprocess execution for tools.
Purpose
-------
:func:`run_cancellable` is a drop-in for ``subprocess.run(capture_output=True,
text=True, timeout=...)`` that additionally polls the process-wide cooperative
tool-cancel flag (:func:`oc_chat.core.cancellation.tool_cancel_requested`) and
**terminates the child** when a Ctrl+C is signalled. Tools that shell out
(``run_command``, the ``oct`` wrappers) route through this so a Ctrl+C actually
stops the underlying process instead of orphaning it — a subprocess, unlike a
pure-Python thread, *can* be killed.
Plan 279 (P1) replaced the single ``communicate()`` drain with one reader thread
per pipe, so a caller can watch the child's output WHILE it runs instead of only
after it exits. The optional ``on_line`` observer is what the status bar's tail
is built from; it is strictly an observer — the capture, the return code, the
cancel path and the timeout path are all exactly what ``communicate()`` produced.
Plan 283 (P1) added the optional ``on_budget_expired`` hook. A timeout here is
two different events wearing one name: the MODEL's clamped ceiling running out,
which must kill the child exactly as it always has, and a HUMAN's budget running
out, which the human may want to extend because they are sitting in front of it.
The hook is how the caller tells the two apart — with no hook this function
behaves byte for byte as it did before the parameter existed.
Responsibilities
----------------
- :func:`run_cancellable` — Popen + one reader thread per pipe + a poll loop that
terminates the child on cancel (returncode ``-2``, ``[cancelled by user]``
appended to stderr) or on timeout (raises ``subprocess.TimeoutExpired``, like
``subprocess.run``) — unless an ``on_budget_expired`` hook buys more time.
- :func:`_observer_for_one_pipe` — pick the observer one reader will use: a
fresh one from ``on_line_factory`` when the caller keeps per-stream state, the
shared ``on_line`` otherwise.
- :func:`_read_pipe` — drain one text pipe line by line into a list, offering
each line to the optional observer.
- :func:`_terminate` — best-effort terminate-then-kill of a child.
Diagnostics
-----------
Domain: OC_CHAT.TOOLS
Levels:
L2 — cancel / terminate lifecycle
L3 — semantic details
L4 — module load
Contracts
---------
- Returns a :class:`subprocess.CompletedProcess` so callers use ``.stdout`` /
``.stderr`` / ``.returncode`` exactly as with ``subprocess.run``.
- The captured text is byte-identical to what ``communicate()`` returned: the
pipes are opened in text mode with universal newlines, and joining the lines a
text-mode iterator yields reproduces the stream exactly.
- One reader thread PER PIPE (not one shared drain) is what prevents the
large-output deadlock that a bare ``poll()`` loop would risk — a full stderr
buffer can no longer block the stdout reader or vice versa.
- ``on_line`` is an observer and nothing more: it never sees a line the capture
does not contain, it cannot change the result or the return code, and an
exception it raises is swallowed so the child is never affected by it.
- ``on_line_factory`` (plan 287) is the same observer contract, bound PER PIPE:
it is called once for each reader, so an observer that carries state from one
line to the next — the status bar's private-key armor filter — never has
stdout's state mixed into stderr's. The binding lives here, in the one place
that knows how many pipes there are, rather than in a caller guessing at
thread identity.
- ``on_budget_expired`` is called AT MOST ONCE per run, only when a *timeout*
was given and actually elapsed. It returns whole seconds to add to the wait,
or ``None`` to decline; declining terminates the child on exactly the path a
hookless run takes. An exception from the hook is the conservative answer —
it is caught and read as a decline, never as an extension, because the hook's
job is to ask a human and a failed question is not a yes.
- Best-effort kill: the direct child is terminated; on ``shell=True`` a
grandchild may briefly outlive it (documented limitation).
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.cancellation (lazy; the cooperative tool-cancel flag)
| "subprocess.CompletedProcess" oc_chat.tools._proc.run_cancellable |
( |
"str | list[str]" | cmd, |
|
|
* | , |
|
|
bool | shell = False, |
|
|
"float | None" | timeout = None, |
|
|
"dict | None" | env = None, |
|
|
"str | None" | cwd = None, |
|
|
float | poll_interval = 0.1, |
|
|
"Callable[[str], None] | None" | on_line = None, |
|
|
"Callable[[], Callable[[str], None]] | None" | on_line_factory = None, |
|
|
"Callable[[], int | None] | None" | on_budget_expired = None ) |
Run *cmd* like ``subprocess.run(capture_output=True, text=True)`` but
cancellable.
Polls the cooperative tool-cancel flag every *poll_interval* seconds. On
cancel, the child is terminated and a :class:`subprocess.CompletedProcess`
with ``returncode == -2`` and a ``[cancelled by user]`` stderr marker is
returned. On *timeout*, the child is terminated and
:class:`subprocess.TimeoutExpired` is raised (matching ``subprocess.run`` so
existing handlers keep working).
*on_line* (plan 279) is called with each stdout and stderr line as the child
writes it, trailing newline stripped. It is a pure observer: it sees only
what the capture also contains, it cannot change the result, and an exception
it raises is caught and ignored. Callers that pass nothing get exactly the
behaviour this function had before the parameter existed. The caller is
responsible for any filtering the line needs — ``run_command`` redacts each
line with the same filter it applies to the result before the line reaches a
sink, because the terminal must not show what the result would have hidden.
*on_line_factory* (plan 287) is how a caller whose filter carries state gets
ONE observer per pipe instead of one shared by both readers: it is called
once per reader and its return value is that reader's *on_line*. Supplying it
ignores *on_line*; supplying neither is exactly the behaviour this function
had before either parameter existed.
*on_budget_expired* (plan 283) turns the *timeout* from a verdict into a
question, ONCE. When it is supplied and the timeout elapses, the child is
left running while the hook is called; the hook returns whole seconds to
continue for, or ``None`` to decline. A decline — and a hook that raises,
which is read as a decline — terminates the child and raises
:class:`subprocess.TimeoutExpired` exactly as a hookless run does. Callers
that pass nothing get precisely the behaviour this function had before the
parameter existed, which is what keeps the MODEL's clamped ceiling a hard
ceiling: only a human-set budget ever arrives here with a hook attached.
Definition at line 171 of file _proc.py.
1.15.0