diff --git a/CHANGELOG.md b/CHANGELOG.md index 46400610..0b8ecb6e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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] diff --git a/tests/_proc_helpers.py b/tests/_proc_helpers.py new file mode 100644 index 00000000..92c26c02 --- /dev/null +++ b/tests/_proc_helpers.py @@ -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 diff --git a/tests/test_background_shells.py b/tests/test_background_shells.py new file mode 100644 index 00000000..334cc45d --- /dev/null +++ b/tests/test_background_shells.py @@ -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() diff --git a/tests/test_bash_background_tool.py b/tests/test_bash_background_tool.py new file mode 100644 index 00000000..2172e033 --- /dev/null +++ b/tests/test_bash_background_tool.py @@ -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 diff --git a/tests/test_bash_tool_background_hang.py b/tests/test_bash_tool_background_hang.py index 901ecb91..e4bfa555 100644 --- a/tests/test_bash_tool_background_hang.py +++ b/tests/test_bash_tool_background_hang.py @@ -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 = {} diff --git a/tests/test_nudge_queue.py b/tests/test_nudge_queue.py index 09d8d94a..d46e54c4 100644 --- a/tests/test_nudge_queue.py +++ b/tests/test_nudge_queue.py @@ -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): diff --git a/tests/test_session.py b/tests/test_session.py index f47c0862..86430dc3 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -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: diff --git a/tests/test_tools_schema.py b/tests/test_tools_schema.py index 6be13995..89d055cc 100644 --- a/tests/test_tools_schema.py +++ b/tests/test_tools_schema.py @@ -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", diff --git a/tests/test_watch_dispatch.py b/tests/test_watch_dispatch.py index e964c124..438930b0 100644 --- a/tests/test_watch_dispatch.py +++ b/tests/test_watch_dispatch.py @@ -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" ) diff --git a/turnstone/cli.py b/turnstone/cli.py index 2ccd4a38..abb3bfaa 100644 --- a/turnstone/cli.py +++ b/turnstone/cli.py @@ -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() diff --git a/turnstone/core/background_shells.py b/turnstone/core/background_shells.py new file mode 100644 index 00000000..42269f2a --- /dev/null +++ b/turnstone/core/background_shells.py @@ -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() diff --git a/turnstone/core/idle_nudge_watcher.py b/turnstone/core/idle_nudge_watcher.py index 0758a22b..6eccae2b 100644 --- a/turnstone/core/idle_nudge_watcher.py +++ b/turnstone/core/idle_nudge_watcher.py @@ -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 diff --git a/turnstone/core/metacognition.py b/turnstone/core/metacognition.py index 7af28b19..7c313eed 100644 --- a/turnstone/core/metacognition.py +++ b/turnstone/core/metacognition.py @@ -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 # (" has joined this shared workstream…") is composed by its producer # (``ChatSession._maybe_note_new_participant``) and emitted via diff --git a/turnstone/core/nudge_queue.py b/turnstone/core/nudge_queue.py index bfc90a9e..53b35a6d 100644 --- a/turnstone/core/nudge_queue.py +++ b/turnstone/core/nudge_queue.py @@ -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``. diff --git a/turnstone/core/session.py b/turnstone/core/session.py index 9f896e7c..9725d068 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -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: diff --git a/turnstone/core/tool_advisory.py b/turnstone/core/tool_advisory.py index f5753a67..18006ed8 100644 --- a/turnstone/core/tool_advisory.py +++ b/turnstone/core/tool_advisory.py @@ -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", } ) diff --git a/turnstone/server.py b/turnstone/server.py index 1fa7b086..1a316e72 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -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() diff --git a/turnstone/tools/bash.json b/turnstone/tools/bash.json index b9bff9e9..000eb2a4 100644 --- a/turnstone/tools/bash.json +++ b/turnstone/tools/bash.json @@ -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"] diff --git a/turnstone/tools/bash_output.json b/turnstone/tools/bash_output.json new file mode 100644 index 00000000..4f27947b --- /dev/null +++ b/turnstone/tools/bash_output.json @@ -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" +} diff --git a/turnstone/tools/kill_shell.json b/turnstone/tools/kill_shell.json new file mode 100644 index 00000000..9ebd7dc2 --- /dev/null +++ b/turnstone/tools/kill_shell.json @@ -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" +}