feat(bash): opt-in background shells with delta output reader and kill tool (#817)

Restore 'start a dev server, use it in a later call' as an explicit opt-in
after #816 made bash reap its whole process group on return. The surface
mirrors the dominant coding-agent convention: bash(run_in_background=true)
returns a bash_N handle immediately; bash_output(id, filter?) returns only
output produced since the previous read plus status and exit code;
kill_shell(id) terminates the shell's whole process group.

- Per-session BackgroundShellRegistry: capped rolling line buffer with
  drop-oldest gap accounting, exit-order record pruning, owner scoping for
  task_agents (shells reaped when the agent finishes), liveness-guarded
  group kills (a stale pgid is never signalled), budgeted teardown joins.
- Exit notices ride a shared external-event rail (sanitize, soft cap,
  channel 'any', idle wake) now common to watch fires; a new 'quiet'
  NudgeQueue channel lets a user cancel defer pending notices without
  letting them re-wake the stopped workstream, and failed wake delivery
  re-queues external notices seq- and predicate-intact without re-arming
  the wake gate.
- The bash_output filter runs in a killable subprocess: sre holds the GIL
  for an entire search, so no in-process timeout can bound a hostile
  pattern. Scrubbed child env, pinned UTF-8 pipes, honest timeout-vs-
  helper-failure error taxonomy, per-line match window with explicit
  clipping notes; a failed filter never consumes the delta.
- run_in_background rides the bash intent-judge projection; bash_output is
  exempt from the repeat warning but still recorded so interleaved polls
  keep breaking other tools' streaks; all bash boolean args share one
  lenient coercion dialect.
- Shells survive generation cancel and die with the workstream: every
  teardown path funnels through ChatSession.close(); CLI exit and the
  server lifespan now close every loaded session, signal-first and
  Ctrl-C-safe, so nothing detached outlives a graceful shutdown.
This commit is contained in:
Patrick Buckley
2026-07-10 15:08:15 -07:00
parent bec757a96b
commit ab7d56e0ba
20 changed files with 3315 additions and 171 deletions
+19 -1
View File
@@ -16,6 +16,23 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen.
## [Unreleased]
### Added
- **Background shells: `bash` gains `run_in_background`, plus `bash_output` /
`kill_shell`.** Setting `run_in_background=true` starts the command as a
detached shell and returns immediately with a `bash_N` handle — "start a dev
server, use it in a later call" is back as an explicit opt-in (the shape
follows the convention the major coding agents converged on). `bash_output`
returns only output produced since the previous read (optionally filtered by
a regex) plus status and exit code; `kill_shell` terminates the shell's
whole process group. Output is buffered per shell with a drop-oldest cap, so
a chatty server can't grow memory unbounded. When a background shell exits,
a system notice lands at the next seam (waking an idle workstream if
needed). Shells survive a generation cancel, die with the workstream, and
never outlive a task_agent that started them; anything a background shell
itself backgrounds is still reaped when that shell exits — the no-leak
guarantee below is unchanged.
### Fixed
- **bash tool: never hang on a backgrounded child.** A command that left a
@@ -28,7 +45,8 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen.
(`errors="replace"`) instead of being dropped as a spurious error.
- **Behavior change:** a process the command backgrounds no longer survives
the call — nothing persists across bash invocations. (First-class
"run this in the background" support is planned as a separate change.)
"run this in the background" support landed separately — see
`run_in_background` under Added.)
## [1.7.3]
+43
View File
@@ -0,0 +1,43 @@
"""Shared process/polling helpers for the bash + background-shell suites.
One copy instead of three: ``test_bash_tool_background_hang``,
``test_background_shells`` and ``test_bash_background_tool`` all assert on
process liveness and poll for asynchronous state. Leading underscore so
pytest doesn't collect it.
"""
from __future__ import annotations
import contextlib
import os
import signal
import time
def pid_alive(pid: int) -> bool:
try:
os.kill(pid, 0)
except ProcessLookupError:
return False
except PermissionError:
return True
return True
def kill_pid(pid: int) -> None:
with contextlib.suppress(OSError):
os.kill(pid, signal.SIGKILL)
def poll_until(predicate, timeout=10.0, interval=0.05):
"""Poll ``predicate`` until truthy or ``timeout``; RETURNS the last value
(falsy on timeout — assert at the call site). Deliberately named apart
from ``tests/_helpers.wait_until``, which RAISES on timeout: two
same-named helpers with opposite failure semantics invite silently-green
tests."""
deadline = time.monotonic() + timeout
value = predicate()
while not value and time.monotonic() < deadline:
time.sleep(interval)
value = predicate()
return value
+675
View File
@@ -0,0 +1,675 @@
"""Unit tests for the per-session background-shell registry (#817).
The registry backs the ``bash(run_in_background=true)`` / ``bash_output`` /
``kill_shell`` tool surface: it spawns detached shells (``bash_N`` handles),
buffers their merged output in a capped rolling buffer, serves delta reads
(only lines since the last read), and reaps whole session groups on kill /
owner reap / close — the #816 rule (the tracked command defines the lifetime,
nothing escapes its process group) extended to explicit backgrounding.
Pure registry tests — no ChatSession. Session wiring is covered in
``test_bash_background_tool.py``.
"""
import re
import threading
import time
import pytest
from tests._proc_helpers import kill_pid as _kill_pid
from tests._proc_helpers import pid_alive as _pid_alive
from tests._proc_helpers import poll_until as _wait_until
from turnstone.core.background_shells import (
BackgroundShellRegistry,
FilterTimeoutError,
TooManyShellsError,
UnknownShellError,
)
def _wait_status(shell, status, timeout=10.0):
return _wait_until(lambda: shell.status == status, timeout=timeout)
@pytest.fixture
def registry():
reg = BackgroundShellRegistry()
yield reg
reg.close()
# ---------------------------------------------------------------------------
# Handles + spawning
# ---------------------------------------------------------------------------
def test_spawn_returns_incrementing_bash_handles(registry):
s1 = registry.spawn("sleep 30")
s2 = registry.spawn("sleep 30")
assert s1.shell_id == "bash_1"
assert s2.shell_id == "bash_2"
def test_spawned_shell_is_running_with_live_pid(registry):
shell = registry.spawn("sleep 30")
assert shell.status == "running"
assert _pid_alive(shell.pid)
def test_spawn_records_command(registry):
shell = registry.spawn("sleep 30")
assert shell.command == "sleep 30"
def test_spawn_after_close_is_refused():
reg = BackgroundShellRegistry()
reg.close()
with pytest.raises(RuntimeError):
reg.spawn("echo hi")
def test_max_live_shells_cap():
reg = BackgroundShellRegistry(max_shells=2)
try:
reg.spawn("sleep 30")
s2 = reg.spawn("sleep 30")
with pytest.raises(TooManyShellsError):
reg.spawn("sleep 30")
# Cap counts LIVE shells: killing one frees a slot.
reg.kill(s2.shell_id)
s3 = reg.spawn("sleep 30")
assert s3.status == "running"
finally:
reg.close()
def test_completed_shells_do_not_count_toward_cap():
reg = BackgroundShellRegistry(max_shells=1)
try:
s1 = reg.spawn("true")
assert _wait_status(s1, "completed")
s2 = reg.spawn("sleep 30")
assert s2.status == "running"
finally:
reg.close()
# ---------------------------------------------------------------------------
# Exit tracking
# ---------------------------------------------------------------------------
def test_natural_exit_sets_completed_and_exit_code(registry):
shell = registry.spawn("exit 7")
assert _wait_status(shell, "completed")
assert shell.exit_code == 7
def test_output_is_complete_once_completed(registry):
"""Status flips to completed only after the drains finish: a read at
completed must see everything the command wrote."""
shell = registry.spawn("echo alpha; echo beta")
assert _wait_status(shell, "completed")
read = registry.read(shell.shell_id)
assert [ln.strip() for ln in read.lines] == ["alpha", "beta"]
def test_leader_exit_reaps_backgrounded_grandchild(registry, tmp_path):
"""#816 consistency: the tracked command defines the lifetime. When the
leader exits, the whole session group is killed — a child the command
backgrounded does not outlive it."""
pidfile = tmp_path / "bg.pid"
shell = registry.spawn(f"sleep 60 & echo $! > {pidfile}; echo done")
bg_pid = None
try:
assert _wait_status(shell, "completed")
bg_pid = int(pidfile.read_text().strip())
assert _wait_until(lambda: not _pid_alive(bg_pid)), (
f"grandchild {bg_pid} leaked past leader exit"
)
read = registry.read(shell.shell_id)
assert "done" in "".join(read.lines)
finally:
if bg_pid is not None:
_kill_pid(bg_pid)
def test_stderr_lines_are_tagged_inline(registry):
shell = registry.spawn("echo out; echo err >&2")
assert _wait_status(shell, "completed")
lines = [ln.strip() for ln in registry.read(shell.shell_id).lines]
assert "out" in lines
assert "[stderr] err" in lines
# ---------------------------------------------------------------------------
# Delta reads
# ---------------------------------------------------------------------------
def test_read_returns_only_new_lines_since_last_read(registry):
"""The load-bearing convention: consecutive reads never overlap and never
drop a line — collecting across polls yields each line exactly once."""
shell = registry.spawn("echo one; echo two; sleep 0.4; echo three; sleep 30")
collected: list[str] = []
def _collect():
collected.extend(ln.strip() for ln in registry.read(shell.shell_id).lines)
return "three" in collected
assert _wait_until(_collect)
assert collected == ["one", "two", "three"]
registry.kill(shell.shell_id)
def test_read_after_exit_then_again_reports_no_new_output(registry):
shell = registry.spawn("echo hi")
assert _wait_status(shell, "completed")
first = registry.read(shell.shell_id)
assert [ln.strip() for ln in first.lines] == ["hi"]
second = registry.read(shell.shell_id)
assert second.lines == []
assert second.status == "completed"
assert second.exit_code == 0
def test_read_reports_status_and_exit_code(registry):
shell = registry.spawn("sleep 30")
read = registry.read(shell.shell_id)
assert read.shell_id == shell.shell_id
assert read.status == "running"
assert read.exit_code is None
registry.kill(shell.shell_id)
def test_read_unknown_id_raises_with_live_ids(registry):
registry.spawn("sleep 30")
with pytest.raises(UnknownShellError) as excinfo:
registry.read("bash_99")
assert "bash_99" in str(excinfo.value)
assert "bash_1" in str(excinfo.value)
def test_read_unknown_id_when_registry_empty(registry):
with pytest.raises(UnknownShellError):
registry.read("bash_1")
# ---------------------------------------------------------------------------
# Filter
# ---------------------------------------------------------------------------
def test_filter_selects_matching_lines_only(registry):
shell = registry.spawn("echo match-a; echo skip-b; echo match-c")
assert _wait_status(shell, "completed")
read = registry.read(shell.shell_id, filter_pattern="^match")
assert [ln.strip() for ln in read.lines] == ["match-a", "match-c"]
def test_filter_is_display_only_and_consumes_the_delta(registry):
"""Filtered-out lines are consumed, not deferred — the cursor advances
past the whole delta (Claude Code ``BashOutput`` semantics)."""
shell = registry.spawn("echo match-a; echo skip-b")
assert _wait_status(shell, "completed")
first = registry.read(shell.shell_id, filter_pattern="^match")
assert [ln.strip() for ln in first.lines] == ["match-a"]
assert first.new_line_count == 2 # both lines were new, one shown
second = registry.read(shell.shell_id)
assert second.lines == []
assert second.new_line_count == 0
def test_filter_uses_search_not_match(registry):
shell = registry.spawn("echo prefix-needle-suffix")
assert _wait_status(shell, "completed")
read = registry.read(shell.shell_id, filter_pattern="needle")
assert len(read.lines) == 1
def test_invalid_filter_regex_raises(registry):
shell = registry.spawn("echo hi")
assert _wait_status(shell, "completed")
with pytest.raises(re.error):
registry.read(shell.shell_id, filter_pattern="[unclosed")
# ---------------------------------------------------------------------------
# Buffer cap
# ---------------------------------------------------------------------------
def test_buffer_cap_drops_oldest_and_reports_gap():
reg = BackgroundShellRegistry(max_buffer_chars=200)
try:
shell = reg.spawn('for i in $(seq 1 50); do echo "line-$i-padded-to-length"; done')
assert _wait_status(shell, "completed")
read = reg.read(shell.shell_id)
assert read.dropped_lines > 0
# Newest output survives; the tail is intact.
assert read.lines, "cap must retain the newest lines, not drop everything"
assert read.lines[-1].strip() == "line-50-padded-to-length"
finally:
reg.close()
def test_unread_lines_excludes_buffer_evicted():
"""The exit notice's line count must not promise evicted output."""
reg = BackgroundShellRegistry(max_buffer_chars=200)
try:
shell = reg.spawn('for i in $(seq 1 50); do echo "line-$i-padded-to-length"; done')
assert _wait_status(shell, "completed")
with shell.lock:
retained = len(shell._buffer)
assert shell.unread_lines == retained
finally:
reg.close()
def test_buffer_gap_is_relative_to_cursor():
"""Lines dropped BEFORE being read are a reported gap; lines already
read and then dropped are not."""
reg = BackgroundShellRegistry(max_buffer_chars=10_000)
try:
shell = reg.spawn("echo early; sleep 30")
# Each poll consumes whatever has arrived; stop once something did.
assert _wait_until(lambda: bool(reg.read(shell.shell_id).lines))
# Everything emitted so far is read; nothing has been dropped.
read = reg.read(shell.shell_id)
assert read.dropped_lines == 0
reg.kill(shell.shell_id)
finally:
reg.close()
# ---------------------------------------------------------------------------
# Kill / reap / close
# ---------------------------------------------------------------------------
def test_kill_marks_killed_and_reaps_group(registry, tmp_path):
pidfile = tmp_path / "bg.pid"
shell = registry.spawn(f"sleep 60 & echo $! > {pidfile}; sleep 60")
assert _wait_until(pidfile.exists)
bg_pid = int(pidfile.read_text().strip())
try:
killed = registry.kill(shell.shell_id)
assert killed.status == "killed"
assert _wait_until(lambda: not _pid_alive(shell.pid))
assert _wait_until(lambda: not _pid_alive(bg_pid)), "grandchild survived kill"
finally:
_kill_pid(bg_pid)
def test_kill_unknown_id_raises(registry):
with pytest.raises(UnknownShellError):
registry.kill("bash_7")
def test_killed_shell_output_remains_readable(registry, tmp_path):
"""Output that arrived before the kill survives it: the record keeps its
buffer, and ``kill`` returns only after the drains have flushed."""
sentinel = tmp_path / "started"
shell = registry.spawn(f"echo before-kill; touch {sentinel}; sleep 60")
assert _wait_until(sentinel.exists)
registry.kill(shell.shell_id)
read = registry.read(shell.shell_id)
assert read.status == "killed"
assert "before-kill" in "".join(read.lines)
def test_signal_all_kills_live_shells_without_closing(registry):
"""signal_all is the instant half of teardown: every live group dies,
but the registry stays open (records intact, spawns still allowed) —
close() remains the complete teardown."""
s1 = registry.spawn("sleep 60")
s2 = registry.spawn("sleep 60")
registry.signal_all()
assert _wait_until(lambda: not _pid_alive(s1.pid))
assert _wait_until(lambda: not _pid_alive(s2.pid))
assert registry.has(s1.shell_id), "signal_all must not drop records"
s3 = registry.spawn("true")
assert _wait_status(s3, "completed"), "registry must remain usable after signal_all"
def test_close_kills_everything_and_is_idempotent():
reg = BackgroundShellRegistry()
s1 = reg.spawn("sleep 60")
s2 = reg.spawn("sleep 60")
reg.close()
assert not _pid_alive(s1.pid)
assert not _pid_alive(s2.pid)
reg.close() # second close is a no-op
def test_reap_owner_kills_only_that_owners_shells(registry):
mine = registry.spawn("sleep 60", owner="agent-1")
other = registry.spawn("sleep 60", owner="agent-2")
main = registry.spawn("sleep 60")
registry.reap(owner="agent-1")
assert _wait_until(lambda: not _pid_alive(mine.pid))
assert _pid_alive(other.pid)
assert _pid_alive(main.pid)
# ---------------------------------------------------------------------------
# Owner scoping
# ---------------------------------------------------------------------------
def test_owner_scoped_lookup_isolates_shells(registry):
agent_shell = registry.spawn("sleep 30", owner="agent-1")
main_shell = registry.spawn("sleep 30")
# Main scope cannot see the agent's shell...
with pytest.raises(UnknownShellError):
registry.read(agent_shell.shell_id)
# ...and the agent scope cannot see the main shell.
with pytest.raises(UnknownShellError):
registry.read(main_shell.shell_id, owner="agent-1")
# Each side reads its own.
assert registry.read(agent_shell.shell_id, owner="agent-1").status == "running"
assert registry.read(main_shell.shell_id).status == "running"
def test_shells_snapshot_is_owner_scoped(registry):
registry.spawn("sleep 30", owner="agent-1")
registry.spawn("sleep 30")
assert [s.owner for s in registry.shells(owner="agent-1")] == ["agent-1"]
assert [s.owner for s in registry.shells()] == [None]
def test_handles_are_unique_across_owners(registry):
a = registry.spawn("sleep 30", owner="agent-1")
b = registry.spawn("sleep 30")
assert a.shell_id != b.shell_id
# ---------------------------------------------------------------------------
# Exit callback (the notice hook)
# ---------------------------------------------------------------------------
def test_on_exit_fires_once_on_natural_exit():
fired = threading.Event()
seen = []
def _on_exit(shell):
seen.append(shell)
fired.set()
reg = BackgroundShellRegistry(on_exit=_on_exit)
try:
shell = reg.spawn("echo done")
assert fired.wait(10)
assert len(seen) == 1
assert seen[0].shell_id == shell.shell_id
assert seen[0].exit_code == 0
finally:
reg.close()
def test_on_exit_not_fired_for_kill():
seen = []
reg = BackgroundShellRegistry(on_exit=seen.append)
try:
shell = reg.spawn("sleep 60")
reg.kill(shell.shell_id)
assert _wait_until(lambda: not _pid_alive(shell.pid))
time.sleep(0.2) # give a buggy late callback a chance to land
assert seen == []
finally:
reg.close()
def test_on_exit_not_fired_for_close():
seen = []
reg = BackgroundShellRegistry(on_exit=seen.append)
shell = reg.spawn("sleep 60")
reg.close()
assert not _pid_alive(shell.pid)
time.sleep(0.2)
assert seen == []
# ---------------------------------------------------------------------------
# Review-hardening regressions (#817 code review)
# ---------------------------------------------------------------------------
def test_kill_on_completed_shell_does_not_signal_group(registry, monkeypatch):
"""A completed shell's pgid is a stale snapshot the OS may have recycled
to an unrelated process group — kill() must not signal it (the waiter's
own group kill already ran at exit, when the pgid was fresh)."""
import turnstone.core.background_shells as bg_mod
shell = registry.spawn("true")
assert _wait_status(shell, "completed")
calls = []
monkeypatch.setattr(bg_mod.os, "killpg", lambda *a: calls.append(a))
killed = registry.kill(shell.shell_id)
assert calls == [], "killpg must not fire for an already-exited shell"
assert killed.status == "completed", "a natural exit must not be relabelled 'killed'"
def test_close_is_time_bounded_with_pipe_holding_escapee(registry, tmp_path):
"""An escaped-group grandchild that holds the output pipes wedges the
drain threads. close() must still return within its total budget —
it can run under the server's async close route, where an unbounded
join would freeze the whole node's event loop."""
pidfile = tmp_path / "holder.pid"
# ``setsid`` puts the sleep in a NEW session (outside our kill group)
# while it still inherits our stdout/stderr pipes — the accepted
# leaked-daemon case from the module docstring.
shell = registry.spawn(f"setsid sleep 60 & echo $! > {pidfile}; echo started")
assert _wait_until(pidfile.exists)
holder_pid = int(pidfile.read_text().strip())
try:
start = time.monotonic()
registry.close()
elapsed = time.monotonic() - start
assert elapsed < 8, f"close() took {elapsed:.1f}s — teardown must be budget-bounded"
finally:
_kill_pid(holder_pid)
# The holder is dead, so the wedged drains EOF promptly; wait for
# them here so the conftest leak guard sees a clean teardown.
assert _wait_until(lambda: not any(t.is_alive() for t in shell._threads))
def test_exited_records_are_pruned_at_cap():
reg = BackgroundShellRegistry(max_exited_records=2)
try:
shells = [reg.spawn(f"echo job-{i}") for i in range(3)]
for s in shells:
assert _wait_status(s, "completed")
# Eviction happens on each exit; poll until the oldest is gone
# (waiter threads race, prune runs per-exit).
assert _wait_until(lambda: not reg.has(shells[0].shell_id))
assert reg.has(shells[1].shell_id)
assert reg.has(shells[2].shell_id)
with pytest.raises(UnknownShellError):
reg.read(shells[0].shell_id)
finally:
reg.close()
def test_catastrophic_filter_times_out_without_consuming(registry):
"""A backtracking-bomb filter must error within the bound and consume
NOTHING — the retry without a filter still gets the output. The match
runs in a killable child process: sre holds the GIL, so an in-process
bomb would freeze the whole interpreter, watchdogs included."""
# One ~3000-char line of a's ending in 'b' — the classic (a+)+$ bomb
# subject — followed by a sentinel line.
shell = registry.spawn("printf 'a%.0s' $(seq 1 3000); echo b; echo tail-line")
assert _wait_status(shell, "completed")
start = time.monotonic()
with pytest.raises(FilterTimeoutError):
registry.read(shell.shell_id, filter_pattern=r"(a+)+$")
assert time.monotonic() - start < 10, "filter timeout must be bounded"
# Nothing was consumed: an unfiltered read sees the whole delta.
read = registry.read(shell.shell_id)
assert any("tail-line" in ln for ln in read.lines)
def test_overlong_filter_pattern_is_rejected(registry):
shell = registry.spawn("echo hi")
assert _wait_status(shell, "completed")
with pytest.raises(re.error):
registry.read(shell.shell_id, filter_pattern="x" * 600)
def test_cap_error_is_owner_scope_honest():
"""The cap is registry-wide, but the advice must only name shells the
caller can actually kill — kill_shell is owner-scoped."""
reg = BackgroundShellRegistry(max_shells=1)
try:
reg.spawn("sleep 30") # main scope fills the cap
with pytest.raises(TooManyShellsError) as excinfo:
reg.spawn("sleep 30", owner="agent-1")
msg = str(excinfo.value)
assert "bash_1" not in msg, "must not advise killing another scope's shell"
assert "other agents" in msg
# The same-scope variant names the killable shell.
with pytest.raises(TooManyShellsError) as excinfo2:
reg.spawn("sleep 30")
assert "bash_1" in str(excinfo2.value)
assert "kill_shell" in str(excinfo2.value)
finally:
reg.close()
def test_prune_evicts_by_exit_order_not_spawn_order():
"""A long-lived first-spawned server must never be evicted by its OWN
exit's prune once enough later jobs have finished — eviction follows
exit order, so the just-exited shell is always the newest record."""
reg = BackgroundShellRegistry(max_exited_records=2)
try:
server = reg.spawn("sleep 30") # bash_1, exits LAST
jobs = [reg.spawn(f"echo job-{i}") for i in range(3)]
for job in jobs:
assert _wait_status(job, "completed")
reg.kill(server.shell_id)
assert reg.has(server.shell_id), "the just-exited shell must survive its own exit's prune"
# The earliest-EXITED job is the eviction victim, not bash_1.
assert _wait_until(lambda: len(reg.shells()) <= 3)
assert reg.read(server.shell_id).status == "killed"
finally:
reg.close()
def test_thread_start_failure_leaves_no_orphan_record(registry, monkeypatch, tmp_path):
"""If Thread.start raises (thread exhaustion), the record must be
unregistered and the fresh group reaped — an orphan with never-started
Thread objects would make every later close()/reap() join raise and
abort session teardown."""
import turnstone.core.background_shells as bg_mod
pidfile = tmp_path / "leader.pid"
real_thread = bg_mod.threading.Thread
class FailingWaiterThread(real_thread):
def start(self):
if "bg-shell-wait" in (self.name or ""):
raise RuntimeError("can't start new thread")
super().start()
monkeypatch.setattr(bg_mod.threading, "Thread", FailingWaiterThread)
with pytest.raises(RuntimeError):
registry.spawn(f"echo $$ > {pidfile}; sleep 60")
assert registry.shells() == [], "failed spawn must not strand a record"
if pidfile.exists():
leader_pid = int(pidfile.read_text().strip())
assert _wait_until(lambda: not _pid_alive(leader_pid)), "fresh group leaked"
monkeypatch.undo()
registry.close() # must not raise on the (empty) registry
def test_filter_helper_failure_reports_exec_error_not_timeout(registry, monkeypatch):
"""A crashed helper must not tell the model its (fine) pattern was too
slow — and must not consume the delta."""
import turnstone.core.background_shells as bg_mod
from turnstone.core.background_shells import FilterExecError
shell = registry.spawn("echo hello")
assert _wait_status(shell, "completed")
monkeypatch.setattr(bg_mod.sys, "executable", "/bin/false")
with pytest.raises(FilterExecError) as excinfo:
registry.read(shell.shell_id, filter_pattern="hello")
assert "not a problem with your pattern" in str(excinfo.value)
monkeypatch.undo()
read = registry.read(shell.shell_id)
assert [ln.strip() for ln in read.lines] == ["hello"]
def test_filter_matches_only_within_line_cap_and_reports_clipping(registry):
"""Lines are truncated parent-side before shipping to the helper: a
match beyond the per-line cap is not found (a filter targets log
lines), and a huge retained line cannot burn the time budget on I/O.
The clipping is NEVER silent — the read reports how many lines were
only partially visible to the pattern."""
shell = registry.spawn("printf 'x%.0s' $(seq 1 5000); echo needle-suffix")
assert _wait_status(shell, "completed")
read = registry.read(shell.shell_id, filter_pattern="needle")
assert read.lines == []
assert read.new_line_count == 1
assert read.clipped_lines == 1
def test_concurrent_reads_never_double_deliver(registry):
"""Two simultaneous reads of one shell must SPLIT the delta between
them, never both return it — the whole pass (snapshot → commit)
serializes per shell. Without that, a parallel tool batch reading the
same handle gets every line twice."""
shell = registry.spawn("seq 1 200")
assert _wait_status(shell, "completed")
results: list[list[str]] = [[], []]
barrier = threading.Barrier(2)
def _reader(slot: int) -> None:
barrier.wait()
results[slot] = [ln.strip() for ln in registry.read(shell.shell_id).lines]
threads = [threading.Thread(target=_reader, args=(i,)) for i in range(2)]
for t in threads:
t.start()
for t in threads:
t.join(timeout=10)
combined = results[0] + results[1]
assert len(combined) == 200, f"expected each line exactly once, got {len(combined)}"
assert sorted(combined, key=int) == [str(i) for i in range(1, 201)]
def test_filter_helper_spawn_failure_is_exec_error(registry, monkeypatch):
"""A helper that fails to LAUNCH (fork pressure) must land in the same
honest FilterExecError as a crashed helper — not escape as a raw
OSError blaming nothing — and must not consume the delta."""
import turnstone.core.background_shells as bg_mod
from turnstone.core.background_shells import FilterExecError
shell = registry.spawn("echo hello")
assert _wait_status(shell, "completed")
def _boom(*args, **kwargs):
raise BlockingIOError("Resource temporarily unavailable")
monkeypatch.setattr(bg_mod.subprocess, "Popen", _boom)
with pytest.raises(FilterExecError):
registry.read(shell.shell_id, filter_pattern="hello")
monkeypatch.undo()
read = registry.read(shell.shell_id)
assert [ln.strip() for ln in read.lines] == ["hello"]
def test_on_exit_exception_does_not_wedge_the_shell():
def _boom(shell):
raise RuntimeError("callback bug")
reg = BackgroundShellRegistry(on_exit=_boom)
try:
shell = reg.spawn("echo hi")
# The waiter thread must survive the callback raising: status still
# lands and output is still readable.
assert _wait_status(shell, "completed")
assert [ln.strip() for ln in reg.read(shell.shell_id).lines] == ["hi"]
finally:
reg.close()
+797
View File
@@ -0,0 +1,797 @@
"""Session-level tests for the background-shell tool surface (#817).
Covers the wiring around :class:`BackgroundShellRegistry`:
* ``bash`` gains ``run_in_background: true`` (alias ``is_background``) —
same approval gate, returns immediately with a ``bash_N`` handle.
* ``bash_output`` — auto-approved delta reader (status + exit code + only
new output since the last call, optional ``filter`` regex).
* ``kill_shell`` — auto-approved kill of a registered shell's whole group.
* Exit notices ride the NudgeQueue on channel ``"any"`` (the watch rail) so
they drain at the next seam and can wake an idle workstream.
* Lifecycle: ``close()`` reaps everything; generation-``cancel()`` does NOT
(a deliberately-detached server survives a stopped turn); shells spawned
inside a task_agent are owner-scoped and reaped when the agent finishes.
"""
import time
import pytest
from tests._proc_helpers import pid_alive as _pid_alive
from tests._proc_helpers import poll_until as _wait_until
from tests._session_helpers import make_session
@pytest.fixture
def session():
s = make_session()
yield s
s.close()
def _start_background(session, command, call_id="bg1", **extra_args):
"""Prepare + execute a backgrounded bash call; return the result text."""
args = {"command": command, "run_in_background": True, **extra_args}
prepared = session._prepare_bash(call_id, args)
assert "error" not in prepared, prepared.get("error")
_cid, output = prepared["execute"](prepared)
return output
def _only_shell(session):
shells = session._background_shells.shells()
assert len(shells) == 1
return shells[0]
# ---------------------------------------------------------------------------
# bash: run_in_background routing
# ---------------------------------------------------------------------------
def test_prepare_bash_background_keeps_approval_gate(session):
prepared = session._prepare_bash("c1", {"command": "sleep 30", "run_in_background": True})
assert prepared["needs_approval"] is True
assert prepared["approval_label"] == "bash"
def test_prepare_bash_background_header_says_background(session):
prepared = session._prepare_bash("c1", {"command": "sleep 30", "run_in_background": True})
assert "background" in prepared["header"]
def test_background_bash_returns_immediately_with_handle(session):
start = time.monotonic()
output = _start_background(session, "sleep 30")
elapsed = time.monotonic() - start
assert elapsed < 5, f"backgrounded call blocked for {elapsed:.1f}s"
assert "bash_1" in output
shell = _only_shell(session)
assert shell.status == "running"
assert _pid_alive(shell.pid)
def test_background_start_mentions_reader_and_killer(session):
"""The immediate result must teach the follow-up tools — weak-prior
models (GPT-5.6) only reach for the poll pattern if the result names it."""
output = _start_background(session, "sleep 30")
assert "bash_output" in output
assert "kill_shell" in output
def test_is_background_alias_accepted(session):
output = _start_background(session, "sleep 30", is_background=True)
assert "bash_1" in output
assert _only_shell(session).status == "running"
def test_foreground_bash_routing_unchanged(session):
prepared = session._prepare_bash("c1", {"command": "echo hi"})
assert prepared["execute"] == session._exec_bash
prepared_false = session._prepare_bash("c2", {"command": "echo hi", "run_in_background": False})
assert prepared_false["execute"] == session._exec_bash
def test_background_respects_command_blocklist(session):
prepared = session._prepare_bash("c1", {"command": "shutdown now", "run_in_background": True})
assert "error" in prepared
assert session._background_shells.shells() == []
def test_background_ignores_timeout(session):
"""No bounded wait exists to time out — a 1s timeout must not kill the
detached shell."""
_start_background(session, "sleep 30", timeout=1)
shell = _only_shell(session)
time.sleep(1.5)
assert shell.status == "running"
assert _pid_alive(shell.pid)
def test_background_spawn_failure_reports_error(session, monkeypatch):
from turnstone.core import background_shells as bg_mod
def _boom(*args, **kwargs):
raise OSError("cannot fork")
monkeypatch.setattr(bg_mod.subprocess, "Popen", _boom)
prepared = session._prepare_bash("c1", {"command": "echo hi", "run_in_background": True})
_cid, output = prepared["execute"](prepared)
assert "cannot fork" in output
def test_too_many_background_shells_reports_error(session, monkeypatch):
monkeypatch.setattr(session._background_shells, "_max_shells", 1)
_start_background(session, "sleep 30", call_id="bg1")
output = _start_background(session, "sleep 30", call_id="bg2")
assert "bash_1" in output # the live shell is named so the model can kill it
assert len(session._background_shells.shells()) == 1
# ---------------------------------------------------------------------------
# bash_output
# ---------------------------------------------------------------------------
def test_bash_output_is_auto_approved(session):
prepared = session._prepare_bash_output("c1", {"id": "bash_1"})
assert prepared["needs_approval"] is False
def test_bash_output_missing_id_errors(session):
prepared = session._prepare_bash_output("c1", {})
assert "error" in prepared
def test_bash_output_returns_delta_then_no_new_output(session):
_start_background(session, "echo hello; sleep 30")
shell = _only_shell(session)
assert _wait_until(lambda: shell.status == "running")
def _read():
prepared = session._prepare_bash_output("r", {"id": shell.shell_id})
assert "error" not in prepared
return prepared["execute"](prepared)[1]
assert _wait_until(lambda: "hello" in _read())
again = _read()
assert "hello" not in again
assert "no new output" in again.lower()
assert "running" in again.lower()
def test_bash_output_reports_exit_code_when_completed(session):
_start_background(session, "exit 3")
shell = _only_shell(session)
assert _wait_until(lambda: shell.status == "completed")
prepared = session._prepare_bash_output("r", {"id": shell.shell_id})
_cid, output = prepared["execute"](prepared)
assert "completed" in output.lower()
assert "3" in output
def test_bash_output_filter_applies(session):
_start_background(session, "echo match-a; echo skip-b")
shell = _only_shell(session)
assert _wait_until(lambda: shell.status == "completed")
prepared = session._prepare_bash_output("r", {"id": shell.shell_id, "filter": "^match"})
_cid, output = prepared["execute"](prepared)
assert "match-a" in output
assert "skip-b" not in output
def test_bash_output_invalid_filter_reports_error(session):
_start_background(session, "sleep 30")
shell = _only_shell(session)
prepared = session._prepare_bash_output("r", {"id": shell.shell_id, "filter": "[bad"})
_cid, output = prepared["execute"](prepared)
assert "regex" in output.lower() or "filter" in output.lower()
def test_bash_output_unknown_id_lists_live_shells(session):
_start_background(session, "sleep 30")
prepared = session._prepare_bash_output("r", {"id": "bash_42"})
_cid, output = prepared["execute"](prepared)
assert "bash_42" in output
assert "bash_1" in output
# ---------------------------------------------------------------------------
# kill_shell
# ---------------------------------------------------------------------------
def test_kill_shell_is_auto_approved(session):
prepared = session._prepare_kill_shell("c1", {"id": "bash_1"})
assert prepared["needs_approval"] is False
def test_kill_shell_missing_id_errors(session):
prepared = session._prepare_kill_shell("c1", {})
assert "error" in prepared
def test_kill_shell_kills_and_reports(session):
_start_background(session, "sleep 60")
shell = _only_shell(session)
prepared = session._prepare_kill_shell("k", {"id": shell.shell_id})
_cid, output = prepared["execute"](prepared)
assert "killed" in output.lower()
assert _wait_until(lambda: not _pid_alive(shell.pid))
def test_kill_shell_unknown_id_reports_error(session):
prepared = session._prepare_kill_shell("k", {"id": "bash_9"})
_cid, output = prepared["execute"](prepared)
assert "bash_9" in output
# ---------------------------------------------------------------------------
# Exit notices (NudgeQueue, channel "any", wake)
# ---------------------------------------------------------------------------
def test_natural_exit_enqueues_any_channel_notice(session):
_start_background(session, "echo done")
assert _wait_until(
lambda: any(t == "background_shell_exit" for t, _ in session._nudge_queue.pending())
)
entries = session._nudge_queue.pending(channel="any")
texts = [text for t, text in entries if t == "background_shell_exit"]
assert texts, "notice must ride channel 'any' so it can wake an idle workstream"
assert "bash_1" in texts[0]
assert "bash_output" in texts[0]
def test_exit_notice_carries_metadata(session):
_start_background(session, "exit 5")
assert _wait_until(
lambda: any(t == "background_shell_exit" for t, _ in session._nudge_queue.pending())
)
metadata = [
meta
for t, _text, meta in session._nudge_queue.pending_with_metadata()
if t == "background_shell_exit"
][0]
assert metadata["shell_id"] == "bash_1"
assert metadata["exit_code"] == 5
def test_exit_notice_triggers_wake_fn(session):
wakes = []
session._watch_wake_fn = lambda: wakes.append(1)
_start_background(session, "echo done")
assert _wait_until(lambda: wakes), "natural exit must wake an idle workstream"
def test_kill_shell_suppresses_exit_notice(session):
_start_background(session, "sleep 60")
shell = _only_shell(session)
prepared = session._prepare_kill_shell("k", {"id": shell.shell_id})
prepared["execute"](prepared)
assert _wait_until(lambda: not _pid_alive(shell.pid))
time.sleep(0.3) # a buggy late notice would land within this window
assert not any(t == "background_shell_exit" for t, _ in session._nudge_queue.pending())
def test_close_drops_pending_exit_notice_via_valid_until(session):
"""A notice for a shell that no longer exists (registry closed) must not
deliver — the valid_until predicate drops it at drain time."""
_start_background(session, "echo done")
assert _wait_until(
lambda: any(t == "background_shell_exit" for t, _ in session._nudge_queue.pending())
)
session.close()
from turnstone.core.nudge_queue import USER_DRAIN
drained = session._nudge_queue.drain(USER_DRAIN)
assert not any(t == "background_shell_exit" for t, _text, _m in drained)
# ---------------------------------------------------------------------------
# Lifecycle
# ---------------------------------------------------------------------------
def test_close_reaps_background_shells(session):
_start_background(session, "sleep 60")
shell = _only_shell(session)
session.close()
assert not _pid_alive(shell.pid)
def test_generation_cancel_does_not_reap_background_shells(session):
"""cancel() fires on mere stop-generation — a deliberately-detached
server must survive it. Only close()/kill_shell end it."""
_start_background(session, "sleep 60")
shell = _only_shell(session)
session.cancel()
time.sleep(0.3)
assert _pid_alive(shell.pid), "generation cancel must not kill detached shells"
# ---------------------------------------------------------------------------
# Review-hardening regressions (#817 code review)
# ---------------------------------------------------------------------------
def test_string_typed_background_flag_is_honored(session):
"""Providers intermittently send booleans as strings; 'true' must not
silently fall through to the foreground executor (where the group kill
would reap the server the model believed it detached)."""
for call_id, args in (
("s1", {"command": "sleep 30", "run_in_background": "true"}),
("s2", {"command": "sleep 30", "is_background": "True"}),
):
prepared = session._prepare_bash(call_id, args)
assert prepared["execute"] == session._exec_bash_background, args
def test_kill_shell_on_completed_shell_reports_already_exited(session):
_start_background(session, "true")
shell = _only_shell(session)
assert _wait_until(lambda: shell.status == "completed")
prepared = session._prepare_kill_shell("k", {"id": shell.shell_id})
_cid, output = prepared["execute"](prepared)
assert "already exited" in output.lower()
def test_exit_notice_survives_generation_abandon_without_waking(session):
"""cancel/interrupt/exception clear generation-scoped advisories, but an
external event (a background shell exited) still happened — its notice
must survive to the next seam or the model keeps talking to a dead
server. It survives DEMOTED to 'quiet': still deliverable, but no
longer wake-eligible, so the workstream the user just stopped cannot
resume itself over it."""
from turnstone.core.nudge_queue import USER_DRAIN, WAKE_PENDING
_start_background(session, "echo done")
assert _wait_until(
lambda: any(t == "background_shell_exit" for t, _ in session._nudge_queue.pending())
)
session._queue_tool_advisory("tool_error", "3 consecutive tool errors")
session._drain_pending_advisories()
kinds = [t for t, _ in session._nudge_queue.pending()]
assert "background_shell_exit" in kinds
assert "tool_error" not in kinds
# Post-cancel quiescence: nothing is wake-eligible...
assert not session._nudge_queue.has_pending(WAKE_PENDING)
# ...yet the notice still delivers at the next legitimate seam.
drained = session._nudge_queue.drain(USER_DRAIN)
assert any(t == "background_shell_exit" for t, _x, _m in drained)
def test_int_typed_background_flag_is_honored(session):
prepared = session._prepare_bash("i1", {"command": "sleep 30", "run_in_background": 1})
assert prepared["execute"] == session._exec_bash_background
prepared_zero = session._prepare_bash("i2", {"command": "echo hi", "run_in_background": 0})
assert prepared_zero["execute"] == session._exec_bash
def test_bash_output_non_string_filter_errors_without_consuming(session):
_start_background(session, "echo hello; sleep 30")
shell = _only_shell(session)
prepared = session._prepare_bash_output("r", {"id": shell.shell_id, "filter": 123})
assert "error" in prepared
assert "filter" in prepared["error"].lower()
# Nothing was consumed by the refused call.
assert _wait_until(lambda: shell.unread_lines > 0)
def test_filter_timeout_reports_error_without_consuming(session, monkeypatch):
from turnstone.core.background_shells import FilterTimeoutError
_start_background(session, "sleep 30")
shell = _only_shell(session)
def _boom(*a, **kw):
raise FilterTimeoutError("filter regex took longer than 2s to run")
monkeypatch.setattr(session._background_shells, "read", _boom)
prepared = session._prepare_bash_output("r", {"id": shell.shell_id, "filter": "(a+)+$"})
_cid, output = prepared["execute"](prepared)
assert "filter" in output.lower()
assert "error" in output.lower()
def test_registries_are_isolated_per_session():
"""Workstream isolation: a handle from one session must be unresolvable
from another — buffers, ids, and kills never cross ChatSessions."""
session_a = make_session()
session_b = make_session()
try:
_start_background(session_a, "sleep 30")
shell_a = _only_shell(session_a)
read_b = session_b._prepare_bash_output("r", {"id": shell_a.shell_id})
_cid, output = read_b["execute"](read_b)
assert "no background shell" in output.lower()
kill_b = session_b._prepare_kill_shell("k", {"id": shell_a.shell_id})
_cid, kill_output = kill_b["execute"](kill_b)
assert "no background shell" in kill_output.lower()
assert _pid_alive(shell_a.pid), "another session must not be able to kill the shell"
finally:
session_a.close()
session_b.close()
def test_bash_output_polling_is_repeat_exempt(session):
"""Repeated identical bash_output calls ARE the documented monitoring
pattern — the repeat detector must not brand them 'identical repeat'
(the delta result differs by construction) nor queue a repeat nudge."""
import json as _json
_start_background(session, "sleep 30")
shell = _only_shell(session)
args = _json.dumps({"id": shell.shell_id})
for i in range(5):
tool_calls = [{"id": f"t{i}", "function": {"name": "bash_output", "arguments": args}}]
results = [(f"t{i}", "bash_1 (running)\nNo new output since the last read.")]
session._apply_post_execute_advisories(tool_calls, results)
assert "identical repeat" not in results[0][1]
assert not any(t == "repeat" for t, _ in session._nudge_queue.pending())
def test_repeat_exempt_calls_still_break_other_streaks(session):
"""The exemption suppresses the WARNING, not the recording: a
bash_output poll interleaved between identical bash calls must reset
the bash streak — otherwise the documented monitor-and-probe loop
(poll, curl health, poll, curl health…) draws a false 'identical
repeat' on the probe."""
import json as _json
_start_background(session, "sleep 30")
shell = _only_shell(session)
poll_args = _json.dumps({"id": shell.shell_id})
probe_args = _json.dumps({"command": "curl -s localhost:8080/health"})
for i in range(6):
probe = [{"id": f"p{i}", "function": {"name": "bash", "arguments": probe_args}}]
probe_results = [(f"p{i}", "ok")]
session._apply_post_execute_advisories(probe, probe_results)
assert "identical repeat" not in probe_results[0][1], (
"interleaved probes are not a stuck loop"
)
poll = [{"id": f"q{i}", "function": {"name": "bash_output", "arguments": poll_args}}]
session._apply_post_execute_advisories(poll, [(f"q{i}", "no new output")])
def test_bash_repeats_still_warn(session):
"""The exemption is bash_output-specific: a genuinely stuck identical
bash loop still gets the warning."""
import json as _json
args = _json.dumps({"command": "echo test"})
warned = False
for i in range(5):
tool_calls = [{"id": f"b{i}", "function": {"name": "bash", "arguments": args}}]
results = [(f"b{i}", "test")]
session._apply_post_execute_advisories(tool_calls, results)
warned = warned or "identical repeat" in results[0][1]
assert warned
def test_quiet_only_entries_do_not_trigger_wake_delivery(session, monkeypatch):
"""A dispatched wake whose wake-eligible entries all evaporated must be
a no-op: quiet entries alone never resume a stopped workstream, and
they stay queued for the next legitimate seam."""
calls = []
monkeypatch.setattr(session, "send", lambda *a, **k: calls.append(1))
session._nudge_queue.enqueue("background_shell_exit", "old news", "quiet")
session.deliver_wake_nudge_from_queue()
assert calls == []
assert session._nudge_queue.pending(channel="quiet") == [("background_shell_exit", "old news")]
def test_wake_delivers_quiet_alongside_eligible_in_insertion_order(session, monkeypatch):
"""Quiet entries ride the wake AND cross-channel chronology holds: an
older demoted notice renders before the newer fire that earned the
wake (a poll counter must never run backwards)."""
seen = {}
def _fake_send(*a, **k):
seen["reminders"] = list(session._wake_drained_reminders or [])
session._wake_drained_reminders = None # emulate emission consuming
monkeypatch.setattr(session, "send", _fake_send)
session._nudge_queue.enqueue("background_shell_exit", "old", "quiet")
session._nudge_queue.enqueue("watch_triggered", "new", "any")
session.deliver_wake_nudge_from_queue()
types = [e["type"] for e in seen["reminders"]]
assert types == ["background_shell_exit", "watch_triggered"], (
"older quiet entry must precede the newer wake-eligible one"
)
assert session._nudge_queue.pending() == []
def test_failed_wake_reenqueue_preserves_valid_until(session, monkeypatch):
"""The re-enqueued notice keeps its staleness predicate — a stale
notice re-queued by a failed wake must still be droppable at its next
drain, not delivered against a gone shell."""
from turnstone.core.nudge_queue import USER_DRAIN
alive = {"value": True}
def _fail(*a, **k):
raise RuntimeError("storage down")
monkeypatch.setattr(session, "send", _fail)
session._nudge_queue.enqueue(
"background_shell_exit",
"server died",
"any",
valid_until=lambda: alive["value"],
)
with pytest.raises(RuntimeError):
session.deliver_wake_nudge_from_queue()
assert session._nudge_queue.pending(channel="quiet"), "notice must be re-queued"
alive["value"] = False # the shell record is gone now
drained = session._nudge_queue.drain(USER_DRAIN)
assert drained == [], "stale re-queued notice must drop via its predicate"
def test_mid_emit_failure_restashes_unemitted_tail(session, monkeypatch):
"""A failure while emitting reminder k of n must leave k..n recoverable
— the wake caller's finally re-enqueues them instead of losing the
suffix."""
calls = {"n": 0}
def _append(source, text, **meta):
calls["n"] += 1
if calls["n"] == 2:
raise RuntimeError("storage down")
monkeypatch.setattr(session, "_append_system_turn", _append)
session._wake_drained_reminders = [
{"type": "a", "text": "1"},
{"type": "b", "text": "2"},
{"type": "c", "text": "3"},
]
with pytest.raises(RuntimeError):
session._emit_pending_user_nudges()
assert session._wake_drained_reminders == [
{"type": "b", "text": "2"},
{"type": "c", "text": "3"},
]
def test_failed_wake_reenqueues_undelivered_as_quiet(session, monkeypatch):
"""A wake send that dies before emitting its drained reminders must not
eat them — a shell's exit notice fires exactly once."""
def _fail(*a, **k):
raise RuntimeError("storage down")
monkeypatch.setattr(session, "send", _fail)
session._nudge_queue.enqueue(
"background_shell_exit", "server died", "any", metadata={"shell_id": "bash_1"}
)
with pytest.raises(RuntimeError):
session.deliver_wake_nudge_from_queue()
pending = session._nudge_queue.pending_with_metadata(channel="quiet")
assert [(t, x) for t, x, _m in pending] == [("background_shell_exit", "server died")]
assert pending[0][2] == {"shell_id": "bash_1"}
def test_failed_wake_preserves_chronology_and_stays_wake_quiescent(session, monkeypatch):
"""Failed-wake recovery invariants: (a) the re-queued external notice
keeps its seq, so the retry renders it BEFORE a newer event that
arrived during the failure; (b) NOTHING wake-eligible remains after
the failure — external notices demote to quiet and user-channel
advisories are dropped outright, because a re-armed WAKE_PENDING gate
plus the zero-backoff worker-exit retry would respawn wake workers in
an unbounded hot loop against a persistent failure."""
from turnstone.core.nudge_queue import WAKE_PENDING
calls = {"n": 0}
seen = {}
def _send(*a, **k):
calls["n"] += 1
if calls["n"] == 1:
raise RuntimeError("transient storage failure")
seen["reminders"] = list(session._wake_drained_reminders or [])
session._wake_drained_reminders = None
monkeypatch.setattr(session, "send", _send)
session._nudge_queue.enqueue("watch_triggered", "poll-4", "any")
session._nudge_queue.enqueue("correction", "user advisory", "user")
with pytest.raises(RuntimeError):
session.deliver_wake_nudge_from_queue()
# (b) bounded: nothing left that could re-trigger the wake gate.
assert not session._nudge_queue.has_pending(WAKE_PENDING), (
"a failed wake must not leave wake-eligible entries (respawn hot loop)"
)
assert [t for t, _x in session._nudge_queue.pending(channel="quiet")] == ["watch_triggered"]
# A NEWER event lands after the failure...
session._nudge_queue.enqueue("watch_triggered", "poll-5", "any")
session.deliver_wake_nudge_from_queue()
texts = [e["text"] for e in seen["reminders"]]
# (a) ...and the retry renders old-before-new despite the round trip.
assert texts.index("poll-4") < texts.index("poll-5")
def test_exit_notice_emits_end_to_end_as_system_turn(session):
"""THE test whose absence hid an undeliverable notice for six review
rounds: drive the notice through REAL emission (make_system_turn +
_append_system_turn), not just queue assertions — an unregistered
``_source`` raises ValueError only at this layer."""
_start_background(session, "echo done")
assert _wait_until(
lambda: any(t == "background_shell_exit" for t, _ in session._nudge_queue.pending())
)
from turnstone.core.trajectory import Role
before = len(session.messages)
session._emit_pending_user_nudges() # must not raise
new_turns = session.messages[before:]
assert any(
turn.role is Role.SYSTEM and turn.source == "background_shell_exit" for turn in new_turns
), f"exit notice must land as a first-class system turn, got {new_turns!r}"
def test_cli_exit_closes_every_loaded_session():
"""CLI exit must reap background shells in EVERY workstream, not just
the active one — a server started before /new must not outlive /exit."""
from unittest.mock import MagicMock
from turnstone.cli import _close_all_sessions
ws_a, ws_b, ws_never_loaded = MagicMock(), MagicMock(), MagicMock()
ws_never_loaded.session = None
ws_a.session.close.side_effect = RuntimeError("bad teardown")
manager = MagicMock()
manager.list_all.return_value = [ws_a, ws_b, ws_never_loaded]
_close_all_sessions(manager) # must not raise
ws_a.session.close.assert_called_once()
ws_b.session.close.assert_called_once(), "one bad teardown must not stop the rest"
# Signal phase ran for every loaded session, before any close.
ws_a.session._background_shells.signal_all.assert_called_once()
ws_b.session._background_shells.signal_all.assert_called_once()
def test_cli_exit_ctrl_c_does_not_abort_the_reap():
"""Ctrl-C during the close phase must not escape the helper: the kill
signals already landed on every session in phase 1, and an escaping
KeyboardInterrupt would also skip MCP/registry shutdown in main()."""
from unittest.mock import MagicMock
from turnstone.cli import _close_all_sessions
ws_a, ws_b = MagicMock(), MagicMock()
ws_a.session.close.side_effect = KeyboardInterrupt
manager = MagicMock()
manager.list_all.return_value = [ws_a, ws_b]
_close_all_sessions(manager) # must not raise
ws_a.session._background_shells.signal_all.assert_called_once()
(
ws_b.session._background_shells.signal_all.assert_called_once(),
("signals must land on every session before the interruptible close phase"),
)
def test_non_string_reminder_text_drops_silently(session):
"""A dict reminder with non-str text must drop at the rail, not
TypeError out of the dispatch closure (WatchRunner would re-fire the
row every tick)."""
runner = type(
"R",
(),
{
"set_dispatch_fn": lambda self, ws, fn: None,
"remove_dispatch_fn": lambda self, ws, owner=None: None,
},
)()
session.set_watch_runner(runner)
session._watch_dispatch_fn({"text": 123, "watch_name": "w"}, "watch-1") # must not raise
assert session._nudge_queue.pending() == []
def test_string_typed_stop_on_error_is_honored(session):
"""One coercion dialect for every bash boolean: a string-typed
stop_on_error must add set -e in both branches, not silently drop it."""
fg = session._prepare_bash("f1", {"command": "echo hi", "stop_on_error": "true"})
assert fg["stop_on_error"] is True
bg = session._prepare_bash(
"b1", {"command": "echo hi", "run_in_background": True, "stop_on_error": "true"}
)
assert bg["stop_on_error"] is True
def test_non_dict_watch_reminder_drops_silently(session):
"""The rebuilt dispatch closure must drop a non-dict reminder like the
old code did — a TypeError would make WatchRunner hold and re-fire the
row every tick."""
runner = type(
"R",
(),
{
"set_dispatch_fn": lambda self, ws, fn: None,
"remove_dispatch_fn": lambda self, ws, owner=None: None,
},
)()
session.set_watch_runner(runner)
dispatch = session._watch_dispatch_fn
dispatch("not a dict", "watch-1") # must not raise
assert session._nudge_queue.pending() == []
def test_truthy_flag_dialect_is_unified():
"""One coercion dialect file-wide — 'on' and nonzero numbers count, so a
provider quirk honored on coordinator tools is honored on bash too."""
from turnstone.core.session import _is_truthy_flag
assert _is_truthy_flag(True)
assert _is_truthy_flag("on")
assert _is_truthy_flag(2)
assert not _is_truthy_flag("off")
assert not _is_truthy_flag(0)
assert not _is_truthy_flag(None)
assert not _is_truthy_flag(False)
def test_bash_output_notes_clipped_lines_under_filter(session):
_start_background(session, "printf 'x%.0s' $(seq 1 5000); echo tail")
shell = _only_shell(session)
assert _wait_until(lambda: shell.status == "completed")
prepared = session._prepare_bash_output("r", {"id": shell.shell_id, "filter": "zzz"})
_cid, output = prepared["execute"](prepared)
assert "partially visible" in output
# ---------------------------------------------------------------------------
# task_agent scoping
# ---------------------------------------------------------------------------
def test_task_agent_shells_are_owner_scoped_and_reaped(session, monkeypatch):
seen = {}
def fake_run_agent(agent_turns, label="task", **kwargs):
out = _start_background(session, "sleep 60", call_id="sub-bash")
seen["start_output"] = out
agent_shells = session._background_shells.shells(owner="task-1")
seen["agent_shells"] = list(agent_shells)
seen["pid"] = agent_shells[0].pid if agent_shells else None
# The sub-agent's shell is invisible to the main scope.
seen["visible_to_parent"] = [s.shell_id for s in session._background_shells.shells()]
return "agent done"
monkeypatch.setattr(session, "_run_agent", fake_run_agent)
call_id, result = session._exec_task({"call_id": "task-1", "prompt": "start a server"})
assert "agent done" in result
assert seen["agent_shells"], "shell spawned inside the agent must carry its owner"
# Scope honesty in the start message: the sub-agent must not promise its
# caller a server that dies the moment it returns.
assert "terminated when the agent finishes" in seen["start_output"]
assert seen["visible_to_parent"] == []
assert seen["pid"] is not None
assert _wait_until(lambda: not _pid_alive(seen["pid"])), (
"sub-agent shells must be reaped when the agent finishes"
)
def test_task_agent_cannot_touch_parent_shells(session, monkeypatch):
_start_background(session, "sleep 60", call_id="parent-bash")
parent_shell = _only_shell(session)
seen = {}
def fake_run_agent(agent_turns, label="task", **kwargs):
prepared = session._prepare_bash_output("r", {"id": parent_shell.shell_id})
seen["read_output"] = prepared["execute"](prepared)[1]
prepared_kill = session._prepare_kill_shell("k", {"id": parent_shell.shell_id})
seen["kill_output"] = prepared_kill["execute"](prepared_kill)[1]
return "done"
monkeypatch.setattr(session, "_run_agent", fake_run_agent)
session._exec_task({"call_id": "task-1", "prompt": "snoop"})
assert "no background shell" in seen["read_output"].lower()
assert "no background shell" in seen["kill_output"].lower()
assert _pid_alive(parent_shell.pid), "agent must not be able to kill a parent shell"
def test_parent_scope_restored_after_task_agent(session, monkeypatch):
monkeypatch.setattr(session, "_run_agent", lambda *a, **k: "done")
session._exec_task({"call_id": "task-1", "prompt": "noop"})
output = _start_background(session, "sleep 30", call_id="after-task")
assert "bash_1" in output
assert _only_shell(session).owner is None
+2 -18
View File
@@ -9,31 +9,15 @@ EOF) bounded by ``tool_timeout`` and kills the whole session group on exit, so
the call always returns and never leaks the background child.
"""
import contextlib
import os
import signal
import threading
import time
from tests._proc_helpers import kill_pid as _kill_pid
from tests._proc_helpers import pid_alive as _pid_alive
from tests._session_helpers import NullUI, make_session
from turnstone.core.trajectory import EffectStatus
def _pid_alive(pid: int) -> bool:
try:
os.kill(pid, 0)
except ProcessLookupError:
return False
except PermissionError:
return True
return True
def _kill_pid(pid: int) -> None:
with contextlib.suppress(OSError):
os.kill(pid, signal.SIGKILL)
def _run_in_thread(fn, timeout):
"""Run ``fn`` in a daemon thread; return ``(finished, result)``."""
box = {}
+92
View File
@@ -98,6 +98,98 @@ class TestLenAndClear:
q = NudgeQueue()
assert q.clear() == 0
def test_clear_channels_drops_only_matching(self):
"""The abandoned-generation path drops advisory channels but must
preserve ``"any"``-channel external events (watch fires,
background-shell exits) in order."""
q = NudgeQueue()
q.enqueue("tool_error", "1", "tool")
q.enqueue("watch_triggered", "2", "any")
q.enqueue("correction", "3", "user")
q.enqueue("background_shell_exit", "4", "any")
assert q.clear_channels({"tool", "user"}) == 2
assert q.pending() == [("watch_triggered", "2"), ("background_shell_exit", "4")]
def test_clear_channels_empty_returns_zero(self):
q = NudgeQueue()
assert q.clear_channels({"tool", "user"}) == 0
def test_demote_channel_retags_preserving_order_and_metadata(self):
"""Cancel demotes 'any''quiet': same entries, same order, same
metadata/valid_until — only wake eligibility changes."""
q = NudgeQueue()
q.enqueue("watch_triggered", "w", "any", metadata={"watch_name": "ci"})
q.enqueue("correction", "c", "user")
q.enqueue("background_shell_exit", "b", "any", valid_until=lambda: True)
assert q.demote_channel("any", "quiet") == 2
assert q.pending(channel="any") == []
assert q.pending(channel="quiet") == [
("watch_triggered", "w"),
("background_shell_exit", "b"),
]
# Metadata and valid_until ride the demotion; USER_DRAIN delivers.
from turnstone.core.nudge_queue import USER_DRAIN, WAKE_PENDING
assert not q.has_pending(WAKE_PENDING - {"user"}) # no 'any' left
drained = q.drain(USER_DRAIN)
assert [(t, x, m) for t, x, m in drained] == [
("watch_triggered", "w", {"watch_name": "ci"}),
("correction", "c", None),
("background_shell_exit", "b", None),
]
def test_cap_channel_none_sees_demoted_entries(self):
"""The watch soft cap counts across channels: entries a cancel
demoted to 'quiet' still occupy the budget, and drop-oldest evicts
the stalest regardless of channel."""
q = NudgeQueue()
q.enqueue("watch_triggered", "1", "any")
q.demote_channel("any", "quiet")
q.enqueue("watch_triggered", "2", "any")
assert q.cap_at_or_drop_oldest("watch_triggered", 2, channel=None) is True
assert q.pending() == [("watch_triggered", "2")]
def test_requeue_preserves_seq_for_chronology(self):
"""A failed delivery gives entries back with their ORIGINAL seq, so
a re-queued poll-4 still sorts before the poll-5 that arrived during
the failed attempt — counters never run backwards."""
q = NudgeQueue()
q.enqueue("watch_triggered", "poll-4", "any")
(drained_entry,) = q.drain_entries({"any"})
q.enqueue("watch_triggered", "poll-5", "any") # newer event lands
q.requeue(drained_entry, channel="quiet")
entries = q.drain_entries({"any", "quiet"})
entries.sort(key=lambda e: e.seq)
assert [e.text for e in entries] == ["poll-4", "poll-5"]
def test_requeue_positions_by_seq_for_fifo_drains(self):
"""Positioned insertion: plain (unsorted) drains also see the
re-queued older entry first."""
q = NudgeQueue()
q.enqueue("a", "old", "quiet")
(old_entry,) = q.drain_entries({"quiet"})
q.enqueue("b", "new", "quiet")
q.requeue(old_entry)
assert [text for _t, text in q.pending()] == ["old", "new"]
def test_requeue_preserves_valid_until_and_metadata(self):
alive = {"value": True}
q = NudgeQueue()
q.enqueue("n", "x", "any", valid_until=lambda: alive["value"], metadata={"k": 1})
(entry,) = q.drain_entries({"any"})
q.requeue(entry, channel="quiet")
alive["value"] = False
assert q.drain({"quiet"}) == [] # predicate survived the round-trip
def test_quiet_is_outside_the_wake_gate(self):
from turnstone.core.nudge_queue import TOOL_DRAIN, USER_DRAIN, WAKE_PENDING
q = NudgeQueue()
q.enqueue("background_shell_exit", "b", "quiet")
assert not q.has_pending(WAKE_PENDING)
assert q.has_pending(USER_DRAIN)
assert q.has_pending(TOOL_DRAIN)
class TestDropOldestByType:
def test_drop_oldest_by_type_removes_earliest_match(self):
+21
View File
@@ -1002,6 +1002,27 @@ class TestEvaluateIntentProjection:
assert fa["edits"][0]["near_line"] == 42
assert fa["replace_all"] is False
# -- bash: backgrounding is part of the intent (#817) -------------------
def test_bash_background_projects_run_in_background(self) -> None:
"""The judge must know a bash command will run detached — a
backgrounded server/miner is a different intent than a bounded run.
Built via the real preparer so the prepared item can't silently drop
the flag before the projection reads it."""
session = _make_session()
item = session._prepare_bash(
"c1", {"command": "python -m http.server 8000", "run_in_background": True}
)
fa = _project_func_args(item)
assert fa["run_in_background"] is True
assert fa["command"] == "python -m http.server 8000"
def test_bash_foreground_projects_run_in_background_false(self) -> None:
session = _make_session()
item = session._prepare_bash("c1", {"command": "echo hi"})
fa = _project_func_args(item)
assert fa["run_in_background"] is False
# -- skills: the dead-assignment bug -----------------------------------
def test_skills_create_projection_is_not_empty(self) -> None:
+11 -3
View File
@@ -60,11 +60,11 @@ class TestToolsMetadata:
"""Validate the metadata extracted from JSON files."""
def test_tool_count(self):
# 17 interactive tools + 12 coordinator-only tools.
assert len(TOOLS) == 29
# 19 interactive tools + 12 coordinator-only tools.
assert len(TOOLS) == 31
def test_task_agent_tools_count(self):
assert len(TASK_AGENT_TOOLS) == 11
assert len(TASK_AGENT_TOOLS) == 13
def test_coordinator_tools_count(self):
from turnstone.core.tools import COORDINATOR_TOOLS
@@ -109,6 +109,12 @@ class TestToolsMetadata:
"web_fetch",
"web_search",
"notify",
# Background-shell follow-ups: ``bash_output`` is read-only;
# ``kill_shell`` only signals process groups the session itself
# spawned via an approved bash call — strictly risk-reducing,
# so gating cleanup behind approval adds friction, not safety.
"bash_output",
"kill_shell",
# Coordinator read-only tools (no-mutation, safe to auto-approve):
"inspect_workstream",
"list_workstreams",
@@ -128,6 +134,8 @@ class TestToolsMetadata:
"web_search": "query",
"open_preview": "target",
"task_agent": "prompt",
"bash_output": "id",
"kill_shell": "id",
"memory": "name",
"recall": "query",
"notify": "message",
+5 -5
View File
@@ -201,9 +201,9 @@ class TestSoftCap:
# Oldest ("body-0") gone; newest ("overflow") present.
assert "body-0" not in bodies
assert "overflow" in bodies
# Warning logged.
assert any("watch_dispatch.queue_full" in r.message for r in caplog.records), (
"expected a watch_dispatch.queue_full warning record"
# Warning logged (the shared external-event rail owns the event now).
assert any("external_event.queue_full" in r.message for r in caplog.records), (
"expected an external_event.queue_full warning record"
)
def test_dispatch_soft_cap_does_not_evict_other_types(self, tmp_db):
@@ -457,8 +457,8 @@ class TestWakeFn:
# Entry survived; the failure surfaced as a warning, not a raise
# up into the poll loop.
assert len(session._nudge_queue) == 1
assert any("watch_dispatch.wake_failed" in r.message for r in caplog.records), (
"expected a watch_dispatch.wake_failed warning record"
assert any("external_event.wake_failed" in r.message for r in caplog.records), (
"expected an external_event.wake_failed warning record"
)
+37 -3
View File
@@ -7,6 +7,7 @@ model auto-detection, workstream management, and the main() REPL entry point.
from __future__ import annotations
import argparse
import contextlib
import logging
import os
import readline
@@ -920,6 +921,41 @@ def resolve_cli_persona_kwargs(
return {}
def _close_all_sessions(manager: SessionManager) -> None:
"""Close EVERY loaded session at CLI exit — not just the active one.
``ChatSession.close()`` removes MCP listeners AND reaps the workstream's
background shells (#817). An active-only close would let a dev server
started in workstream 1 survive ``/new`` + ``/exit`` forever: its
detached process group outlives this process — the exact leaked-server
class #816 removed.
Two phases so total exit latency doesn't stack per workstream: the kill
signals land on EVERY session's shells first (microseconds each — after
which nothing can outlive us), then the per-session closes pay their
join budgets, which are near-zero once the kills have landed. A Ctrl-C
during the close phase degrades gracefully instead of aborting the
sweep: the signals are already delivered, the remaining joins are
skipped, and the caller still runs MCP/registry shutdown. Best-effort
per workstream either way — one bad teardown must not stop the rest.
"""
loaded = [(ws.id, ws.session) for ws in manager.list_all() if ws.session is not None]
for _ws_id, session in loaded:
with contextlib.suppress(Exception):
session._background_shells.signal_all()
try:
for ws_id, session in loaded:
try:
session.close()
except Exception:
print(dim(f" (workstream {ws_id[:8]} teardown error, continuing)"))
except KeyboardInterrupt:
# The kills above already landed; skipping the remaining joins
# leaks nothing — it only abandons wedged drain threads that die
# with this process anyway.
print(dim(" (interrupted — background shells already signalled)"))
def main() -> None:
parser = argparse.ArgumentParser(
description="Interactive CLI for OpenAI-compatible models with tool calling.",
@@ -1389,9 +1425,7 @@ def main() -> None:
except Exception as e:
print(f"\n{red(f'Error: {e}')}")
# Close active session (removes MCP listener) before shutting down MCP
if active and active.session:
active.session.close()
_close_all_sessions(manager)
if mcp_client:
mcp_client.shutdown()
registry.shutdown()
+833
View File
@@ -0,0 +1,833 @@
"""Per-session registry for explicitly backgrounded bash shells (#817).
#816 made the ``bash`` tool terminate its whole process group when the call
returns — no leaked servers, no hangs, but also no way to keep a dev server
alive across calls. This registry restores that as an explicit opt-in with
the model-facing shape the frontier coding agents converged on: a boolean on
the shell tool, a short ``bash_N`` handle, a delta-output reader that returns
only lines produced since the previous read, and a kill tool.
Lifetime rules (the #816 rule, extended):
* The tracked command defines the shell's lifetime. When it exits —
naturally, by ``kill``, or by registry teardown — its whole session group
is SIGKILLed, so nothing the command backgrounded can outlive it.
* Shells survive generation-cancel (they are deliberately detached) and die
with the owning session: :meth:`BackgroundShellRegistry.close` runs from
``ChatSession.close()``, which every workstream-teardown path funnels
through.
* Shells spawned inside a task_agent carry that agent's ``owner`` tag; the
agent's ``finally`` reaps them, and owner-scoped lookup keeps parallel
agents (and the parent) from touching each other's handles.
Output is buffered per shell as a rolling deque of lines (stderr tagged
``[stderr] `` inline, arrival order) capped by total characters with
drop-oldest semantics — a chatty server cannot grow a session's memory
unbounded. Reads advance a cursor over the *logical* line stream, so a
line dropped before it was ever read surfaces as an explicit gap count
rather than silently vanishing.
"""
from __future__ import annotations
import contextlib
import itertools
import json
import os
import re
import signal
import subprocess
import sys
import tempfile
import threading
import time
from collections import deque
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Literal
from turnstone.core.log import get_logger
if TYPE_CHECKING:
from collections.abc import Callable
log = get_logger(__name__)
ShellStatus = Literal["running", "completed", "killed"]
# Live (status == "running") shells per session. A hard backstop against a
# runaway loop of spawns on a multi-tenant node, not an operator knob.
_DEFAULT_MAX_SHELLS = 8
# Rolling per-shell buffer cap, in characters. Oldest whole lines drop
# first; the newest line always survives even if it alone exceeds the cap.
_DEFAULT_MAX_BUFFER_CHARS = 200_000
# How long to wait for the drain threads after the group kill forces their
# pipes to EOF. A grandchild that double-``setsid``-escaped the group can
# hold a pipe open past this — the drain is a daemon thread and leaks
# (logged) until that process dies, same acceptance as the foreground tool.
_DRAIN_JOIN_TIMEOUT_S = 5
# TOTAL join budget for ``kill``/``reap`` across all of a shell's threads
# (not per-thread — a wedged drain must not stack timeouts).
_WAITER_JOIN_TIMEOUT_S = 10
# TOTAL join budget for ``close()`` across ALL shells. close() runs on the
# workstream-teardown funnel, which the server can reach from an async
# handler — an unbounded (or per-shell-stacking) wait here would freeze the
# node's event loop, not just this workstream. Threads still alive past the
# budget are daemons: logged and abandoned, they die with their pipes.
_CLOSE_JOIN_BUDGET_S = 5
# Exited records retained per registry (drop-oldest). Keeps a long-lived
# workstream that backgrounds thousands of short jobs from accumulating
# dead records (each can pin up to ``max_buffer_chars`` of buffer) while
# still letting the model read recently-exited shells' output.
_MAX_EXITED_RECORDS = 32
# Bounds on the model-supplied ``filter`` regex: pattern length, how much of
# each line the pattern sees, and wall-clock for the whole filter pass. The
# pass runs in a SUBPROCESS, not a thread: CPython's sre engine holds the
# GIL for the entire duration of one ``search`` call, so a catastrophic-
# backtracking pattern freezes every thread in the interpreter — no
# in-process timeout (thread join, signal, anything) can fire. A child
# process is killable from outside the GIL; on timeout the read errors
# WITHOUT consuming the delta (the cursor only commits on a completed pass).
_MAX_FILTER_PATTERN_CHARS = 512
_FILTER_MAX_LINE_CHARS = 4096
_FILTER_TIMEOUT_S = 2.0
# Runs inside ``sys.executable -c``: reads {pattern, lines} as JSON on
# stdin (lines already truncated parent-side), writes the MATCHING INDEXES
# as JSON on stdout (indexes, not lines — no need to echo a 200K buffer
# back through a pipe).
_FILTER_HELPER_SRC = (
"import json, re, sys\n"
"d = json.load(sys.stdin)\n"
"p = re.compile(d['pattern'])\n"
"sys.stdout.write(json.dumps([i for i, ln in enumerate(d['lines']) if p.search(ln)]))\n"
)
class UnknownShellError(LookupError):
"""No shell with that id is visible in the caller's owner scope."""
class TooManyShellsError(RuntimeError):
"""The per-session live-shell cap would be exceeded."""
class FilterTimeoutError(ValueError):
"""The ``filter`` regex did not finish within the time bound."""
class FilterExecError(RuntimeError):
"""The filter helper process failed for a non-pattern reason."""
def _filter_lines_bounded(pattern: re.Pattern[str], lines: list[str], shell_id: str) -> list[str]:
"""Apply ``pattern`` per line with a wall-clock bound.
A catastrophic-backtracking pattern would wedge the (auto-approved)
tool call — the exact never-returns class #816 removed — and it cannot
be bounded IN-PROCESS: sre holds the GIL for the whole ``search`` call,
freezing every interpreter thread including any watchdog. So the pass
runs in a small child process (killable from the OS): each line
truncated PARENT-side to :data:`_FILTER_MAX_LINE_CHARS` before
serialization (a filter targets log lines; shipping a retained multi-MB
line through the pipe would spend the time budget on I/O and misreport
a fine pattern as slow), the whole pass bounded by
:data:`_FILTER_TIMEOUT_S`, SIGKILL on the child's group past that.
Raises :class:`FilterTimeoutError` on timeout and
:class:`FilterExecError` on a helper failure that is NOT the pattern's
fault (fork/OOM/env) — distinct messages, so the model doesn't
"simplify" an innocent regex. Either way the caller consumes nothing.
The ~tens-of-ms interpreter startup is paid only on filtered reads.
Threat model for the auto-approved path (``bash_output`` runs without
operator approval): the only model-controlled inputs are the PATTERN
and, transitively, the buffered text. The pattern is compiled
parent-side before the fork (a non-regex payload fails there), the
child executes only the fixed ``_FILTER_HELPER_SRC`` — the pattern is
DATA on stdin, never code —, the child gets a scrubbed environment, no
shell, read-only work, and a SIGKILL at the time bound. Worst case a
hostile pattern buys ~2s of one core.
"""
from turnstone.core.env import scrubbed_env
payload = json.dumps(
{
"pattern": pattern.pattern,
"lines": [ln[:_FILTER_MAX_LINE_CHARS] for ln in lines],
},
ensure_ascii=False,
)
try:
proc = subprocess.Popen(
[sys.executable, "-c", _FILTER_HELPER_SRC],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
# Pin BOTH pipe directions to UTF-8: ``text=True`` alone uses
# the locale encoding, and on a C/POSIX-locale node a single
# U+FFFD (from the drain's ``errors="replace"``) would raise
# UnicodeEncodeError out of communicate() — escaping the
# Timeout/Exec error taxonomy as a generic crash. The child's
# own stdio decode is pinned via PYTHONIOENCODING.
encoding="utf-8",
errors="replace",
# scrubbed_env, not os.environ: the helper needs no secrets (it
# runs only our trusted source over already-buffered text), and
# every other fork in this codebase strips API keys/tokens —
# this one must not be the exception.
env={**scrubbed_env(), "PYTHONIOENCODING": "utf-8"},
start_new_session=True,
)
except OSError as e:
# Fork pressure (EAGAIN) / exec failure — same containment class as
# spawn()'s thread-start guard, and by contract NOT the pattern's
# fault.
log.warning("bg_shell.filter_helper_spawn_failed", shell_id=shell_id, error=str(e))
raise FilterExecError(
"the filter could not be applied (helper failed to start); this "
"is not a problem with your pattern — no output was consumed; "
"retry, or read without a filter"
) from e
try:
out, _ = proc.communicate(payload, timeout=_FILTER_TIMEOUT_S)
except subprocess.TimeoutExpired:
with contextlib.suppress(OSError, ProcessLookupError):
os.killpg(proc.pid, signal.SIGKILL)
with contextlib.suppress(subprocess.TimeoutExpired):
proc.wait(timeout=5)
log.warning("bg_shell.filter_timeout", shell_id=shell_id, pattern=pattern.pattern[:80])
raise FilterTimeoutError(
f"filter regex took longer than {_FILTER_TIMEOUT_S:g}s to run; no "
"output was consumed — simplify the pattern or retry without a filter"
) from None
if proc.returncode != 0:
# The parent validated the compile, so a child failure is exotic
# (fork pressure, interpreter env) — NOT the pattern's fault.
log.warning(
"bg_shell.filter_helper_failed",
shell_id=shell_id,
returncode=proc.returncode,
)
raise FilterExecError(
f"the filter could not be applied (helper exited {proc.returncode}); "
"this is not a problem with your pattern — no output was consumed; "
"retry, or read without a filter"
)
try:
indexes = json.loads(out)
except ValueError:
log.warning("bg_shell.filter_helper_bad_output", shell_id=shell_id)
raise FilterExecError(
"the filter could not be applied (helper returned malformed data); "
"no output was consumed — retry, or read without a filter"
) from None
return [lines[i] for i in indexes if isinstance(i, int) and 0 <= i < len(lines)]
def drain_pipe_lines(pipe: Any, on_line: Callable[[str], None]) -> None:
"""Read ``pipe`` line-by-line until EOF, forwarding each to ``on_line``.
The drain half of the shared bash recipe (see :func:`spawn_group_leader`
for the spawn half): both variants of the tool tolerate the same two
end-of-stream shapes. A pipe torn down by the session-group kill is the
expected end; anything else must not kill the drain silently. (The
``errors="replace"`` on the shared Popen pre-empts UnicodeDecodeError —
a ValueError that would otherwise end the drain early and drop ALL
remaining output while reporting a clean success.)
"""
try:
for line in pipe:
on_line(line)
except (ValueError, OSError):
log.debug("bash.drain_read_error", exc_info=True)
def spawn_group_leader(
command: str, *, stop_on_error: bool, env: dict[str, str] | None
) -> tuple[subprocess.Popen[str], int, str]:
"""Write the script, fork the detached group leader, snapshot its pgid.
THE shared prologue for both runs of the model-facing bash tool — the
foreground executor (``ChatSession._exec_bash``) and this registry — so
the two variants of one tool cannot drift: same ``pipefail``/``set -e``
preamble, same decode policy (``errors="replace"``), same session-group
discipline. The script file exists because bash reads scripts lazily
(robust to quoting/length; unlinking early could truncate a long script
mid-run) — the CALLER owns the unlink on its own exit path. On a
failed fork the script is unlinked here and the error propagates. The
pgid snapshot happens while the leader is alive (``start_new_session``
makes ``pgid == pid``); the microseconds-wide pid-wraparound TOCTOU is
the same accepted one as always.
"""
with tempfile.NamedTemporaryFile(mode="w", suffix=".sh", delete=False) as f:
preamble = "set -o pipefail\n"
if stop_on_error:
preamble += "set -e\n"
f.write(preamble + command)
script_path = f.name
try:
proc = subprocess.Popen(
["bash", script_path],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
errors="replace",
start_new_session=True,
env=env,
)
except BaseException:
with contextlib.suppress(OSError):
os.unlink(script_path)
raise
try:
pgid = os.getpgid(proc.pid)
except OSError:
pgid = proc.pid
return proc, pgid, script_path
@dataclass
class ShellRead:
"""One delta read: lines since the previous read, plus shell state.
``lines`` is post-filter (what the caller shows); ``new_line_count`` is
the pre-filter delta size — the cursor advanced past all of them, so a
filtered-out line is consumed, never deferred to a later read.
``dropped_lines`` counts lines lost to the buffer cap before they were
ever read (an explicit gap, not silence).
"""
shell_id: str
status: ShellStatus
exit_code: int | None
lines: list[str]
new_line_count: int
dropped_lines: int
# Lines in this delta longer than the per-line filter window — their
# tails were invisible to the pattern. Only populated on filtered
# reads; the caller surfaces it so a "none matching" answer over
# clipped evidence is never silent.
clipped_lines: int = 0
class BackgroundShell:
"""One detached shell: process handles, rolling buffer, read cursor.
Mutable state is guarded by ``self.lock`` — the drain threads append
while reads snapshot; the waiter thread flips ``status`` exactly once.
"""
def __init__(
self,
shell_id: str,
command: str,
proc: subprocess.Popen[str],
pgid: int,
owner: str | None,
script_path: str,
max_buffer_chars: int,
) -> None:
self.shell_id = shell_id
self.command = command
self.proc = proc
self.pid = proc.pid
self.pgid = pgid
self.owner = owner
self.status: ShellStatus = "running"
self.exit_code: int | None = None
self.lock = threading.Lock()
self._script_path = script_path
self._max_buffer_chars = max_buffer_chars
# Rolling buffer over the logical line stream: ``_buffer`` holds the
# retained tail; ``_dropped_total``/``_total_lines`` are absolute
# line counts so the cursor survives drop-oldest evictions.
self._buffer: deque[str] = deque()
self._buffered_chars = 0
self._dropped_total = 0
self._total_lines = 0
self._read_cursor = 0
# Set (under ``lock``) before the group kill on every deliberate
# termination path so the waiter can distinguish "killed" from
# "completed" and suppress the exit callback.
self._killed = False
self._threads: list[threading.Thread] = []
# Serializes whole read passes (snapshot → filter → commit). The
# buffer lock alone leaves a window where two concurrent reads of
# the same shell snapshot the same cursor and BOTH return the delta
# as new — double-delivering every line. Held across the filter
# subprocess too: correctness over parallel reads of one shell.
self.read_serial = threading.Lock()
# Monotonic EXIT order (registry-assigned by the waiter), None while
# running. Dead-record eviction sorts on this, never on spawn
# order: a long-lived first-spawned server must not be the first
# record evicted — least of all by its own exit's prune, which
# would drop its promised exit notice and crash output unread.
self._exit_seq: int | None = None
@property
def unread_lines(self) -> int:
"""Lines still READABLE that the cursor hasn't consumed — excludes
lines the buffer cap already evicted, so an exit notice never
promises more output than ``bash_output`` can actually return."""
with self.lock:
return self._total_lines - max(self._read_cursor, self._dropped_total)
def _append(self, line: str) -> None:
with self.lock:
self._buffer.append(line)
self._buffered_chars += len(line)
self._total_lines += 1
# Drop oldest whole lines past the cap, but always keep the
# newest — a single oversized line must not empty the buffer.
while self._buffered_chars > self._max_buffer_chars and len(self._buffer) > 1:
dropped = self._buffer.popleft()
self._buffered_chars -= len(dropped)
self._dropped_total += 1
def _snapshot_delta(self) -> tuple[list[str], int, int, ShellStatus, int | None]:
"""Snapshot unread lines WITHOUT consuming them.
Returns ``(delta, gap, new_cursor, status, exit_code)``. The caller
commits ``new_cursor`` via :meth:`_commit_cursor` only after any
filtering succeeded — a failed/timed-out filter must not eat output.
"""
with self.lock:
start = max(self._read_cursor, self._dropped_total)
gap = start - self._read_cursor
delta = list(itertools.islice(self._buffer, start - self._dropped_total, None))
return delta, gap, self._total_lines, self.status, self.exit_code
def _commit_cursor(self, new_cursor: int) -> None:
with self.lock:
# max(): monotonic under concurrent reads of the same scope.
self._read_cursor = max(self._read_cursor, new_cursor)
class BackgroundShellRegistry:
"""Session-scoped table of background shells, ``bash_N``-keyed.
Thread-safe: tool calls (spawn/read/kill), waiter threads (exit
transitions), and teardown (close/reap) may interleave freely.
``on_exit`` fires from the waiter thread on NATURAL exit only — never
for ``kill``/``reap``/``close`` — after the drains have flushed, so a
read triggered by the callback sees the complete output.
"""
def __init__(
self,
*,
max_shells: int = _DEFAULT_MAX_SHELLS,
max_buffer_chars: int = _DEFAULT_MAX_BUFFER_CHARS,
max_exited_records: int = _MAX_EXITED_RECORDS,
on_exit: Callable[[BackgroundShell], None] | None = None,
) -> None:
self._max_shells = max_shells
self._max_buffer_chars = max_buffer_chars
self._max_exited_records = max_exited_records
self._on_exit = on_exit
self._shells: dict[str, BackgroundShell] = {}
self._lock = threading.Lock()
self._counter = 0
self._exit_counter = 0
self._closed = False
# -- Spawning -----------------------------------------------------------
def spawn(
self,
command: str,
*,
env: dict[str, str] | None = None,
owner: str | None = None,
stop_on_error: bool = False,
) -> BackgroundShell:
"""Start ``command`` as a detached shell; return its record.
Raises ``RuntimeError`` after :meth:`close`, :class:`TooManyShellsError`
at the live-shell cap, and propagates ``OSError`` from a failed spawn.
"""
if env is None:
from turnstone.core.env import scrubbed_env
env = scrubbed_env()
# Fast-fail before paying disk + fork; re-checked authoritatively
# under the lock after the fork (spawn stays lock-free through the
# slow syscalls so close()/reap() — which serialize on the registry
# lock with a total time budget — can never be blocked behind a
# stalled filesystem write or fork).
with self._lock:
self._check_capacity_locked(owner)
# Shared prologue with the foreground bash tool — the waiter unlinks
# the script after exit.
proc, pgid, script_path = spawn_group_leader(command, stop_on_error=stop_on_error, env=env)
try:
with self._lock:
# Authoritative re-check: a concurrent spawn/close may have
# won the race while we were forking. Refusal lands in the
# outer handler, which reaps the freshly-forked group —
# nothing may outlive a failed call (#816 rule).
self._check_capacity_locked(owner)
self._counter += 1
shell = BackgroundShell(
shell_id=f"bash_{self._counter}",
command=command,
proc=proc,
pgid=pgid,
owner=owner,
script_path=script_path,
max_buffer_chars=self._max_buffer_chars,
)
# Publish, wire and START the threads under the registry
# lock: close()/reap() take the same lock, so they can never
# observe a registered shell whose threads aren't started
# (they would "join" nothing and return while the drains /
# waiter start up behind them). Registration is popped on a
# start failure IN the same hold, so a thread-exhausted node
# (RLIMIT_NPROC) can't strand an orphan record whose
# never-started Thread objects would make every later
# ``join`` — hence every teardown — raise. The thread
# bodies only ever take ``shell.lock`` or re-take the
# registry lock AFTER this hold is released (the waiter's
# prune), so starting them here cannot deadlock.
self._shells[shell.shell_id] = shell
assert proc.stdout is not None and proc.stderr is not None
out_thread = threading.Thread(
target=self._drain,
args=(proc.stdout, shell, False),
name=f"bg-shell-out-{shell.shell_id}",
daemon=True,
)
err_thread = threading.Thread(
target=self._drain,
args=(proc.stderr, shell, True),
name=f"bg-shell-err-{shell.shell_id}",
daemon=True,
)
waiter = threading.Thread(
target=self._wait_for_exit,
args=(shell, out_thread, err_thread),
name=f"bg-shell-wait-{shell.shell_id}",
daemon=True,
)
shell._threads = [out_thread, err_thread, waiter]
try:
out_thread.start()
err_thread.start()
waiter.start()
except BaseException:
self._shells.pop(shell.shell_id, None)
raise
except BaseException:
# Refused post-fork or thread start failed: reap the fresh group
# (any started drain then EOFs and exits on its own) and surface
# the original error to the tool layer.
with contextlib.suppress(OSError, ProcessLookupError):
os.killpg(pgid, signal.SIGKILL)
with contextlib.suppress(subprocess.TimeoutExpired):
proc.wait(timeout=5)
with contextlib.suppress(OSError):
os.unlink(script_path)
raise
log.info(
"bg_shell.spawned",
shell_id=shell.shell_id,
pid=shell.pid,
owner=owner or "",
)
return shell
def _check_capacity_locked(self, owner: str | None) -> None:
"""Raise if closed or at the live-shell cap. Caller holds the lock."""
if self._closed:
raise RuntimeError("background shells unavailable: session is closing")
live = [s for s in self._shells.values() if s.status == "running"]
if len(live) < self._max_shells:
return
# The cap is registry-wide (it protects the node), but the advice
# must be scope-honest: kill_shell is owner-scoped, so naming
# another scope's ids would send the caller in circles.
mine = [s.shell_id for s in live if s.owner == owner]
others = len(live) - len(mine)
if mine:
detail = f"In your scope: {', '.join(mine)} — stop one with kill_shell"
if others:
detail += f"; {others} more belong to other agents"
detail += "."
else:
detail = (
f"All {others} belong to other agents' scopes and end when "
"those agents finish; wait and retry."
)
raise TooManyShellsError(
f"Background shell limit reached ({self._max_shells} running). {detail}"
)
@staticmethod
def _drain(pipe: Any, shell: BackgroundShell, is_stderr: bool) -> None:
drain_pipe_lines(
pipe, lambda line: shell._append(f"[stderr] {line}" if is_stderr else line)
)
def _wait_for_exit(
self,
shell: BackgroundShell,
out_thread: threading.Thread,
err_thread: threading.Thread,
) -> None:
"""Waiter thread: block on the leader, then tear down the group.
The kill-on-exit is what keeps the #816 guarantee: a child the
command backgrounded dies with the command, and the drains hit EOF
promptly instead of hanging on an inherited pipe write-end.
"""
shell.proc.wait()
with contextlib.suppress(OSError, ProcessLookupError):
os.killpg(shell.pgid, signal.SIGKILL)
out_thread.join(timeout=_DRAIN_JOIN_TIMEOUT_S)
err_thread.join(timeout=_DRAIN_JOIN_TIMEOUT_S)
if out_thread.is_alive() or err_thread.is_alive():
log.warning("bg_shell.drain_leaked", shell_id=shell.shell_id, pid=shell.pid)
with contextlib.suppress(OSError):
os.unlink(shell._script_path)
with shell.lock:
shell.exit_code = shell.proc.returncode
shell.status = "killed" if shell._killed else "completed"
notify = not shell._killed
with self._lock:
self._exit_counter += 1
shell._exit_seq = self._exit_counter
log.info(
"bg_shell.exited",
shell_id=shell.shell_id,
exit_code=shell.exit_code,
status=shell.status,
)
self._prune_exited()
if notify and self._on_exit is not None:
try:
self._on_exit(shell)
except Exception:
log.warning("bg_shell.on_exit_failed", shell_id=shell.shell_id, exc_info=True)
def _prune_exited(self) -> None:
"""Drop the OLDEST-EXITED records past ``max_exited_records``.
Exited records are kept so the model can read a finished shell's
output later, but a workstream that backgrounds thousands of short
jobs must not accumulate them (each can pin ``max_buffer_chars`` of
buffer). Eviction sorts on exit order, NOT spawn order — the shell
whose exit triggered this prune is by definition the newest-exited
and therefore never its own victim (its exit notice and unread
output survive). An exited shell whose ``_exit_seq`` isn't
assigned yet (waiter mid-transition) sorts as newest for the same
reason.
"""
with self._lock:
if self._closed:
return
exited = sorted(
(s for s in self._shells.values() if s.status != "running"),
key=lambda s: s._exit_seq if s._exit_seq is not None else float("inf"),
)
for stale in exited[: max(0, len(exited) - self._max_exited_records)]:
self._shells.pop(stale.shell_id, None)
# -- Lookup / reads ------------------------------------------------------
def _get(self, shell_id: str, owner: str | None) -> BackgroundShell:
with self._lock:
shell = self._shells.get(shell_id)
if shell is not None and shell.owner == owner:
return shell
visible = [
f"{s.shell_id} ({s.status})" for s in self._shells.values() if s.owner == owner
]
known = (
f" Known shells: {', '.join(visible)}." if visible else " No background shells exist."
)
raise UnknownShellError(f"No background shell with id '{shell_id}'.{known}")
def has(self, shell_id: str) -> bool:
with self._lock:
return shell_id in self._shells
def shells(self, owner: str | None = None) -> list[BackgroundShell]:
"""Snapshot of the given scope's shells, in spawn order."""
with self._lock:
return [s for s in self._shells.values() if s.owner == owner]
def read(
self, shell_id: str, *, owner: str | None = None, filter_pattern: str | None = None
) -> ShellRead:
"""Return output produced since the last read of ``shell_id``.
``filter_pattern`` (a regex, ``search`` semantics per line — the
tool-facing ``filter`` arg) narrows what is RETURNED, not what is
consumed: on a successful read the cursor advances past the whole
delta. A failed or timed-out filter consumes NOTHING — the model
can retry without the filter and still get its output. Raises
:class:`UnknownShellError` outside the caller's scope, ``re.error``
for a bad pattern, and :class:`FilterTimeoutError` for a pattern
that blows the time bound (catastrophic backtracking).
"""
shell = self._get(shell_id, owner)
pattern: re.Pattern[str] | None = None
if filter_pattern:
if len(filter_pattern) > _MAX_FILTER_PATTERN_CHARS:
raise re.error( # noqa: TRY003 — mirrors re.compile's own error type
f"filter pattern too long ({len(filter_pattern)} chars, "
f"max {_MAX_FILTER_PATTERN_CHARS})"
)
pattern = re.compile(filter_pattern)
# Serialize the whole pass: concurrent reads of one shell (a
# parallel tool batch) would otherwise snapshot the same cursor and
# each return the full delta as "new".
with shell.read_serial:
delta, gap, new_cursor, status, exit_code = shell._snapshot_delta()
clipped = 0
if pattern is None:
shown = delta
else:
# Raises FilterTimeoutError / FilterExecError BEFORE the
# commit below — a failed filter consumes nothing.
shown = _filter_lines_bounded(pattern, delta, shell.shell_id)
clipped = sum(1 for ln in delta if len(ln) > _FILTER_MAX_LINE_CHARS)
shell._commit_cursor(new_cursor)
return ShellRead(
shell_id=shell.shell_id,
status=status,
exit_code=exit_code,
lines=shown,
new_line_count=len(delta),
dropped_lines=gap,
clipped_lines=clipped,
)
# -- Termination ---------------------------------------------------------
@staticmethod
def _signal_group(shell: BackgroundShell) -> None:
"""SIGKILL the shell's group IFF its leader is still running.
The liveness guard is load-bearing: a completed shell's ``pgid`` is
an hours-stale snapshot the OS may have recycled to an unrelated
process group — signalling it unconditionally would let the
auto-approved ``kill_shell`` (or a routine ``close()``) SIGKILL
another tenant's processes. A leader that exits between the
``poll()`` and the ``killpg`` leaves the same microseconds-wide
pid-wraparound TOCTOU as the foreground tool — accepted there,
accepted here. The guard also keeps a kill racing a natural exit
honest: the waiter labels the shell ``completed`` (with its real
exit code and notice) instead of ``killed``.
"""
with shell.lock:
if shell.proc.poll() is not None:
return # already exited — the waiter's own group kill ran/runs
shell._killed = True
# killpg INSIDE the lock: poll-and-signal is atomic wrt our own
# bookkeeping (nothing can observe _killed without the signal
# having been attempted). The lock is never held around other
# locks, so this cannot deadlock; the OS-level microseconds
# pid-wraparound TOCTOU is the same accepted one as always.
with contextlib.suppress(OSError, ProcessLookupError):
os.killpg(shell.pgid, signal.SIGKILL)
@staticmethod
def _join_threads(shells: list[BackgroundShell], budget_s: float) -> bool:
"""Join every shell thread under ONE shared deadline; True if all done.
The budget is total, not per-thread: teardown latency must not stack
by shell count (``close()`` can run under the server's async close
route — see :data:`_CLOSE_JOIN_BUDGET_S`). Stragglers are daemons;
the caller logs and abandons them.
"""
deadline = time.monotonic() + budget_s
done = True
for shell in shells:
for t in shell._threads:
# suppress: joining a never-started Thread raises
# RuntimeError. spawn() unregisters on a start failure, so
# this is pure belt — teardown must never die on a join.
with contextlib.suppress(RuntimeError):
t.join(timeout=max(0.0, deadline - time.monotonic()))
done = done and not t.is_alive()
return done
def kill(self, shell_id: str, *, owner: str | None = None) -> BackgroundShell:
"""SIGKILL ``shell_id``'s whole group; return its (updated) record.
Suppresses the exit callback — the caller asked for this exit, so
there is nothing to announce. Killing an already-exited shell
signals nothing (see :meth:`_signal_group`) and returns the record
unchanged. On return the record is usually terminal; a leader in
uninterruptible sleep can still read ``running`` after the join
budget — callers report that honestly rather than assuming.
"""
shell = self._get(shell_id, owner)
self._signal_group(shell)
if not self._join_threads([shell], _WAITER_JOIN_TIMEOUT_S):
log.warning("bg_shell.kill_join_timeout", shell_id=shell.shell_id, pid=shell.pid)
return shell
def reap(self, *, owner: str | None) -> None:
"""Kill every shell belonging to ``owner`` and drop their records.
Used by the task_agent teardown: a sub-agent's shells are bound to
the sub-agent's lifetime (never handed to the parent), and dropping
the records keeps dead ``bash_N`` handles from cluttering scope
listings. Suppresses the exit callback for the shells it kills,
same as :meth:`kill` — teardown is the caller's own act, there is
nothing to announce.
"""
with self._lock:
mine = [s for s in self._shells.values() if s.owner == owner]
for shell in mine:
self._signal_group(shell)
if not self._join_threads(mine, _WAITER_JOIN_TIMEOUT_S):
log.warning("bg_shell.reap_join_timeout", owner=owner or "")
with self._lock:
for shell in mine:
self._shells.pop(shell.shell_id, None)
def signal_all(self) -> None:
"""SIGKILL every live shell's group WITHOUT joining or unregistering.
The instant half of teardown, separated so multi-session frontends
can bound their total exit latency: signal every session's groups
first (microseconds each), then pay the join budgets — or, on an
impatient Ctrl-C, signal alone still guarantees no process outlives
the frontend even though the joins are skipped. Liveness-guarded
per shell (:meth:`_signal_group`), so completed shells' stale pgids
are never touched. Idempotent; :meth:`close` remains the complete
teardown.
"""
with self._lock:
shells = list(self._shells.values())
for shell in shells:
self._signal_group(shell)
def close(self) -> None:
"""Kill everything, join threads under a total budget, refuse spawns.
Idempotent; called from ``ChatSession.close()`` (the funnel every
workstream-teardown path runs through). Signals are issued to all
groups first (instant), then ONE shared join budget covers every
thread — a pathological shell (escaped-group grandchild holding the
pipes, D-state leader) delays teardown by at most
:data:`_CLOSE_JOIN_BUDGET_S`, with the stragglers logged and left
to die as daemons. Records are dropped so a queued exit notice's
``valid_until`` predicate (``has(shell_id)``) goes stale and the
drain discards it — nobody is left to read it.
"""
with self._lock:
if self._closed:
return
self._closed = True
shells = list(self._shells.values())
for shell in shells:
self._signal_group(shell)
if not self._join_threads(shells, _CLOSE_JOIN_BUDGET_S):
leaked = [t.name for s in shells for t in s._threads if t.is_alive()]
log.warning("bg_shell.close_join_timeout", leaked=",".join(leaked))
with self._lock:
self._shells.clear()
+7 -4
View File
@@ -26,7 +26,7 @@ from typing import TYPE_CHECKING, Any
from turnstone.core import session_worker
from turnstone.core.log import get_logger
from turnstone.core.nudge_queue import USER_DRAIN, NudgeQueue
from turnstone.core.nudge_queue import WAKE_PENDING, NudgeQueue
from turnstone.core.workstream import WorkstreamState
if TYPE_CHECKING:
@@ -75,7 +75,7 @@ def wake_workstream_if_pending(ws: Workstream, *, trigger: str = "unspecified")
queue at its own seams (``ATTENTION``/``THINKING``/``RUNNING``
all imply a live worker), and ``ERROR`` stays parked for the
operator rather than burning inference unattended.
* nothing drainable under ``USER_DRAIN`` — tool-only entries
* nothing gate-eligible under ``WAKE_PENDING`` — tool-only/quiet entries
belong to the next tool-result seam, not a synthetic empty user
turn (``deliver_wake_nudge_from_queue`` would no-op on them).
@@ -102,7 +102,10 @@ def wake_workstream_if_pending(ws: Workstream, *, trigger: str = "unspecified")
if session is None or ws._closed or ws.state is not WorkstreamState.IDLE:
return False
nudge_queue = getattr(session, "_nudge_queue", None)
if not isinstance(nudge_queue, NudgeQueue) or not nudge_queue.has_pending(USER_DRAIN):
# Gate on WAKE_PENDING, not USER_DRAIN: ``"quiet"`` entries (external
# events demoted by a user cancel) deliver at the next legitimate seam
# but must never themselves wake the workstream the user just stopped.
if not isinstance(nudge_queue, NudgeQueue) or not nudge_queue.has_pending(WAKE_PENDING):
return False
deferred = False
@@ -135,7 +138,7 @@ class IdleNudgeWatcher:
:func:`wake_workstream_if_pending` (the shared gate — see its
docstring for the full gate order). If the workstream's
:class:`NudgeQueue` has any drainable entry for the wake's drain
filter (``USER_DRAIN`` — channels ``"user"`` or ``"any"``), the
gate (``WAKE_PENDING`` — channels ``"user"`` or ``"any"``), the
gate dispatches via ``session_worker.send`` with a no-op
``enqueue`` callback. Tool-only entries don't fire the wake —
they belong to the next tool-result seam, not a synthetic empty
+4
View File
@@ -158,6 +158,10 @@ _NUDGE_MAP: dict[str, str] = {
# consumers recognise the type.
"idle_children": "",
"watch_triggered": "",
# background_shell_exit (#817) likewise: per-fire text is composed by
# ``ChatSession._on_background_shell_exit`` and rides the shared
# external-event rail, never :func:`format_nudge`.
"background_shell_exit": "",
# participant_joined likewise carries no static body — the per-fire text
# ("<name> has joined this shared workstream…") is composed by its producer
# (``ChatSession._maybe_note_new_participant``) and emitted via
+112 -14
View File
@@ -15,7 +15,13 @@ Channels:
* ``"tool"`` — only drains at tool-result seams.
* ``"any"`` — drains at whichever seam fires first (used for
wake-trigger-driven nudges that should not be pinned to a
specific drain seam).
specific drain seam) AND counts toward the idle-wake gate
(:data:`WAKE_PENDING`).
* ``"quiet"`` — drains at whichever seam fires first, but does NOT
count toward the idle-wake gate. A user cancel demotes pending
``"any"`` entries here: the external event (watch fire,
background-shell exit) is still delivered at the next seam, but it
must not wake the workstream the user just stopped.
Drain preserves FIFO order; non-matching entries stay queued. Each
entry can carry an optional ``valid_until`` predicate that drain
@@ -40,19 +46,38 @@ if TYPE_CHECKING:
log = get_logger(__name__)
Channel = Literal["user", "tool", "any"]
_VALID_CHANNELS: frozenset[str] = frozenset({"user", "tool", "any"})
Channel = Literal["user", "tool", "any", "quiet"]
_VALID_CHANNELS: frozenset[str] = frozenset({"user", "tool", "any", "quiet"})
# Module-level filter constants — most callers want one of these and
# pre-allocating spares us a frozenset construction at every drain seam.
USER_DRAIN: frozenset[str] = frozenset({"user", "any"})
TOOL_DRAIN: frozenset[str] = frozenset({"tool", "any"})
USER_DRAIN: frozenset[str] = frozenset({"user", "any", "quiet"})
TOOL_DRAIN: frozenset[str] = frozenset({"tool", "any", "quiet"})
# The idle-wake GATE (``IdleNudgeWatcher``): which pending channels justify
# waking an idle workstream. Deliberately excludes ``"quiet"`` — entries a
# user cancel demoted must ride the next legitimate seam/wake, never cause
# one, or Stop is followed seconds later by an autonomous resume.
WAKE_PENDING: frozenset[str] = frozenset({"user", "any"})
# The quiet channel, named once: the demotion target for external events a
# user cancel must not let re-wake the workstream, and the ride-along drain
# the wake path uses after its WAKE_PENDING pass.
QUIET_CHANNEL: Channel = "quiet"
QUIET_DRAIN: frozenset[str] = frozenset({QUIET_CHANNEL})
class _Entry(NamedTuple):
class Entry(NamedTuple):
"""One queued nudge. Public so consumers of
:meth:`NudgeQueue.drain_entries` can give entries back via
:meth:`NudgeQueue.requeue` — which preserves ``valid_until`` AND the
original ``seq`` (a plain :meth:`NudgeQueue.enqueue` would assign a
fresh seq and re-order a recovered older notice after newer events).
``seq`` is the queue-global insertion number multi-channel drains sort
on to restore chronology."""
nudge_type: str
text: str
channel: Channel
seq: int = 0
valid_until: Callable[[], bool] | None = None
# Producer-supplied optional fields that ride alongside ``text`` when
# drained — used by ``watch_triggered`` to carry ``watch_name`` /
@@ -70,7 +95,8 @@ class NudgeQueue:
"""Single-session FIFO queue with channel-tagged entries."""
def __init__(self) -> None:
self._items: deque[_Entry] = deque()
self._items: deque[Entry] = deque()
self._seq = 0
self._lock = threading.Lock()
def enqueue(
@@ -105,7 +131,8 @@ class NudgeQueue:
if channel not in _VALID_CHANNELS:
raise ValueError(f"channel={channel!r}; expected one of {sorted(_VALID_CHANNELS)}")
with self._lock:
self._items.append(_Entry(nudge_type, text, channel, valid_until, metadata))
self._seq += 1
self._items.append(Entry(nudge_type, text, channel, self._seq, valid_until, metadata))
def drain(
self, channels: frozenset[str] | set[str]
@@ -122,6 +149,15 @@ class NudgeQueue:
without delivering it. Already-removed-from-queue either way —
dropped entries don't ride a future drain.
"""
return [(e.nudge_type, e.text, e.metadata) for e in self.drain_entries(channels)]
def drain_entries(self, channels: frozenset[str] | set[str]) -> list[Entry]:
"""Like :meth:`drain` but returns the surviving :class:`Entry`
records whole — ``seq`` for cross-channel chronology merges and
``valid_until`` so a consumer that must give an entry back (the
wake path's failed-send re-enqueue) can do so without stripping
its staleness predicate.
"""
with self._lock:
if not self._items:
return []
@@ -131,10 +167,10 @@ class NudgeQueue:
# ``USER_DRAIN`` / ``TOOL_DRAIN`` (channel + "any") and
# most queues hold only one channel's entries at a time.
if all(entry.channel in channels for entry in self._items):
candidates: list[_Entry] = list(self._items)
candidates: list[Entry] = list(self._items)
self._items = deque()
else:
kept: deque[_Entry] = deque()
kept: deque[Entry] = deque()
candidates = []
for entry in self._items:
if entry.channel in channels:
@@ -150,14 +186,14 @@ class NudgeQueue:
# every child closed) and logs at ``info``; a raised exception
# is a wiring bug (predicate is misbehaving) and stays at
# ``warning`` with ``exc_info`` so the traceback surfaces.
out: list[tuple[str, str, dict[str, Any] | None]] = []
out: list[Entry] = []
for entry in candidates:
if entry.valid_until is None:
out.append((entry.nudge_type, entry.text, entry.metadata))
out.append(entry)
continue
try:
if entry.valid_until():
out.append((entry.nudge_type, entry.text, entry.metadata))
out.append(entry)
continue
log.info(
"nudge_queue.predicate_dropped",
@@ -187,12 +223,74 @@ class NudgeQueue:
return len(self._items)
def clear(self) -> int:
"""Drop every entry; return the count cleared. Used in cancel paths."""
"""Drop every entry regardless of channel; return the count cleared.
No longer on the cancel path — abandoned generations use
:meth:`clear_channels` + :meth:`demote_channel` so external events
survive. Kept for tests and for full-reset callers that truly mean
"everything".
"""
with self._lock:
n = len(self._items)
self._items.clear()
return n
def requeue(self, entry: Entry, *, channel: Channel | None = None) -> None:
"""Give a drained :class:`Entry` back to the queue, KEEPING its seq.
A plain :meth:`enqueue` would assign a fresh (higher) seq, so a
failed delivery's re-queued OLDER notice would sort after events
that arrived during the failed attempt — running poll counters
backwards at the next seq-merged wake. Insertion is positioned by
seq so plain FIFO drains stay chronological too. ``channel``
overrides the entry's channel (the wake path demotes ``"any"`` →
``"quiet"``); ``valid_until`` and ``metadata`` ride unchanged.
"""
dst = channel if channel is not None else entry.channel
if dst not in _VALID_CHANNELS:
raise ValueError(f"channel={dst!r}; expected one of {sorted(_VALID_CHANNELS)}")
restored = entry._replace(channel=dst)
with self._lock:
for i, existing in enumerate(self._items):
if existing.seq > restored.seq:
self._items.insert(i, restored)
return
self._items.append(restored)
def demote_channel(self, src: Channel, dst: Channel) -> int:
"""Atomically re-tag every ``src``-channel entry as ``dst``; return
the count. Order, text, metadata and ``valid_until`` are preserved
— only drain/wake eligibility changes. The cancel path uses this to
take ``"any"`` entries out of the idle-wake gate (→ ``"quiet"``)
without dropping the external events they announce.
"""
if dst not in _VALID_CHANNELS:
raise ValueError(f"channel={dst!r}; expected one of {sorted(_VALID_CHANNELS)}")
with self._lock:
demoted = 0
for i, entry in enumerate(self._items):
if entry.channel == src:
self._items[i] = entry._replace(channel=dst)
demoted += 1
return demoted
def clear_channels(self, channels: frozenset[str] | set[str]) -> int:
"""Drop entries whose channel is in ``channels``; return the count.
The abandoned-generation paths use this instead of :meth:`clear`:
``"tool"``/``"user"`` advisories are generation-scoped commentary
(a stale ``repeat`` nudge must not bleed into the next send), but
``"any"``-channel entries are EXTERNAL events — a watch fire or a
background-shell exit that happened during the doomed generation
still happened, and dropping it would silently break the "you will
be notified" contract those producers promised the model.
"""
with self._lock:
kept = deque(e for e in self._items if e.channel not in channels)
n = len(self._items) - len(kept)
self._items = kept
return n
def count_by_type(self, nudge_type: str, channel: Channel | None = None) -> int:
"""Return the number of queued entries matching ``nudge_type``.
+586 -122
View File
@@ -48,6 +48,16 @@ from turnstone.core.attachments import (
safe_attachment_label,
unreadable_placeholder,
)
from turnstone.core.background_shells import (
_FILTER_MAX_LINE_CHARS,
BackgroundShell,
BackgroundShellRegistry,
FilterExecError,
FilterTimeoutError,
UnknownShellError,
drain_pipe_lines,
spawn_group_leader,
)
from turnstone.core.config import get_searxng_engines, get_searxng_url
from turnstone.core.edit import find_occurrences, pick_nearest
from turnstone.core.history_decoration import (
@@ -117,7 +127,15 @@ from turnstone.core.metacognition import (
sanitize_payload,
should_nudge,
)
from turnstone.core.nudge_queue import TOOL_DRAIN, USER_DRAIN, NudgeQueue
from turnstone.core.nudge_queue import (
QUIET_CHANNEL,
QUIET_DRAIN,
TOOL_DRAIN,
USER_DRAIN,
WAKE_PENDING,
Entry,
NudgeQueue,
)
from turnstone.core.personas import (
PersonaSnapshot,
resolve_persona_for_kind,
@@ -389,6 +407,50 @@ _active_read_files: contextvars.ContextVar[set[str] | None] = contextvars.Contex
"turnstone_active_read_files", default=None
)
# Owner scope for background shells (#817). ``_exec_task`` sets it to the
# task_agent's call_id for the sub-agent's duration: shells spawned inside
# carry that owner tag, owner-scoped lookup keeps parallel agents (and the
# parent) from touching each other's handles, and the agent's ``finally``
# reaps its own. ``None`` outside a sub-agent → main-session scope.
_active_shell_owner: contextvars.ContextVar[str | None] = contextvars.ContextVar(
"turnstone_active_shell_owner", default=None
)
# Tools exempt from consecutive-identical-call repeat detection: delta-cursor
# readers whose repeated identical call is the documented polling pattern.
_REPEAT_EXEMPT_TOOLS: frozenset[str] = frozenset({"bash_output"})
# ONE source of truth for recognized boolean-arg strings — both coercers
# below derive from these, so a new provider quirk added here reaches every
# tool at once instead of drifting per-tool.
_TRUTHY_STRINGS: frozenset[str] = frozenset({"true", "1", "yes", "on"})
_FALSY_STRINGS: frozenset[str] = frozenset({"false", "0", "no", "off", ""})
_KNOWN_BOOL_STRINGS: frozenset[str] = _TRUTHY_STRINGS | _FALSY_STRINGS
def _is_truthy_flag(value: Any) -> bool:
"""A model-sent boolean arg: bool ``True`` or the string forms providers
intermittently emit ("true"/"1"/"yes", any case). Everything else
including ``None`` and ``False`` strings is ``False``. Used for flags
where silently taking the wrong branch is worse than being lenient
(``run_in_background``: a string-typed true would otherwise run the
command in the FOREGROUND and then group-kill the server the model
believed it detached). ONE dialect for the whole file
``_coord_bool_arg`` delegates here so a provider quirk honored on one
tool is never silently ignored on another."""
if isinstance(value, bool):
return value
if isinstance(value, str):
return value.strip().lower() in _TRUTHY_STRINGS
if isinstance(value, (int, float)):
# Numeric booleans (run_in_background: 1) — same provider-drift
# class as the string forms.
return bool(value)
return False
# Cap on the *content portion* (text after ``path:lineno:``) of an
# emitted search result line. Defends the context budget against
# pathological lines (minified blobs, base64 data, etc.).
@@ -1561,6 +1623,10 @@ class ChatSession:
self._generation: int = 0 # monotonic counter; orphaned threads skip cleanup
self._active_procs: set[subprocess.Popen[str]] = set() # for force-kill
self._procs_lock = threading.Lock()
# Detached shells from bash(run_in_background=true) (#817). Deliberately
# NOT reaped by cancel(): stopping a generation must not kill a server
# the model detached on purpose. close() reaps everything.
self._background_shells = BackgroundShellRegistry(on_exit=self._on_background_shell_exit)
self._cancelled_partial_msg: dict[str, Any] | None = None
self._pending_retry: str | None = None
# True when a fatal exception's text has been persisted to
@@ -2887,59 +2953,116 @@ class ChatSession:
"""
self._watch_runner = runner
self._watch_wake_fn = wake_fn
nudge_queue = self._nudge_queue
ws_id = self._ws_id
def _dispatch(reminder: dict[str, Any], watch_id: str) -> None:
# ``reminder`` is the structured dict produced by
# :func:`build_watch_reminder`. ``text`` carries the
# formatted body — sanitised here over the full string so
# steering-vector / control-char payloads sourced from
# arbitrary shell output can't tamper with the envelope at
# interpolation time. The remaining fields ride as the
# :func:`build_watch_reminder`; the optional fields ride as the
# queue entry's ``metadata`` → sibling keys on the
# ``watch_triggered`` system turn, surfaced in the operator
# bubble (command preview + poll counter).
text = reminder.get("text", "") if isinstance(reminder, dict) else ""
sanitized = sanitize_payload(text)
if not sanitized:
# All control chars / empty after strip — silently drop.
# bubble (command preview + poll counter). Sanitization, the
# drop-oldest soft cap (latest output is most useful) and the
# wake live on the shared external-event rail —
# ``self._watch_wake_fn`` is read at FIRE time there, so an
# identity rebind that re-invoked ``set_watch_runner`` is
# honored without rebuilding this closure.
if not isinstance(reminder, dict):
# Untrusted boundary: a non-dict reminder must drop silently
# (as the old empty-text early-return did), not TypeError out
# of the dispatch closure — WatchRunner would hold the row
# and re-fire it every tick.
return
# Soft cap drops oldest — latest output is most useful.
if nudge_queue.cap_at_or_drop_oldest(
"watch_triggered", _WATCH_QUEUE_SOFT_CAP, channel="any"
):
log.warning(
"watch_dispatch.queue_full ws=%s cap=%d dropped_oldest=True",
ws_id,
_WATCH_QUEUE_SOFT_CAP,
)
def _maybe_sanitize(v: Any) -> Any:
return sanitize_payload(v) if isinstance(v, str) else v
metadata = {
k: _maybe_sanitize(reminder[k])
for k in WATCH_REMINDER_OPTIONAL_KEYS
if k in reminder
}
nudge_queue.enqueue(
text = reminder.get("text", "")
metadata = {k: reminder[k] for k in WATCH_REMINDER_OPTIONAL_KEYS if k in reminder}
self._notify_external_event(
"watch_triggered",
sanitized,
"any",
text,
metadata=metadata or None,
soft_cap=_WATCH_QUEUE_SOFT_CAP,
)
if wake_fn is not None:
try:
wake_fn()
except Exception:
log.warning("watch_dispatch.wake_failed ws=%s", ws_id, exc_info=True)
self._watch_dispatch_fn = _dispatch
runner.set_dispatch_fn(self._ws_id, _dispatch)
def _notify_external_event(
self,
nudge_type: str,
text: str,
*,
metadata: dict[str, Any] | None = None,
valid_until: Callable[[], bool] | None = None,
soft_cap: int | None = None,
) -> None:
"""THE external-event rail: sanitize → (cap) → enqueue ``"any"`` → wake.
Every producer of autonomous notices (watch fires, background-shell
exits) rides this one helper, so policy fixes caps, sanitization,
wake-failure handling can never silently apply to only one of
them. ``sanitize_payload`` runs over the full text (and string
metadata values) so steering-vector / control-char payloads sourced
from arbitrary process output can't tamper with the envelope; an
all-control-chars text drops the event silently. ``soft_cap``
drop-oldest counts across ALL channels (``channel=None``) so
entries a user cancel demoted to ``"quiet"`` still occupy the
budget. The wake makes an already-idle workstream deliver the
entry now busy workstreams are safe, ``session_worker.send``
downgrades to a no-op while a worker owns the session.
"""
if not isinstance(text, str):
# Untrusted producers (watch reminder payloads) can carry a
# non-string text; sanitize_payload would TypeError and a raise
# out of a dispatch closure makes WatchRunner hold + re-fire the
# row every tick. Drop silently, same as empty-after-sanitize.
log.debug("external_event.non_string_text ws=%s type=%s", self._ws_id, nudge_type)
return
sanitized = sanitize_payload(text)
if not sanitized:
return
if soft_cap is not None and self._nudge_queue.cap_at_or_drop_oldest(
nudge_type, soft_cap, channel=None
):
log.warning(
"external_event.queue_full ws=%s type=%s cap=%d dropped_oldest=True",
self._ws_id,
nudge_type,
soft_cap,
)
clean_meta: dict[str, Any] | None = None
if metadata:
clean_meta = {
k: (sanitize_payload(v) if isinstance(v, str) else v) for k, v in metadata.items()
}
self._nudge_queue.enqueue(
nudge_type,
sanitized,
"any",
valid_until=valid_until,
metadata=clean_meta or None,
)
wake_fn = self._watch_wake_fn
if wake_fn is not None:
try:
wake_fn()
except Exception:
# The enqueue already happened; a raise here would abort the
# producer (e.g. WatchRunner._poll_watch before its watch-row
# update commits — re-firing the same reminder every tick).
log.warning(
"external_event.wake_failed ws=%s type=%s",
self._ws_id,
nudge_type,
exc_info=True,
)
def close(self) -> None:
"""Release resources (listener registrations, etc.)."""
"""Release resources (listener registrations, etc.).
Instant signal operations run FIRST (judge cancel events, listener
deregistrations); the background-shell teardown runs last because it
is the only step with a blocking phase (a bounded thread-join, up to
``_CLOSE_JOIN_BUDGET_S`` when a drain is wedged) and nothing here
depends on it serializing instant steps behind it would keep judge
daemons burning inference for the whole join budget on every close.
"""
if self._judge_cancel_event is not None:
self._judge_cancel_event.set()
# Abort every in-flight judge daemon — with parallel task agents
@@ -2983,6 +3106,12 @@ class ChatSession:
self._coord_client.close()
except Exception:
log.debug("chat_session.coord_client_close_failed", exc_info=True)
# Last: the only step with a blocking phase (see docstring). Kill
# signals fire at its start; every teardown path funnels through
# close(), so nothing detached outlives the workstream. Queued exit
# notices go stale with the registry (``valid_until``) and drop at
# the next drain.
self._background_shells.close()
self._cleanup_skill_resources()
def _drop_mcp_surface(self) -> None:
@@ -6321,18 +6450,32 @@ class ChatSession:
self._cancel_event.clear()
def _drain_pending_advisories(self) -> None:
"""Drop every pending nudge regardless of channel.
"""Drop the abandoned generation's advisory nudges — not external events.
Tool-channel nudges (``tool_error``, ``repeat``, ``denial``)
queued earlier in this batch and user-channel nudges
(``correction``, ) queued during ``_check_metacognitive_nudge``
but not yet drained share the same per-session
:class:`NudgeQueue`.
When a generation is abandoned (cancel, KeyboardInterrupt,
unexpected exception) the entire queue drops so nothing bleeds
into the next send's tool loop or next user turn.
are commentary ABOUT the generation being abandoned (cancel,
KeyboardInterrupt, unexpected exception) they drop so nothing
stale bleeds into the next send's tool loop or user turn.
``"any"``-channel entries survive but are DEMOTED to ``"quiet"``:
those are external events (``watch_triggered``,
``background_shell_exit``) that happened regardless of the
generation's fate, and their producers promised the model a notice
a background dev server that crashed during a cancelled turn must
still be announced at the next seam, or the model keeps talking to
a dead server. But they must not CAUSE that seam: an abandoned
generation ends in ``_emit_state("idle")``, and a wake-eligible
entry there would make the ``IdleNudgeWatcher`` resume the
workstream seconds after the user pressed Stop. ``"quiet"``
delivers at the next legitimate seam (user message, tool batch, or
a wake earned by a NEW event) without ever being the wake reason.
Stale entries are handled at drain time by their ``valid_until``
predicates.
"""
self._nudge_queue.clear()
self._nudge_queue.clear_channels({"tool", "user"})
self._nudge_queue.demote_channel("any", QUIET_CHANNEL)
def _synthesize_cancelled_results(self, reason: str) -> None:
"""Synthesize tool_result messages for orphaned tool_calls after cancel.
@@ -7973,6 +8116,10 @@ class ChatSession:
"command": it.get("command", ""),
"timeout": it.get("timeout"),
"stop_on_error": bool(it.get("stop_on_error")),
# Detachment is part of the intent: a backgrounded
# process outlives the call (#817), which changes what
# the judge is approving — never amputate it.
"run_in_background": bool(it.get("run_in_background")),
}
elif name == "write_file":
it["func_args"] = {
@@ -9232,6 +9379,8 @@ class ChatSession:
preparers = {
"bash": self._prepare_bash,
"bash_output": self._prepare_bash_output,
"kill_shell": self._prepare_kill_shell,
"read_file": self._prepare_read_file,
"search": self._prepare_search,
"diff_file": self._prepare_diff,
@@ -9345,6 +9494,31 @@ class ChatSession:
if is_multiline:
preview = f"{DIM}{textwrap.indent(command, ' ')}{RESET}"
# ``is_background`` accepted as an undocumented alias (Gemini's shell
# tool trained that name); ``run_in_background`` is the documented one.
# Lenient coercion: a string-typed "true" must not silently run the
# command in the foreground (where the group kill would then reap the
# server the model believed it detached).
background = _is_truthy_flag(args.get("run_in_background")) or _is_truthy_flag(
args.get("is_background")
)
if background:
# Same approval gate as foreground — the command is what's
# dangerous, not the detachment. ``timeout`` is ignored: there
# is no bounded wait to time out (documented in the schema).
return {
"call_id": call_id,
"func_name": "bash",
"header": f"\u2699 bash (background): {display_cmd}",
"preview": preview,
"needs_approval": True,
"approval_label": "bash",
"execute": self._exec_bash_background,
"command": command,
"run_in_background": True,
"stop_on_error": _is_truthy_flag(args.get("stop_on_error")),
}
return {
"call_id": call_id,
"func_name": "bash",
@@ -9359,7 +9533,71 @@ class ChatSession:
"execute": self._exec_bash,
"command": command,
"timeout": timeout,
"stop_on_error": args.get("stop_on_error") is True,
# Same lenient dialect as run_in_background — a string-typed
# "true" must add ``set -e``, not silently drop it.
"stop_on_error": _is_truthy_flag(args.get("stop_on_error")),
}
def _prepare_bash_output(self, call_id: str, args: dict[str, Any]) -> dict[str, Any]:
shell_id = str(args.get("id") or "").strip()
if not shell_id:
return {
"call_id": call_id,
"func_name": "bash_output",
"header": "\u2717 bash_output: missing id",
"preview": "",
"needs_approval": False,
"error": "Error: missing id (the bash_N handle returned when the shell started)",
}
filter_arg = args.get("filter")
if filter_arg is not None and not isinstance(filter_arg, str):
# An ill-typed filter must error, not silently run unfiltered —
# the unfiltered read would consume the whole delta the model
# wanted narrowed.
return {
"call_id": call_id,
"func_name": "bash_output",
"header": "\u2717 bash_output: invalid filter",
"preview": "",
"needs_approval": False,
"error": (
f"Error: filter must be a regex string "
f"(got {type(filter_arg).__name__}); no output was consumed"
),
}
return {
"call_id": call_id,
"func_name": "bash_output",
"header": f"\u2699 bash_output: {shell_id}",
"preview": "",
"needs_approval": False,
"execute": self._exec_bash_output,
"shell_id": shell_id,
"filter": filter_arg or None,
}
def _prepare_kill_shell(self, call_id: str, args: dict[str, Any]) -> dict[str, Any]:
shell_id = str(args.get("id") or "").strip()
if not shell_id:
return {
"call_id": call_id,
"func_name": "kill_shell",
"header": "\u2717 kill_shell: missing id",
"preview": "",
"needs_approval": False,
"error": "Error: missing id (the bash_N handle returned when the shell started)",
}
# Auto-approved: the argument space is closed (this session's own
# registered shells) and killing one is strictly risk-reducing —
# the dangerous direction was gated when the shell was started.
return {
"call_id": call_id,
"func_name": "kill_shell",
"header": f"\u2699 kill_shell: {shell_id}",
"preview": "",
"needs_approval": False,
"execute": self._exec_kill_shell,
"shell_id": shell_id,
}
@property
@@ -10732,8 +10970,9 @@ class ChatSession:
rather than draining again (the predicate-aware drain runs once, in
``deliver_wake_nudge_from_queue``).
"""
if self._wake_drained_reminders is not None:
entries = self._wake_drained_reminders
from_wake = self._wake_drained_reminders is not None
if from_wake:
entries = self._wake_drained_reminders or []
self._wake_drained_reminders = None # consume — only delivered once
else:
items = self._nudge_queue.drain(USER_DRAIN)
@@ -10743,12 +10982,24 @@ class ChatSession:
if meta:
entry.update(meta)
entries.append(entry)
for entry in entries:
for i, entry in enumerate(entries):
source = str(entry.get("type") or "")
if not source:
continue
meta = {k: v for k, v in entry.items() if k not in ("type", "text")}
self._append_system_turn(source, str(entry.get("text") or ""), **meta)
try:
self._append_system_turn(source, str(entry.get("text") or ""), **meta)
except BaseException:
if from_wake:
# Mid-batch failure on a wake: re-stash the un-emitted
# TAIL (including the failing entry — its persistence is
# UNKNOWN; at-least-once beats silently-eaten for an
# exit notice that fires exactly once) so the wake
# caller's finally can re-enqueue instead of losing the
# suffix. Non-wake callers drain directly and keep the
# pre-existing best-effort semantics.
self._wake_drained_reminders = entries[i:]
raise
def _queue_tool_advisory(self, nudge_type: str, text: str) -> None:
"""Queue a metacognitive nudge for the next tool-result batch.
@@ -10805,16 +11056,34 @@ class ChatSession:
post-retry stream failure leaves them in place operator
intervention is required for the underlying failure anyway.
"""
items = self._nudge_queue.drain(USER_DRAIN)
if not items:
# Two-pass drain: wake-eligible channels first. ``"quiet"`` entries
# (external events demoted by a user cancel) ride a wake earned by
# others but never justify one — if every wake-eligible candidate
# evaporated at drain time (``valid_until``), bail WITHOUT touching
# the quiet entries: they stay queued for the next legitimate seam
# instead of resuming a workstream the user stopped. The merged
# batch is re-sorted by queue insertion ``seq`` so cross-channel
# chronology survives the two passes (a demoted poll-4 fire must
# not render after the poll-5 fire that earned the wake).
drained = self._nudge_queue.drain_entries(WAKE_PENDING)
if not drained:
return
drained += self._nudge_queue.drain_entries(QUIET_DRAIN)
drained.sort(key=lambda e: e.seq)
self._wake_source_tag = "system_nudge"
wake_reminders: list[dict[str, Any]] = []
for nudge_type, text, meta in items:
entry: dict[str, Any] = {"type": nudge_type, "text": text}
if meta:
entry.update(meta)
# Identity map from reminder dict → its source Entry: the failure
# path recovers each un-emitted entry by ``id(reminder)`` lookup,
# never by index arithmetic, so it stays correct even if a future
# edit filters or reorders the reminder list between here and
# ``_emit_pending_user_nudges``.
entry_by_reminder: dict[int, Entry] = {}
for queued in drained:
entry: dict[str, Any] = {"type": queued.nudge_type, "text": queued.text}
if queued.metadata:
entry.update(queued.metadata)
wake_reminders.append(entry)
entry_by_reminder[id(entry)] = queued
self._wake_drained_reminders = wake_reminders
try:
self.send("", from_wake=True)
@@ -10828,7 +11097,36 @@ class ChatSession:
log.info("wake_nudge.cancelled ws=%s", self._ws_id[:8])
finally:
self._wake_source_tag = ""
undelivered = self._wake_drained_reminders
self._wake_drained_reminders = None
if undelivered:
# The send died before ``_emit_pending_user_nudges`` finished
# the batch (it re-stashes the un-emitted TAIL on a mid-batch
# failure, and nulls the attr only when done). Recovery is
# EXTERNAL-notices-only, and always to ``"quiet"``:
# ``requeue`` keeps seq (a re-queued poll-4 still renders
# before poll-5 on the retry) and the ``valid_until``
# predicate (a stale notice stays droppable), while quiet
# keeps the entry OUT of the wake gate. A ``"user"``
# advisory is deliberately DROPPED instead: re-queueing it
# wake-eligible re-arms ``_retry_pending_wake``'s zero-
# backoff worker-exit gate — a repeatable pre-consumption
# send failure would respawn wake workers in an unbounded
# hot loop (persisting an orphan synthetic user turn per
# spin). Losing a generation-scoped metacog hint on a
# rare failed wake is the strictly smaller harm.
for reminder in undelivered:
recovered = entry_by_reminder.get(id(reminder))
if recovered is None or not recovered.text:
continue
if recovered.channel == "user":
log.debug(
"wake_nudge.user_advisory_dropped ws=%s type=%s",
self._ws_id[:8],
recovered.nudge_type,
)
continue
self._nudge_queue.requeue(recovered, channel=QUIET_CHANNEL)
def _apply_post_execute_advisories(
self,
@@ -10871,10 +11169,19 @@ class ChatSession:
for i, (tc_id, output) in enumerate(results):
tc = _tc_by_id.get(tc_id)
if tc and isinstance(output, str):
# Delta-cursor readers (``_REPEAT_EXEMPT_TOOLS``): identical
# args ARE the documented usage (poll the same handle) and
# the result differs by construction — a "result is the
# same" warning would be factually false. They are still
# RECORDED (never skipped): the detector's contract is that
# any different signature breaks a streak, so an exempt call
# interleaved between identical bash calls must keep those
# bash calls from reading as consecutive.
exempt = tc["function"]["name"] in _REPEAT_EXEMPT_TOOLS
raw = tc["function"]["name"] + ":" + tc["function"]["arguments"]
sig = hashlib.sha256(raw.encode()).hexdigest()
is_json = output.lstrip().startswith(("{", "["))
if self._repeat_detector.record(sig):
if self._repeat_detector.record(sig) and not exempt:
_repeat_detected = True
if not is_json:
output += (
@@ -10976,21 +11283,16 @@ class ChatSession:
"""Return ``args[key]`` as a bool with robust string coercion.
Plain ``bool(x)`` treats ``"false"`` as truthy (non-empty string).
Accept actual bools verbatim; parse common string forms; return
``default`` for anything else.
Accept actual bools verbatim; delegate the truthy dialect to
:func:`_is_truthy_flag` (ONE coercion dialect file-wide); return
``default`` for a missing key or an unrecognized string.
"""
val = args.get(key)
if isinstance(val, bool):
return val
if isinstance(val, str):
normalized = val.strip().lower()
if normalized in ("true", "1", "yes", "on"):
return True
if normalized in ("false", "0", "no", "off", ""):
return False
if isinstance(val, (int, float)) and not isinstance(val, bool):
return bool(val)
return default
if val is None:
return default
if isinstance(val, str) and val.strip().lower() not in _KNOWN_BOOL_STRINGS:
return default
return _is_truthy_flag(val)
@staticmethod
def _flatten_spawn_arg(value: Any, cap: int) -> str:
@@ -14031,25 +14333,19 @@ class ChatSession:
call_id, command = item["call_id"], item["command"]
timeout = item.get("timeout") or self.tool_timeout
try:
with tempfile.NamedTemporaryFile(mode="w", suffix=".sh", delete=False) as f:
preamble = "set -o pipefail\n"
if item.get("stop_on_error"):
preamble += "set -e\n"
f.write(preamble + command)
script_path = f.name
# Pre-bind so the ``finally`` can't raise ``UnboundLocalError`` and
# mask the real error if ``Popen`` below fails.
# Pre-bind so the ``finally`` can't raise ``UnboundLocalError``
# and mask the real error if the spawn below fails.
proc: subprocess.Popen[str] | None = None
script_path: str | None = None
try:
from turnstone.core.env import scrubbed_env
proc = subprocess.Popen(
["bash", script_path],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
errors="replace",
start_new_session=True,
# Shared prologue with the background registry (script file +
# detached group leader + pgid snapshot) — one spawn recipe
# for both variants of the tool, so they cannot drift.
proc, pgid, script_path = spawn_group_leader(
command,
stop_on_error=item.get("stop_on_error") is True,
env=scrubbed_env(extra=self._skill_resource_env()),
)
with self._procs_lock:
@@ -14065,54 +14361,38 @@ class ChatSession:
stdout_parts: list[str] = []
stderr_lines: list[str] = []
def _drain(pipe: Any, sink: list[str], to_ui: bool) -> None:
# End-of-stream tolerance lives in the shared
# ``drain_pipe_lines`` (the drain half of the recipe both
# bash variants share); only the sinks differ — stdout also
# streams to the UI.
def _on_stdout(line: str) -> None:
stdout_parts.append(line)
try:
for line in pipe:
sink.append(line)
if to_ui:
try:
self.ui.on_tool_output_chunk(call_id, line)
except Exception:
log.debug(
"UI callback error during tool output",
exc_info=True,
)
except (ValueError, OSError):
# Pipe torn down by the session-group kill below is the
# expected case; anything else must not kill the drain
# silently. (Undecodable bytes can't land here — the
# ``errors="replace"`` on Popen pre-empts UnicodeDecodeError,
# which is a ValueError that would otherwise drop all output.)
log.debug("bash.drain_read_error", exc_info=True)
self.ui.on_tool_output_chunk(call_id, line)
except Exception:
log.debug("UI callback error during tool output", exc_info=True)
assert proc.stdout is not None and proc.stderr is not None
stdout_thread = threading.Thread(
target=_drain,
args=(proc.stdout, stdout_parts, True),
target=drain_pipe_lines,
args=(proc.stdout, _on_stdout),
name=f"bash-drain-out-{call_id}",
daemon=True,
)
stderr_thread = threading.Thread(
target=_drain,
args=(proc.stderr, stderr_lines, False),
target=drain_pipe_lines,
args=(proc.stderr, stderr_lines.append),
name=f"bash-drain-err-{call_id}",
daemon=True,
)
stdout_thread.start()
stderr_thread.start()
# Snapshot the session-group id while the leader is alive
# (``start_new_session=True`` makes ``pgid == proc.pid``). While
# any member survives, the group names only our own descendants;
# once they have all exited it is empty and the kill below is a
# harmless no-op. (A pid-wraparound landing a fresh session
# leader on this exact id in the microseconds after a normal-exit
# reap is the standard accepted TOCTOU — negligible, and only when
# nothing needs killing anyway.)
try:
pgid = os.getpgid(proc.pid)
except OSError:
pgid = proc.pid
# ``pgid`` was snapshotted by ``spawn_group_leader`` while the
# leader was alive. While any member survives, the group
# names only our own descendants; once they have all exited
# it is empty and the kill below is a harmless no-op (the
# accepted microseconds pid-wraparound TOCTOU).
# Wait for the tracked command, bounded by ``timeout`` and
# cooperatively cancellable. Keyed on process exit, never pipe
@@ -14156,7 +14436,10 @@ class ChatSession:
if proc is not None:
with self._procs_lock:
self._active_procs.discard(proc)
os.unlink(script_path)
# ``spawn_group_leader`` already unlinked on a failed fork —
# ``script_path`` stays None on that path.
if script_path is not None:
os.unlink(script_path)
if timed_out.is_set():
raise subprocess.TimeoutExpired(cmd="bash", timeout=timeout)
@@ -14224,6 +14507,181 @@ class ChatSession:
self._report_tool_result(call_id, "bash", msg, is_error=True)
return call_id, msg
def _exec_bash_background(self, item: dict[str, Any]) -> tuple[str, str]:
"""Start ``command`` as a detached background shell (#817).
Returns immediately with the ``bash_N`` handle the shell's later
output/exit is a NEW event (a NudgeQueue notice at the next seam),
never a deferred resolution of this call_id, so the canonical
trajectory stays faithful on replay.
"""
call_id, command = item["call_id"], item["command"]
from turnstone.core.env import scrubbed_env
try:
shell = self._background_shells.spawn(
command,
env=scrubbed_env(extra=self._skill_resource_env()),
owner=_active_shell_owner.get(),
stop_on_error=item.get("stop_on_error") is True,
)
except (RuntimeError, OSError) as e:
# TooManyShellsError / registry-closed / spawn failure — all
# actionable by the model (kill one, or just don't background).
msg = f"Error: {e}"
self._report_tool_result(call_id, "bash", msg, is_error=True)
return call_id, msg
msg = (
f"Started background shell {shell.shell_id} (pid {shell.pid}). "
f'Read new output with bash_output(id="{shell.shell_id}"); stop it with '
f'kill_shell(id="{shell.shell_id}").'
)
if shell.owner is None:
msg += " It runs until it exits or is killed; a system notice will announce its exit."
else:
# Sub-agent scope: the shell dies with this agent. Said here so
# the agent doesn't promise its parent a server that will be
# reaped the moment it returns.
msg += (
" It is scoped to this agent and will be terminated when the "
"agent finishes — do not report it to your caller as still "
"running; read/verify what you need before returning."
)
self._report_tool_result(call_id, "bash", msg)
return call_id, msg
def _exec_bash_output(self, item: dict[str, Any]) -> tuple[str, str]:
"""Delta read of a background shell: only output since the last read."""
call_id, shell_id = item["call_id"], item["shell_id"]
filter_arg = item.get("filter")
try:
read = self._background_shells.read(
shell_id, owner=_active_shell_owner.get(), filter_pattern=filter_arg
)
except UnknownShellError as e:
msg = f"Error: {e}"
self._report_tool_result(call_id, "bash_output", msg, is_error=True)
return call_id, msg
except (FilterTimeoutError, FilterExecError) as e:
# A VALID pattern that blew the time bound, or a helper failure
# that wasn't the pattern's fault — either way nothing was
# consumed and the message says which and what to do.
msg = f"Error: {e}"
self._report_tool_result(call_id, "bash_output", msg, is_error=True)
return call_id, msg
except re.error as e:
msg = f"Error: invalid filter regex: {e}"
self._report_tool_result(call_id, "bash_output", msg, is_error=True)
return call_id, msg
if read.status == "completed":
state = f"completed, exit code {read.exit_code}"
else:
state = read.status
parts = [f"{shell_id} ({state})"]
if read.dropped_lines:
parts.append(f"[{read.dropped_lines} earlier line(s) dropped from the buffer]")
if read.clipped_lines:
# A "none matching" answer over clipped evidence must never be
# silent — the model would report a clean run whose error sat
# past the match window.
parts.append(
f"[{read.clipped_lines} line(s) longer than {_FILTER_MAX_LINE_CHARS} "
"chars were only partially visible to the filter; matches beyond "
"that window are not detected — read without a filter to see them]"
)
if read.lines:
if filter_arg:
parts.append(
f"{len(read.lines)} of {read.new_line_count} new line(s) match the filter:"
)
else:
parts.append(f"{read.new_line_count} new line(s):")
parts.append("".join(read.lines).rstrip("\n"))
elif read.new_line_count:
parts.append(f"{read.new_line_count} new line(s), none matching the filter.")
else:
parts.append("No new output since the last read.")
full = "\n".join(parts)
output = self._truncate_output(full)
if len(output) < len(full):
# Unlike foreground bash (re-run to re-see), the delta cursor has
# already consumed the elided middle — say so, or a "no new
# output" follow-up reads as "nothing was missed".
output += (
"\n[the truncated middle was consumed and cannot be re-read; "
"use filter to narrow future reads]"
)
self._report_tool_result(call_id, "bash_output", output)
return call_id, output
def _exec_kill_shell(self, item: dict[str, Any]) -> tuple[str, str]:
"""Kill a background shell's whole process group."""
call_id, shell_id = item["call_id"], item["shell_id"]
try:
shell = self._background_shells.kill(shell_id, owner=_active_shell_owner.get())
except UnknownShellError as e:
msg = f"Error: {e}"
self._report_tool_result(call_id, "kill_shell", msg, is_error=True)
return call_id, msg
if shell.status == "killed":
msg = (
f"Killed background shell {shell_id}. Output produced before the "
f'kill remains readable via bash_output(id="{shell_id}") until it '
"ages out of the recent-shells list."
)
elif shell.status == "running":
# SIGKILL was sent but the leader didn't exit within the join
# budget (uninterruptible sleep — NFS, a driver). Outcome
# honesty: never tell the model a live process already exited.
msg = (
f"SIGKILL sent to background shell {shell_id}, but it has not "
"exited yet (possibly uninterruptible I/O). It may still die "
f'shortly — check bash_output(id="{shell_id}") before assuming '
"it is gone."
)
else:
msg = (
f"Background shell {shell_id} had already exited "
f"(status: {shell.status}, exit code {shell.exit_code}); nothing to kill."
)
self._report_tool_result(call_id, "kill_shell", msg)
return call_id, msg
def _on_background_shell_exit(self, shell: BackgroundShell) -> None:
"""Waiter-thread callback: queue an exit notice for a detached shell.
Rides the watch rail: channel ``"any"`` (drains at whichever seam
fires first AND can wake an idle workstream a ``"tool"`` entry
could do neither), an explicit wake for the already-idle case, and
a ``valid_until`` predicate so a notice whose shell is gone
(registry closed with the workstream) is dropped, not delivered.
Push the notice, let the model pull the detail via ``bash_output``
bounds context. Sub-agent shells get no notice: the sub-loop is
synchronous and polls; its shells die with it.
"""
if shell.owner is not None:
return
cmd_excerpt = shell.command.split("\n")[0][:80]
unread = shell.unread_lines
registry = self._background_shells
shell_id = shell.shell_id
self._notify_external_event(
"background_shell_exit",
(
f"Background shell {shell_id} ({cmd_excerpt}) exited with code "
f"{shell.exit_code}{unread} unread line(s); use "
f'bash_output(id="{shell_id}") to read them.'
),
metadata={
"shell_id": shell_id,
"command": cmd_excerpt,
"exit_code": shell.exit_code,
"unread_lines": unread,
},
valid_until=lambda: registry.has(shell_id),
)
@staticmethod
def _read_text_lines(path: str) -> tuple[list[str], str, str | None]:
"""Read a text file with binary detection and symlink resolution.
@@ -15174,6 +15632,10 @@ class ChatSession:
# sibling's reads can't suppress THIS agent's blind-overwrite guard. The
# agent's own reads merge back to the parent in ``finally``.
read_token = _active_read_files.set(set(self._current_read_files))
# Background shells spawned by this sub-agent carry its call_id as
# owner: scoped lookup (parallel agents + parent can't touch them)
# and bound to the agent's lifetime — reaped in ``finally`` below.
shell_token = _active_shell_owner.set(call_id)
try:
result = self._run_agent(
agent_turns,
@@ -15228,6 +15690,8 @@ class ChatSession:
_active_read_files.reset(read_token)
if sub_reads:
self._current_read_files.update(sub_reads)
_active_shell_owner.reset(shell_token)
self._background_shells.reap(owner=call_id)
self._end_agent_scope()
self._clear_agent_children(call_id)
try:
+4
View File
@@ -125,6 +125,10 @@ SYSTEM_TURN_SOURCES: Final = frozenset(
"compaction_pending",
"idle_children",
"watch_triggered",
# Background-shell exit notice (#817) — rides the same external-event
# rail as ``watch_triggered``; carries ``shell_id`` / ``command`` /
# ``exit_code`` / ``unread_lines`` metadata.
"background_shell_exit",
"participant_joined",
}
)
+24
View File
@@ -4179,6 +4179,30 @@ async def _lifespan(app: Starlette) -> AsyncGenerator[None, None]:
state_writer = getattr(app.state, "state_writer", None)
if state_writer is not None:
await asyncio.to_thread(state_writer.shutdown)
# Reap every loaded session's background shells (#817) before MCP
# teardown — a GRACEFUL server shutdown must not orphan detached
# process groups (the leaked-server class #816 removed; a hard crash
# remains the documented acceptance). Two phases like the CLI exit:
# signal every session's shells first (instant — after this nothing
# can outlive us), then pay the bounded per-session join budgets off
# the event loop. Session close() also removes MCP listeners, hence
# the ordering before mcp_client.shutdown().
mgr = WebUI._workstream_mgr
if mgr is not None:
loaded = [(ws.id, ws.session) for ws in mgr.list_all() if ws.session is not None]
for _ws_id, session in loaded:
with contextlib.suppress(Exception):
session._background_shells.signal_all()
def _close_loaded() -> None:
for ws_id, session in loaded:
try:
session.close()
except Exception:
log.exception("server.session_close_failed", ws_id=ws_id[:8])
if loaded:
await asyncio.to_thread(_close_loaded)
# health_registry is stateless (no background threads) — nothing to stop
if app.state.mcp_client:
app.state.mcp_client.shutdown()
+5 -1
View File
@@ -1,6 +1,6 @@
{
"name": "bash",
"description": "Execute a bash command and return stdout + stderr. Use for running programs, git, tests, system commands, installing packages, etc. Environment questions ('What Python version?', 'Is X installed?') are tool-use tasks — e.g. bash(command='python --version'). For file creation use write_file instead; for man pages use man instead. Long output is truncated (head+tail preserved, middle elided). Stderr lines prefixed with [stderr]. Runs to completion and returns: any process the command leaves running in the background (e.g. 'server &') is terminated when the command returns — nothing persists across calls.",
"description": "Execute a bash command and return stdout + stderr. Use for running programs, git, tests, system commands, installing packages, etc. Environment questions ('What Python version?', 'Is X installed?') are tool-use tasks — e.g. bash(command='python --version'). For file creation use write_file instead; for man pages use man instead. Long output is truncated (head+tail preserved, middle elided). Stderr lines prefixed with [stderr]. Runs to completion and returns: any process the command leaves running in the background (e.g. 'server &') is terminated when the command returns — nothing persists across calls. To keep a long-lived process (dev server, watcher) running across calls, set run_in_background=true instead of using '&'.",
"parameters": {
"type": "object",
"properties": {
@@ -15,6 +15,10 @@
"stop_on_error": {
"type": "boolean",
"description": "If true, enables 'set -e' so the script exits on the first command failure. Default false. Use for multi-step scripts where intermediate failures should halt execution."
},
"run_in_background": {
"type": "boolean",
"description": "If true, start the command as a detached background shell and return immediately with a shell id (e.g. bash_1). Default false. Read new output later with bash_output(id=...); stop it with kill_shell(id=...). The shell runs until it exits, is killed, or the workstream closes; in the main session a system notice announces its exit. Inside a task agent there is no exit notice — poll bash_output — and the shell is also terminated when the agent finishes. The timeout parameter does not apply. Use for long-lived processes like dev servers — not for ordinary commands whose result you want now."
}
},
"required": ["command"]
+21
View File
@@ -0,0 +1,21 @@
{
"name": "bash_output",
"description": "Read new output from a background shell started with bash(run_in_background=true). Returns only output produced since your previous bash_output call for that shell, plus its status (running / completed / killed) and exit code once it has exited. Stderr lines are prefixed with [stderr]. Poll this to monitor a long-running process. In the main session a system notice announces when the shell exits, so you do not need to poll a shell you are merely waiting on; inside a task agent there is no notice — poll before you finish.",
"parameters": {
"type": "object",
"properties": {
"id": {
"type": "string",
"description": "Background shell id, e.g. bash_1 (returned when the shell was started)."
},
"filter": {
"type": "string",
"description": "Optional regular expression; only new lines matching it are returned. Non-matching lines in this read are consumed and will not be returned by later calls."
}
},
"required": ["id"]
},
"task_agent": true,
"auto_approve": true,
"primary_key": "id"
}
+17
View File
@@ -0,0 +1,17 @@
{
"name": "kill_shell",
"description": "Terminate a background shell started with bash(run_in_background=true), killing its whole process group. Use it when the process is no longer needed or is misbehaving. Output already produced remains readable via bash_output afterwards.",
"parameters": {
"type": "object",
"properties": {
"id": {
"type": "string",
"description": "Background shell id to terminate, e.g. bash_1."
}
},
"required": ["id"]
},
"task_agent": true,
"auto_approve": true,
"primary_key": "id"
}