fix(runtime): ordered bash output completion sentinel (#1838)
* fix(runtime): fence bash output completion * docs(runtime): document post-fence output visibility in bash() docstring * fix(runtime): emit bash completion fence via absolute printf path A shell function named `command` shadows `\command -p printf` (the backslash only defeats aliases), swallowing both fence frames and wedging the await behind background jobs until shell death. A slash-qualified printf resolved on the system default utility PATH bypasses function and alias lookup in every POSIX shell. Also scope the changelog claim: EXIT-trap/background output lands after the completion marker by design and stays visible via handle.output(). * docs(runtime): soften the fence-printf comment per review
174 unmodified linesExpand all175176177178179179180181182182183184185186187188189190184185186187188189192190191192193194195196139 unmodified linesExpand all336337338339340340341342343344193 unmodified linesExpand all5385395405415425435445425435445455465475485495515505515525535555545555575565575585595605615625635605615645655665675683 unmodified lines572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631645 unmodified linesExpand all174 unmodified linesExpand all records = await _poll_journal(journal, count=2, timeout=10) self.assertFalse(records[-1]["active"]) await _poll_group_dead(handle.pid)
async def test_delivered_status_wins_when_shell_dies_during_drain(self): async def test_delivered_status_wins_when_shell_dies_during_completion(self): entered = threading.Event() release = threading.Event() original = bash_module.BashHandle._drain_grace original = bash_module.BashHandle._wait_for_completion
def held_drain(handle_self): # Hold only the first caller (the reporter, which drains right after # reading status 0); any later caller must proceed normally. if not entered.is_set(): entered.set() release.wait(10) original(handle_self) def held_completion(handle_self): output = original(handle_self) entered.set() release.wait(10) return output
with mock.patch.object(bash_module.BashHandle, "_drain_grace", held_drain): with mock.patch.object( bash_module.BashHandle, "_wait_for_completion", held_completion ): # Background job keeps the shell alive in `wait` after status 0 is delivered. handle = bash("sleep 30 & true") try: self.assertTrue(await asyncio.to_thread(entered.wait, 5))139 unmodified linesExpand all if not bash_module._IS_POSIX: self.skipTest("POSIX-only gate") with tempfile.TemporaryDirectory() as tmp: marker = os.path.join(tmp, "ran") script = bash_module._status_script(f"touch {marker}") script = bash_module._status_script(f"touch {marker}", "a" * 32, "b" * 32) parent, child = socket.socketpair() proc = subprocess.Popen( [bash_module._shell(), "-c", script], stdin=child.fileno(),193 unmodified linesExpand all with self.assertRaisesRegex(RuntimeError, "PRIME_AGENT_BASH_SHELL"): bash_module._shell() which.assert_not_called()
async def test_pump_paused_between_read_and_commit_does_not_lose_output(self): # Reviewer repro: the chunk is out of the pipe (FIONREAD 0) but not yet # in the buffer; the drain fence must wait for the commit. async def test_pump_delayed_past_old_quiescence_bound_captures_all_output(self): # The ordered sentinel must wait through a pump delay beyond the old 500 ms bound. original_write = bash_module._BoundedBuffer.write delayed_once = threading.Event()
def delayed_write(buffer_self, chunk): if not delayed_once.is_set(): delayed_once.set() time.sleep(0.3) time.sleep(0.7) original_write(buffer_self, chunk)
with mock.patch.object(bash_module._BoundedBuffer, "write", delayed_write): result = await asyncio.wait_for(bash("printf between-read-and-write"), timeout=5) result = await asyncio.wait_for(bash("printf delayed-output-complete"), timeout=5) self.assertEqual(result.exit_code, 0) self.assertIn("between-read-and-write", result.output) self.assertEqual(result.output, "delayed-output-complete")
async def test_completion_marker_split_across_reads_is_removed(self): with mock.patch.object(bash_module, "_READ_CHUNK", 7): result = await asyncio.wait_for(bash("printf exact-pre-fence-output"), timeout=5) self.assertEqual(result.output, "exact-pre-fence-output")
async def test_slow_pump_does_not_lose_foreground_output(self): # A descheduled pump must not let _drain_grace conclude quiescence # while the shell's output still sits unread in the pipe. # Finalization waits for the pump to parse the ordered sentinel. original_pump = bash_module.BashHandle._pump
def slow_pump(handle_self): time.sleep(0.3)3 unmodified lines result = await asyncio.wait_for(bash("printf slow-pump-x"), timeout=5) self.assertEqual(result.exit_code, 0) self.assertIn("slow-pump-x", result.output)
async def test_sentinel_like_output_and_echoed_wrapper_do_not_truncate(self): token = "0123456789abcdef" * 4 raw_lookalike = "\x1eprime-agent-complete:not-this-invocation\x1f" command = ( "set -x\n" "printf '\\036prime-agent-complete:not-this-invocation\\037'\n" "if [ -r /proc/$$/cmdline ]; then cat /proc/$$/cmdline; fi\n" "printf '\\nafter-sentinel-lookalike\\n'" ) with mock.patch.object(bash_module.secrets, "token_hex", return_value=token): result = await asyncio.wait_for(bash(command), timeout=5) actual_marker = ( bash_module._COMPLETION_PREFIX + token.encode() + bash_module._COMPLETION_SUFFIX ) self.assertIn(raw_lookalike, result.output) self.assertIn("after-sentinel-lookalike", result.output) self.assertNotIn(actual_marker.decode(), result.output)
async def test_user_alias_cannot_replace_completion_emitter(self): if os.path.basename(bash_module._shell()) != "bash": self.skipTest("bash alias expansion semantics") handle = bash( "shopt -s expand_aliases; alias command='printf alias-expanded'; sleep 30 &" ) try: result = await asyncio.wait_for(handle, timeout=5) self.assertEqual(result.exit_code, 0) self.assertNotIn("alias-expanded", result.output) finally: handle.kill(signal.SIGKILL) await _poll_group_dead(handle.pid)
async def test_user_function_cannot_replace_completion_emitter(self): # The backslash in `\command` defeats alias expansion only: a shell # function named `command` would otherwise swallow both fence frames # and wedge the await behind the background job until shell death. handle = bash("command() { printf function-expanded; }; sleep 30 &") try: result = await asyncio.wait_for(handle, timeout=5) self.assertEqual(result.exit_code, 0) self.assertNotIn("function-expanded", result.output) finally: handle.kill(signal.SIGKILL) await _poll_group_dead(handle.pid)
async def test_shell_killed_before_sentinel_finalizes_from_exit(self): result = await asyncio.wait_for( bash("printf output-before-shell-kill; kill -KILL $$"), timeout=5 ) self.assertEqual(result.exit_code, -signal.SIGKILL) self.assertIn("output-before-shell-kill", result.output)
async def test_relative_bash_shell_override_rejected(self): with mock.patch.dict(os.environ, {"PRIME_AGENT_BASH_SHELL": "bash"}): with self.assertRaises(ValueError): bash("echo hi")645 unmodified linesExpand all1- Fixed `bash()` to capture all foreground command output before finalizing results by using an ordered per-command completion marker; output written after the marker (e.g. by `EXIT` traps or background jobs) is not in the awaited result but stays visible via `handle.output()`/`tail()`.4 unmodified lines567891011121321 unmodified lines353637383940414243444573 unmodified lines1191201211221231241251261271281291309 unmodified lines1401411421431441451461471489 unmodified lines158159160161153162163164165166167168169170171172173155174175176177178113 unmodified linesExpand all2922932942952772782962972982993003013023032872882893043053063073 unmodified lines3113123133143013153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653668 unmodified lines37537637737837938032832938138238338438538612 unmodified lines39940040140240340440540640754 unmodified lines46246346446541241341441546646746846947021 unmodified lines492493494495445496497498499500501451502503504505506152 unmodified linesExpand all65966066166266366466566666766866925 unmodified lines695696697698645646647648649650651652653699700701702703704705706707708709710711712713714655715716717718659719720721722723661662724725726727728729161 unmodified linesExpand all4 unmodified linesimport asyncioimport atexitimport jsonimport osimport secretsimport selectorsimport shutilimport signalimport socket21 unmodified lines_READ_CHUNK = 65536# Fixed child-side fd for the status channel; POSIX shells (notably dash) only# guarantee single-digit fds in redirection syntax._STATUS_FD = 9_OUTPUT_FD = 8_COMPLETION_PREFIX = b"\x1eprime-agent-complete:"_COMPLETION_SUFFIX = b"\x1f"# Cancelled one-shot awaits: TERM grace before the group KILL, then the bounded# wait for a confirmed group exit before CancelledError propagates._CANCEL_TERM_GRACE = 0.5_CANCEL_KILL_WAIT = 2.073 unmodified lines self.command = command self._buffer = _BoundedBuffer() self._done = threading.Event() self._eof = threading.Event() self._completion_terminal = threading.Event() self._completion_output: str | None = None self._completion_lock = threading.Lock() self._completion_pending = b"" self._status: int | None = None self._status_known = threading.Event() self._reaped = False self._result: BashResult | None = None9 unmodified lines self._wake_write = -1 # True only while the pump moves a chunk from the pipe into the buffer. self._pump_transfer = False self._job: int | None = None self._completion_marker: bytes | None = None status_write = -1 if _IS_POSIX: # Full-duplex status channel: the child end rides in as stdin (fd 0) # and the script remaps it to _STATUS_FD before swapping in /dev/null9 unmodified lines except BaseException: os.close(self._status_read) os.close(status_write) raise script = _status_script(_with_prefix(command)) completion_token = secrets.token_hex(32) # Halves stop passive echoes; a deliberate forgery freezes only this call while later bytes stay live. token_midpoint = len(completion_token) // 2 self._completion_marker = ( _COMPLETION_PREFIX + completion_token.encode("ascii") + _COMPLETION_SUFFIX ) script = _status_script( _with_prefix(command), completion_token[:token_midpoint], completion_token[token_midpoint:], ) else: # Windows: the child is created suspended, atomically inside the kill-on-close job. # Windows lacks a foreground-status channel, so its exit drain stays best-effort. script = _with_prefix(command) self._job = _winjob.create_job() if self._job is None: # Nothing spawned yet, so nothing can leak: refuse to start.113 unmodified linesExpand all def _pump(self) -> None: stdout = self._proc.stdout assert stdout is not None if not _IS_POSIX: # Windows: no FIONREAD fence; the drain keeps its quiescence # heuristic (best-effort parity). try: while chunk := stdout.read1(_READ_CHUNK): self._buffer.write(chunk) except (OSError, ValueError): pass stdout.close() self._eof.set() return # POSIX: select-gate the read and flag the read->commit window, so the # drain fence never sees "pipe empty + buffer quiescent" while a chunk # is in flight between the pipe read and the buffer commit. fd = stdout.fileno() try: with selectors.DefaultSelector() as sel: sel.register(fd, selectors.EVENT_READ)3 unmodified lines try: chunk = os.read(fd, _READ_CHUNK) if not chunk: break self._buffer.write(chunk) self._consume_output(chunk) finally: self._pump_transfer = False except (OSError, ValueError): pass self._abandon_completion() try: stdout.close() except OSError: pass self._eof.set()
def _consume_output(self, chunk: bytes) -> None: marker = self._completion_marker assert marker is not None with self._completion_lock: if self._completion_terminal.is_set(): self._buffer.write(chunk) return data = self._completion_pending + chunk marker_at = data.find(marker) if marker_at >= 0: self._buffer.write(data[:marker_at]) self._completion_pending = b"" self._completion_output = self._buffer.text() self._completion_terminal.set() self._buffer.write(data[marker_at + len(marker) :]) return retained = 0 for size in range(min(len(data), len(marker) - 1), 0, -1): if data.endswith(marker[:size]): retained = size break self._buffer.write(data[:-retained] if retained else data) self._completion_pending = data[-retained:] if retained else b""
def _abandon_completion(self) -> None: with self._completion_lock: if self._completion_terminal.is_set(): return self._buffer.write(self._completion_pending) self._completion_pending = b"" self._completion_terminal.set()
def _wait_for_completion(self) -> str | None: self._completion_terminal.wait() return self._completion_output
def _report(self) -> None: # Finalize at foreground completion (status channel), not EOF, so # `cmd &` does not hang the await; the shell then `wait`s for its # background jobs, keeping the journaled group identity alive.8 unmodified lines # _watch blocks on this event without a timeout, so every exit path # (parsed status, EOF, garbage, exception) must set it. self._status_known.set() if status is not None: output = self._wait_for_completion() if output is None: self._drain_grace() self._finalize(status) self._drain_grace() self._finalize(status, output)
def _watch(self) -> None: # Observe shell death independently of the status socket: an early # `exit`/`exec`/`set -e`/fatal signal skips `printf`, and background12 unmodified lines self._status_known.wait() with self._callback_lock: delivered = self._status if delivered is None and not self._done.is_set(): self._abandon_completion() self._drain_grace() self._finalize(exit_code) with self._kill_lock: delivered = self._reap_group()54 unmodified lines os.close(self._status_read) os.close(self._wake_read)
def _drain_grace(self) -> None: # Bounded wait so the result includes foreground output still in the pipe: # EOF arrives immediately without background jobs, otherwise stop once the # buffer is quiescent for one tick AND the pipe holds no unread bytes (a # slow pump must not lose output the shell wrote before its status). # Best-effort fallback when process exit/EOF arrives without a sentinel. deadline = time.monotonic() + 0.5 size = self._buffer.size() while time.monotonic() < deadline: if self._eof.wait(0.05):21 unmodified lines except (OSError, ValueError): return False return pending > 0
def _finalize(self, exit_code: int) -> None: def _finalize(self, exit_code: int, output: str | None = None) -> None: with self._callback_lock: if self._done.is_set(): return self._result = BashResult( exit_code=exit_code, output=self._buffer.text(), output=self._buffer.text() if output is None else output, duration=time.monotonic() - self._started, ) self._done.set() callbacks = self._callbacks152 unmodified linesExpand all containment is per-platform: process groups plus the orphan journal on POSIX; a kill-on-close job object on Windows entered while the child is still suspended, so no descendant can escape it and kill()/crash cleanup are unconditional -- bash() raises if containment cannot be established. Output written after the completion fence (e.g. by an EXIT trap or a background job) is not in BashResult.output but stays visible via handle.output()/tail(). """ if not isinstance(command, str) or not command: raise TypeError("command must be a non-empty str") _install_shutdown_hook()25 unmodified lines prefix = os.environ.get("PRIME_AGENT_BASH_COMMAND_PREFIX") return f"{prefix}\n{command}" if prefix else command
def _status_script(command: str) -> str: # The status socket arrives as stdin (fd 0); the prologue dups it to the # single-digit _STATUS_FD (dash rejects multi-digit fds in redirections at # parse time) and points stdin at /dev/null, so no other copy remains. The # gate read blocks until the parent has journaled the pid; EOF (parent died # first) exits without running the command. The brace group runs the command # with the status fd closed so `&` children do not inherit it; the trailing # `wait` keeps the shell alive as group leader until its own jobs exit # (double-forked daemons stay out of scope).def _fence_printf() -> str: # `\command -p printf` defeats alias expansion but not a user-defined shell # function named `command`, which would swallow both fence frames and leave # the await hanging until the shell dies (wedged behind background jobs). A # slash-qualified command name bypasses function and alias lookup for # ordinary command names, so resolve printf on the system default utility PATH. path = shutil.which("printf", path=os.confstr("CS_PATH") or os.defpath) if path and "'" not in path: return f"'{path}'" return "\\command -p printf"
def _status_script(command: str, completion_a: str, completion_b: str) -> str: # Closed control fds preserve background behavior; supported shells atomically write the frame. emit = _fence_printf() return ( f"exec {_STATUS_FD}>&0 0</dev/null\n" f"exec {_STATUS_FD}>&0 {_OUTPUT_FD}>&1 0</dev/null\n" f"read -r _prime_agent_gate <&{_STATUS_FD} || exit 127\n" "{\n" f"{command}\n" f"}} {_STATUS_FD}>&-\n" f"}} {_OUTPUT_FD}>&- {_STATUS_FD}>&-\n" "__prime_status=$?\n" "\\set +x\n" f"{emit} '\\036prime-agent-complete:%s%s\\037' " f"'{completion_a}' '{completion_b}' >&{_OUTPUT_FD} || exit \"$__prime_status\"\n" f"printf '%s\\n' \"$__prime_status\" >&{_STATUS_FD}\n" f"exec {_STATUS_FD}>&-\n" f"{emit} '%s\\n' \"$__prime_status\" >&{_STATUS_FD}\n" f"exec {_OUTPUT_FD}>&- {_STATUS_FD}>&-\n" "wait\n" 'exit "$__prime_status"\n' )
161 unmodified linesExpand all