From 83354cb3c9d001a6eba4848c5361603d801802bc Mon Sep 17 00:00:00 2001 From: mohamed-elkholy95 Date: Thu, 28 May 2026 23:23:54 -0400 Subject: [PATCH 01/12] fix(todo): parse JSON-encoded string passed as todos list LLMs occasionally serialize the todos array as a JSON string instead of a proper JSON array, causing Pydantic validation to fail with "Input should be a valid list". Add a before-validator that transparently parses the string via json.loads when detected. --- src/pythinker_code/tools/todo/__init__.py | 13 +++++++++++- tests/tools/test_todo.py | 24 +++++++++++++++++++++++ 2 files changed, 36 insertions(+), 1 deletion(-) diff --git a/src/pythinker_code/tools/todo/__init__.py b/src/pythinker_code/tools/todo/__init__.py index d9f9f7fc..ab767aad 100644 --- a/src/pythinker_code/tools/todo/__init__.py +++ b/src/pythinker_code/tools/todo/__init__.py @@ -2,7 +2,7 @@ from pathlib import Path from typing import Any, Literal, cast, override -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, field_validator from pythinker_core.tooling import CallableTool2, ToolReturnValue from pythinker_code.session_state import TodoItemState @@ -26,6 +26,17 @@ class Params(BaseModel): ), ) + @field_validator("todos", mode="before") + @classmethod + def _parse_todos_string(cls, v: Any) -> Any: + # LLMs occasionally pass the list as a JSON-encoded string; parse it transparently. + if isinstance(v, str): + try: + return json.loads(v) + except json.JSONDecodeError: + pass + return v + class SetTodoList(CallableTool2[Params]): name: str = "SetTodoList" diff --git a/tests/tools/test_todo.py b/tests/tools/test_todo.py index 9a5173ce..070e3473 100644 --- a/tests/tools/test_todo.py +++ b/tests/tools/test_todo.py @@ -15,6 +15,30 @@ def set_todo_list_tool(runtime: Runtime) -> SetTodoList: return SetTodoList(runtime) +class TestParamsJsonStringCoercion: + """Regression: LLM occasionally passes todos as a JSON-encoded string instead of a list.""" + + def test_todos_as_json_string_is_parsed(self): + """Params must accept todos as a JSON string and coerce it to a list.""" + import json + + raw = json.dumps([{"title": "Explore agent", "status": "pending"}]) + params = Params(todos=raw) # type: ignore[arg-type] + assert params.todos is not None + assert len(params.todos) == 1 + assert params.todos[0].title == "Explore agent" + assert params.todos[0].status == "pending" + + def test_todos_as_normal_list_still_works(self): + params = Params(todos=[Todo(title="Normal task", status="done")]) + assert params.todos is not None + assert params.todos[0].title == "Normal task" + + def test_todos_none_still_works(self): + params = Params(todos=None) + assert params.todos is None + + class TestSetTodoListOutputNotEmpty: """Regression test for issue #1710: SetTodoList storm. From 107fc206bb59314c3680bd563b1597ace02307f2 Mon Sep 17 00:00:00 2001 From: mohamed-elkholy95 Date: Fri, 29 May 2026 05:03:43 -0400 Subject: [PATCH 02/12] fix: harden background task sync and reliability across subsystems Validated subset of the TUI/background-agent reliability scan. Changes: - background: serialise the runtime read-modify-write in every _mark_task_* and in recover() under a cross-process per-task lock (store._runtime_lock + _write_runtime_unlocked) so a worker heartbeat landing mid-sequence is no longer lost; add a SIGTERM->SIGKILL escalation fallback in _best_effort_kill that only fires if the task is still running; cap a bash task's output.log at config.background.max_output_bytes (default 50 MiB) so a chatty task cannot exhaust disk. - pythinker-host: skip the process-group kill once the child has exited and been reaped, since the OS may have recycled its pid/pgid. - web fetch: follow redirects manually and re-validate every hop against the SSRF guard, closing a public->link-local (metadata endpoint) redirect bypass. - grep fallback: bound the pure-Python search with a wall-clock deadline mirroring the ripgrep timeout and report partial results. - browser launch: route OAuth/feedback URL opens through a detached open_url_in_browser() so browser chatter cannot corrupt the TUI or consume key presses meant for it. - cli: restore the terminal to a sane state on SIGTERM/SIGQUIT and via atexit. - live view: use the theme "warning" token instead of a hardcoded accent. - mcp/toolset: annotate the fastmcp OAuth provider as Any for pyright. Tests added/updated across the background, tools, ui, and host suites. --- .../src/pythinker_host/local.py | 5 + .../pythinker-host/tests/test_local_host.py | 57 +++++ src/pythinker_code/auth/github_feedback.py | 5 +- src/pythinker_code/auth/oauth.py | 5 +- src/pythinker_code/auth/openai.py | 5 +- src/pythinker_code/background/manager.py | 212 +++++++++++------- src/pythinker_code/background/store.py | 30 +++ src/pythinker_code/background/worker.py | 54 ++++- src/pythinker_code/cli/__init__.py | 22 ++ src/pythinker_code/cli/mcp.py | 2 +- src/pythinker_code/config.py | 4 + src/pythinker_code/soul/toolset.py | 2 +- src/pythinker_code/tools/file/grep_local.py | 13 ++ src/pythinker_code/tools/web/fetch.py | 119 ++++++---- src/pythinker_code/ui/shell/slash.py | 9 +- .../ui/shell/visualize/_live_view.py | 8 +- src/pythinker_code/utils/term.py | 29 +++ tests/background/test_manager.py | 115 +++++++++- tests/background/test_worker.py | 43 ++++ tests/tools/test_fetch_url.py | 92 +++++++- tests/tools/test_grep.py | 25 +++ tests/ui_and_conv/test_live_view_todos.py | 8 +- .../ui_and_conv/test_shell_feedback_slash.py | 6 +- 23 files changed, 724 insertions(+), 146 deletions(-) diff --git a/packages/pythinker-host/src/pythinker_host/local.py b/packages/pythinker-host/src/pythinker_host/local.py index 461c9d5d..35689a69 100644 --- a/packages/pythinker-host/src/pythinker_host/local.py +++ b/packages/pythinker-host/src/pythinker_host/local.py @@ -68,6 +68,11 @@ async def wait(self) -> int: return await self._process.wait() async def kill(self) -> None: + # If the process has already exited (and been reaped), its pid/pgid + # may have been recycled by the OS; signaling it could hit an + # unrelated process group. Skip the group-kill in that case. + if self._process.returncode is not None: + return if os.name != "nt": try: os.killpg(os.getpgid(self._process.pid), signal.SIGKILL) diff --git a/packages/pythinker-host/tests/test_local_host.py b/packages/pythinker-host/tests/test_local_host.py index 591cc595..0c42bb36 100644 --- a/packages/pythinker-host/tests/test_local_host.py +++ b/packages/pythinker-host/tests/test_local_host.py @@ -2,6 +2,7 @@ import asyncio import os +import signal import sys from collections.abc import Generator from pathlib import Path, PurePosixPath, PureWindowsPath @@ -196,3 +197,59 @@ async def test_exec_wait_timeout(local_host: LocalHost): if process.returncode is None: await process.kill() await process.wait() + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX process-group signal path") +async def test_kill_skips_signal_after_process_exit( + local_host: LocalHost, monkeypatch: pytest.MonkeyPatch +): + """Once a process has exited (and been reaped), its pid/pgid may be recycled + by the OS, so kill() must not send a process-group signal.""" + import pythinker_host.local as local_module + + process = await local_host.exec(*_python_code_args("import sys; sys.exit(0)")) + await process.wait() + assert process.returncode is not None + + # Fail loudly if kill() reaches the signal path at all: with the returncode + # guard it must short-circuit before even resolving the process group. + calls: list[str] = [] + + def _record_getpgid(pid: int) -> int: + calls.append("getpgid") + return 0 + + def _record_killpg(pgid: int, sig: int) -> None: + calls.append("killpg") + + monkeypatch.setattr(local_module.os, "getpgid", _record_getpgid) + monkeypatch.setattr(local_module.os, "killpg", _record_killpg) + + await process.kill() + + assert calls == [] + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX process-group signal path") +async def test_kill_signals_running_process( + local_host: LocalHost, monkeypatch: pytest.MonkeyPatch +): + """A still-running process is killed via its process group.""" + import pythinker_host.local as local_module + + process = await local_host.exec(*_python_code_args("import time; time.sleep(30)")) + assert process.returncode is None + + real_killpg = local_module.os.killpg + sent: list[int] = [] + + def _record_killpg(pgid: int, sig: int) -> None: + sent.append(sig) + real_killpg(pgid, sig) # actually terminate so the process is not leaked + + monkeypatch.setattr(local_module.os, "killpg", _record_killpg) + + await process.kill() + await process.wait() + + assert sent and sent[0] == signal.SIGKILL diff --git a/src/pythinker_code/auth/github_feedback.py b/src/pythinker_code/auth/github_feedback.py index 9f01ee6b..d3a07748 100644 --- a/src/pythinker_code/auth/github_feedback.py +++ b/src/pythinker_code/auth/github_feedback.py @@ -2,7 +2,6 @@ import asyncio import time -import webbrowser from dataclasses import dataclass from typing import Any, cast @@ -150,7 +149,9 @@ async def login_github_feedback( ) if open_browser: try: - webbrowser.open(auth.verification_uri) + from pythinker_code.utils.term import open_url_in_browser + + open_url_in_browser(auth.verification_uri) except Exception as exc: logger.warning("Failed to open browser: {error}", error=exc) yield OAuthEvent("waiting", "Waiting for GitHub authorization...") diff --git a/src/pythinker_code/auth/oauth.py b/src/pythinker_code/auth/oauth.py index 89495e33..07c8fff9 100644 --- a/src/pythinker_code/auth/oauth.py +++ b/src/pythinker_code/auth/oauth.py @@ -10,7 +10,6 @@ import tempfile import time import uuid -import webbrowser from collections.abc import AsyncGenerator, AsyncIterator from contextlib import asynccontextmanager, suppress from dataclasses import dataclass @@ -660,7 +659,9 @@ async def login_pythinker_code( ) if open_browser: try: - webbrowser.open(auth.verification_uri_complete) + from pythinker_code.utils.term import open_url_in_browser + + open_url_in_browser(auth.verification_uri_complete) except Exception as exc: logger.warning("Failed to open browser: {error}", error=exc) diff --git a/src/pythinker_code/auth/openai.py b/src/pythinker_code/auth/openai.py index 0bbd4235..488d33a5 100644 --- a/src/pythinker_code/auth/openai.py +++ b/src/pythinker_code/auth/openai.py @@ -7,7 +7,6 @@ import json import secrets import time -import webbrowser from collections.abc import AsyncIterator from dataclasses import dataclass from typing import Any, cast @@ -329,7 +328,9 @@ async def _wait_for_browser_code(open_browser: bool = True) -> tuple[str, str, s auth_url = _build_authorize_url(redirect_uri=redirect_uri, pkce=pkce, state=state) if open_browser: - webbrowser.open(auth_url) + from pythinker_code.utils.term import open_url_in_browser + + open_url_in_browser(auth_url) try: code, error = await asyncio.wait_for(result, timeout=15 * 60) diff --git a/src/pythinker_code/background/manager.py b/src/pythinker_code/background/manager.py index 2af7f51c..be561464 100644 --- a/src/pythinker_code/background/manager.py +++ b/src/pythinker_code/background/manager.py @@ -5,6 +5,7 @@ import signal import subprocess import sys +import threading import time from pathlib import Path from typing import TYPE_CHECKING, Any @@ -158,6 +159,8 @@ def _worker_command(self, task_dir: Path) -> list[str]: str(self._config.wait_poll_interval_ms), "--kill-grace-period-ms", str(self._config.kill_grace_period_ms), + "--max-output-bytes", + str(self._config.max_output_bytes), ] return [ sys.executable, @@ -172,6 +175,8 @@ def _worker_command(self, task_dir: Path) -> list[str]: str(self._config.wait_poll_interval_ms), "--kill-grace-period-ms", str(self._config.kill_grace_period_ms), + "--max-output-bytes", + str(self._config.max_output_bytes), ] def _launch_worker(self, task_dir: Path) -> int: @@ -417,7 +422,9 @@ async def wait(self, task_id: str, *, timeout_s: int = 30) -> TaskView: return view await asyncio.sleep(self._config.wait_poll_interval_ms / 1000) - def _best_effort_kill(self, runtime: TaskRuntime) -> None: + _SIGKILL_ESCALATION_DELAY_S = 5.0 + + def _best_effort_kill(self, task_id: str, runtime: TaskRuntime) -> None: try: if os.name == "nt": pid = runtime.child_pid or runtime.worker_pid @@ -433,14 +440,52 @@ def _best_effort_kill(self, runtime: TaskRuntime) -> None: if runtime.child_pgid is not None: os.killpg(runtime.child_pgid, signal.SIGTERM) + self._schedule_sigkill(task_id, pgid=runtime.child_pgid) return if runtime.child_pid is not None: os.kill(runtime.child_pid, signal.SIGTERM) + self._schedule_sigkill(task_id, pid=runtime.child_pid) except ProcessLookupError: pass except Exception: logger.exception("Failed to send best-effort kill signal") + def _schedule_sigkill( + self, task_id: str, *, pgid: int | None = None, pid: int | None = None + ) -> None: + """Escalate to SIGKILL after a grace delay if the task is still running.""" + t = threading.Timer( + self._SIGKILL_ESCALATION_DELAY_S, + self._escalate_sigkill, + kwargs={"task_id": task_id, "pgid": pgid, "pid": pid}, + ) + t.daemon = True + t.start() + + def _escalate_sigkill( + self, task_id: str, *, pgid: int | None = None, pid: int | None = None + ) -> None: + """SIGKILL the process group/pid, but only if the task is still running. + + Re-reading the task status first avoids two problems with a blind timer: + (a) sending a needless SIGKILL on the common path where SIGTERM already + worked, and (b) signaling a pgid/pid the OS may have recycled once the + original child exited and was reaped by its parent worker. The worker + owns the child and runs its own SIGTERM->SIGKILL escalation, so this is + only a fallback for when the worker itself is gone. + """ + try: + if is_terminal_status(self._store.read_runtime(task_id).status): + return + if pgid is not None: + os.killpg(pgid, signal.SIGKILL) + elif pid is not None: + os.kill(pid, signal.SIGKILL) + except ProcessLookupError: + pass + except Exception: + logger.warning("Failed to escalate kill signal to SIGKILL") + def kill(self, task_id: str, *, reason: str = "Killed by user") -> TaskView: self._ensure_root() view = self._store.merged_view(task_id) @@ -471,7 +516,7 @@ def kill(self, task_id: str, *, reason: str = "Killed by user") -> TaskView: } ) self._store.write_control(task_id, control) - self._best_effort_kill(view.runtime) + self._best_effort_kill(task_id, view.runtime) return self._store.merged_view(task_id) def kill_all_active(self, *, reason: str = "CLI session ended") -> list[str]: @@ -529,34 +574,37 @@ def recover(self) -> None: if now - last_progress_at <= stale_after: continue - # Re-read runtime to narrow the race window with the worker process. - fresh_runtime = self._store.read_runtime(view.spec.id) - if is_terminal_status(fresh_runtime.status): - continue - fresh_progress = ( - fresh_runtime.heartbeat_at - or fresh_runtime.started_at - or fresh_runtime.updated_at - or view.spec.created_at - ) - if now - fresh_progress <= stale_after: - continue - - runtime = fresh_runtime.model_copy() - runtime.finished_at = now - runtime.updated_at = now - if view.control.kill_requested_at is not None: - runtime.status = "killed" - runtime.interrupted = True - runtime.failure_reason = view.control.kill_reason or "Killed during recovery" - else: - runtime.status = "lost" - runtime.failure_reason = ( - "Background worker never heartbeat after startup" - if fresh_runtime.heartbeat_at is None - else "Background worker heartbeat expired" + # Hold the cross-process lock for the read-then-write to eliminate + # the race with a worker that heartbeats between our staleness check + # and the status write. + with self._store._runtime_lock(view.spec.id): # pyright: ignore[reportPrivateUsage] + fresh_runtime = self._store.read_runtime(view.spec.id) + if is_terminal_status(fresh_runtime.status): + continue + fresh_progress = ( + fresh_runtime.heartbeat_at + or fresh_runtime.started_at + or fresh_runtime.updated_at + or view.spec.created_at ) - self._store.write_runtime(view.spec.id, runtime) + if now - fresh_progress <= stale_after: + continue + + runtime = fresh_runtime.model_copy() + runtime.finished_at = now + runtime.updated_at = now + if view.control.kill_requested_at is not None: + runtime.status = "killed" + runtime.interrupted = True + runtime.failure_reason = view.control.kill_reason or "Killed during recovery" + else: + runtime.status = "lost" + runtime.failure_reason = ( + "Background worker never heartbeat after startup" + if fresh_runtime.heartbeat_at is None + else "Background worker heartbeat expired" + ) + self._store._write_runtime_unlocked(view.spec.id, runtime) # pyright: ignore[reportPrivateUsage] def reconcile(self, *, limit: int | None = None) -> list[str]: self.recover() @@ -637,33 +685,36 @@ def publish_terminal_notifications(self, *, limit: int | None = None) -> list[st return published def _mark_task_running(self, task_id: str) -> None: - runtime = self._store.read_runtime(task_id) - if is_terminal_status(runtime.status): - return - runtime.status = "running" - runtime.updated_at = time.time() - runtime.heartbeat_at = runtime.updated_at - runtime.failure_reason = None - self._store.write_runtime(task_id, runtime) + with self._store._runtime_lock(task_id): # pyright: ignore[reportPrivateUsage] + runtime = self._store.read_runtime(task_id) + if is_terminal_status(runtime.status): + return + runtime.status = "running" + runtime.updated_at = time.time() + runtime.heartbeat_at = runtime.updated_at + runtime.failure_reason = None + self._store._write_runtime_unlocked(task_id, runtime) # pyright: ignore[reportPrivateUsage] def _mark_task_awaiting_approval(self, task_id: str, reason: str) -> None: - runtime = self._store.read_runtime(task_id) - if is_terminal_status(runtime.status): - return - runtime.status = "awaiting_approval" - runtime.updated_at = time.time() - runtime.failure_reason = reason - self._store.write_runtime(task_id, runtime) + with self._store._runtime_lock(task_id): # pyright: ignore[reportPrivateUsage] + runtime = self._store.read_runtime(task_id) + if is_terminal_status(runtime.status): + return + runtime.status = "awaiting_approval" + runtime.updated_at = time.time() + runtime.failure_reason = reason + self._store._write_runtime_unlocked(task_id, runtime) # pyright: ignore[reportPrivateUsage] def _mark_task_completed(self, task_id: str) -> None: - runtime = self._store.read_runtime(task_id) - if is_terminal_status(runtime.status): - return - runtime.status = "completed" - runtime.updated_at = time.time() - runtime.finished_at = runtime.updated_at - runtime.failure_reason = None - self._store.write_runtime(task_id, runtime) + with self._store._runtime_lock(task_id): # pyright: ignore[reportPrivateUsage] + runtime = self._store.read_runtime(task_id) + if is_terminal_status(runtime.status): + return + runtime.status = "completed" + runtime.updated_at = time.time() + runtime.finished_at = runtime.updated_at + runtime.failure_reason = None + self._store._write_runtime_unlocked(task_id, runtime) # pyright: ignore[reportPrivateUsage] from pythinker_code.telemetry import track if runtime.started_at and runtime.finished_at: @@ -671,14 +722,15 @@ def _mark_task_completed(self, task_id: str) -> None: track("background_task_completed", success=True, duration_s=duration) def _mark_task_failed(self, task_id: str, reason: str) -> None: - runtime = self._store.read_runtime(task_id) - if is_terminal_status(runtime.status): - return - runtime.status = "failed" - runtime.updated_at = time.time() - runtime.finished_at = runtime.updated_at - runtime.failure_reason = reason - self._store.write_runtime(task_id, runtime) + with self._store._runtime_lock(task_id): # pyright: ignore[reportPrivateUsage] + runtime = self._store.read_runtime(task_id) + if is_terminal_status(runtime.status): + return + runtime.status = "failed" + runtime.updated_at = time.time() + runtime.finished_at = runtime.updated_at + runtime.failure_reason = reason + self._store._write_runtime_unlocked(task_id, runtime) # pyright: ignore[reportPrivateUsage] from pythinker_code.telemetry import track if runtime.started_at and runtime.finished_at: @@ -691,16 +743,17 @@ def _mark_task_failed(self, task_id: str, reason: str) -> None: ) def _mark_task_timed_out(self, task_id: str, reason: str) -> None: - runtime = self._store.read_runtime(task_id) - if is_terminal_status(runtime.status): - return - runtime.status = "failed" - runtime.updated_at = time.time() - runtime.finished_at = runtime.updated_at - runtime.interrupted = True - runtime.timed_out = True - runtime.failure_reason = reason - self._store.write_runtime(task_id, runtime) + with self._store._runtime_lock(task_id): # pyright: ignore[reportPrivateUsage] + runtime = self._store.read_runtime(task_id) + if is_terminal_status(runtime.status): + return + runtime.status = "failed" + runtime.updated_at = time.time() + runtime.finished_at = runtime.updated_at + runtime.interrupted = True + runtime.timed_out = True + runtime.failure_reason = reason + self._store._write_runtime_unlocked(task_id, runtime) # pyright: ignore[reportPrivateUsage] from pythinker_code.telemetry import track if runtime.started_at and runtime.finished_at: @@ -713,15 +766,16 @@ def _mark_task_timed_out(self, task_id: str, reason: str) -> None: ) def _mark_task_killed(self, task_id: str, reason: str) -> None: - runtime = self._store.read_runtime(task_id) - if is_terminal_status(runtime.status): - return - runtime.status = "killed" - runtime.updated_at = time.time() - runtime.finished_at = runtime.updated_at - runtime.interrupted = True - runtime.failure_reason = reason - self._store.write_runtime(task_id, runtime) + with self._store._runtime_lock(task_id): # pyright: ignore[reportPrivateUsage] + runtime = self._store.read_runtime(task_id) + if is_terminal_status(runtime.status): + return + runtime.status = "killed" + runtime.updated_at = time.time() + runtime.finished_at = runtime.updated_at + runtime.interrupted = True + runtime.failure_reason = reason + self._store._write_runtime_unlocked(task_id, runtime) # pyright: ignore[reportPrivateUsage] from pythinker_code.telemetry import track if runtime.started_at and runtime.finished_at: diff --git a/src/pythinker_code/background/store.py b/src/pythinker_code/background/store.py index 9f5391bc..83904217 100644 --- a/src/pythinker_code/background/store.py +++ b/src/pythinker_code/background/store.py @@ -1,8 +1,10 @@ from __future__ import annotations +import contextlib import json import os import re +from contextlib import contextmanager from pathlib import Path from pydantic import BaseModel, ValidationError @@ -108,7 +110,35 @@ def write_spec(self, spec: TaskSpec) -> None: def read_spec(self, task_id: str) -> TaskSpec: return TaskSpec.model_validate_json(self.spec_path(task_id).read_text(encoding="utf-8")) + @contextmanager + def _runtime_lock(self, task_id: str): + """Exclusive cross-process lock for the task's runtime file. + + Serialises concurrent reads and writes between the manager and worker + processes, eliminating the race between heartbeat updates and stale-task + recovery. Falls back to a no-op on platforms where fcntl is unavailable + (Windows). + """ + lock_path = self.task_path(task_id) / "runtime.lock" + try: + import fcntl + except ImportError: + yield + return + with open(lock_path, "a") as lock_fd: + fcntl.flock(lock_fd, fcntl.LOCK_EX) + try: + yield + finally: + with contextlib.suppress(OSError): + fcntl.flock(lock_fd, fcntl.LOCK_UN) + def write_runtime(self, task_id: str, runtime: TaskRuntime) -> None: + with self._runtime_lock(task_id): + self._write_runtime_unlocked(task_id, runtime) + + def _write_runtime_unlocked(self, task_id: str, runtime: TaskRuntime) -> None: + """Write runtime without acquiring the per-task lock (caller holds it).""" path = self.runtime_path(task_id) if path.exists(): current = self.read_runtime(task_id) diff --git a/src/pythinker_code/background/worker.py b/src/pythinker_code/background/worker.py index 34592e97..bac930b7 100644 --- a/src/pythinker_code/background/worker.py +++ b/src/pythinker_code/background/worker.py @@ -34,6 +34,7 @@ async def run_background_task_worker( heartbeat_interval_ms: int = 5000, control_poll_interval_ms: int = 500, kill_grace_period_ms: int = 2000, + max_output_bytes: int = 0, ) -> None: task_dir = task_dir.expanduser().resolve() task_id = task_dir.name @@ -74,6 +75,8 @@ async def run_background_task_worker( kill_sent_at: float | None = None timed_out = False timeout_reason: str | None = None + output_limit_exceeded = False + output_limit_reason: str | None = None async def _heartbeat_loop() -> None: while not stop_event.is_set(): @@ -104,10 +107,54 @@ async def _terminate_process(force: bool = False) -> None: except ProcessLookupError: pass + output_path = store.output_path(task_id) + + async def _check_output_limit() -> None: + """Terminate the task if its output.log grew past ``max_output_bytes``. + + Writes a single marker line and records a failure the first time the + limit is hit; the ``output_limit_exceeded`` guard keeps it from + re-marking on subsequent polls or once the process is already exiting. + """ + nonlocal output_limit_exceeded, output_limit_reason + if max_output_bytes <= 0 or output_limit_exceeded: + return + if process is None or process.returncode is not None: + return + try: + size = output_path.stat().st_size + except OSError: + return + if size <= max_output_bytes: + return + + output_limit_exceeded = True + output_limit_reason = f"Output exceeded max_output_bytes ({max_output_bytes})" + marker = f"\n... output limit exceeded ({size} bytes); task terminated ...\n" + with contextlib.suppress(OSError), output_path.open("ab") as marker_file: + marker_file.write(marker.encode("utf-8")) + with store._runtime_lock(task_id): # pyright: ignore[reportPrivateUsage] + current = store.read_runtime(task_id) + if not current.finished_at: + current.status = "failed" + current.interrupted = True + current.failure_reason = output_limit_reason + current.updated_at = time.time() + store._write_runtime_unlocked(task_id, current) # pyright: ignore[reportPrivateUsage] + await _terminate_process(force=False) + async def _control_loop() -> None: nonlocal kill_sent_at while not stop_event.is_set(): await asyncio.sleep(control_poll_interval_ms / 1000) + await _check_output_limit() + if output_limit_exceeded and ( + kill_sent_at is not None + and process is not None + and process.returncode is None + and time.time() - kill_sent_at >= kill_grace_period_ms / 1000 + ): + await _terminate_process(force=True) current_control: TaskControl = store.read_control(task_id) if current_control.kill_requested_at is not None: await _terminate_process(force=current_control.force) @@ -137,7 +184,6 @@ async def _input_loop() -> None: return try: - output_path = store.output_path(task_id) with output_path.open("ab") as output_file: spawn_kwargs: dict[str, Any] = { "stdin": asyncio.subprocess.PIPE, @@ -214,7 +260,11 @@ async def _input_loop() -> None: runtime.updated_at = runtime.finished_at runtime.exit_code = returncode runtime.heartbeat_at = runtime.finished_at - if timed_out: + if output_limit_exceeded: + runtime.status = "failed" + runtime.interrupted = True + runtime.failure_reason = output_limit_reason + elif timed_out: runtime.status = "failed" runtime.interrupted = True runtime.timed_out = True diff --git a/src/pythinker_code/cli/__init__.py b/src/pythinker_code/cli/__init__.py index 95e401d7..83b82076 100644 --- a/src/pythinker_code/cli/__init__.py +++ b/src/pythinker_code/cli/__init__.py @@ -1041,6 +1041,26 @@ async def _pick_session() -> str: session_id = asyncio.run(_pick_session()) + # Ensure the terminal is restored to a sane state if the process is killed + # by SIGTERM or SIGQUIT while the keyboard listener has raw mode active. + # atexit covers SystemExit/normal exits; the signal handlers cover signals + # whose default action would bypass atexit entirely. + import atexit + import signal as _signal + + from pythinker_code.utils.term import ensure_tty_sane + + atexit.register(ensure_tty_sane) + + def _restore_term_and_exit(signum: int, frame: object) -> None: + ensure_tty_sane() + _signal.signal(signum, _signal.SIG_DFL) + os.kill(os.getpid(), signum) + + for _sig in (_signal.SIGTERM, _signal.SIGQUIT): + with contextlib.suppress(OSError, ValueError): + _signal.signal(_sig, _restore_term_and_exit) + try: switch_target, exit_code = asyncio.run(_reload_loop(session_id)) except (typer.BadParameter, typer.Exit): @@ -1336,6 +1356,7 @@ def background_task_worker( heartbeat_interval_ms: Annotated[int, typer.Option("--heartbeat-interval-ms")] = 5000, control_poll_interval_ms: Annotated[int, typer.Option("--control-poll-interval-ms")] = 500, kill_grace_period_ms: Annotated[int, typer.Option("--kill-grace-period-ms")] = 2000, + max_output_bytes: Annotated[int, typer.Option("--max-output-bytes")] = 0, ) -> None: """Run background task worker subprocess (internal).""" import asyncio @@ -1354,6 +1375,7 @@ def background_task_worker( heartbeat_interval_ms=heartbeat_interval_ms, control_poll_interval_ms=control_poll_interval_ms, kill_grace_period_ms=kill_grace_period_ms, + max_output_bytes=max_output_bytes, ) ) diff --git a/src/pythinker_code/cli/mcp.py b/src/pythinker_code/cli/mcp.py index 80961622..7e4e3ed6 100644 --- a/src/pythinker_code/cli/mcp.py +++ b/src/pythinker_code/cli/mcp.py @@ -214,7 +214,7 @@ def _oauth_token_storage(server_url: str) -> Any: file_token_storage = getattr(fastmcp_oauth, "FileTokenStorage", None) if file_token_storage is not None: return file_token_storage(server_url=server_url) - provider = fastmcp_oauth.OAuth(mcp_url=server_url) + provider: Any = fastmcp_oauth.OAuth(mcp_url=server_url) return provider.token_storage_adapter diff --git a/src/pythinker_code/config.py b/src/pythinker_code/config.py index 7a725a19..d700d90e 100644 --- a/src/pythinker_code/config.py +++ b/src/pythinker_code/config.py @@ -116,6 +116,10 @@ class BackgroundConfig(BaseModel): worker_heartbeat_interval_ms: int = Field(default=5_000, ge=100) worker_stale_after_ms: int = Field(default=15_000, ge=1000) kill_grace_period_ms: int = Field(default=2_000, ge=100) + max_output_bytes: int = Field(default=50 * 1024 * 1024, ge=0) + """Maximum size of a bash task's output.log in bytes. When a task's output + grows past this, the worker terminates it and marks it failed, preventing a + chatty task from exhausting disk. ``0`` means unlimited. Default: 50 MiB.""" keep_alive_on_exit: bool = Field( default=False, description="Keep background tasks alive when CLI exits. Default: kill on exit.", diff --git a/src/pythinker_code/soul/toolset.py b/src/pythinker_code/soul/toolset.py index 4b5251f2..890e55fc 100644 --- a/src/pythinker_code/soul/toolset.py +++ b/src/pythinker_code/soul/toolset.py @@ -536,7 +536,7 @@ async def _check_oauth_tokens(server_url: str) -> bool: if file_token_storage is not None: storage: Any = file_token_storage(server_url=server_url) else: - provider = fastmcp_oauth.OAuth(mcp_url=server_url) + provider: Any = fastmcp_oauth.OAuth(mcp_url=server_url) storage = provider.token_storage_adapter tokens = await storage.get_tokens() return tokens is not None diff --git a/src/pythinker_code/tools/file/grep_local.py b/src/pythinker_code/tools/file/grep_local.py index bbc3bdfe..ea0571d4 100644 --- a/src/pythinker_code/tools/file/grep_local.py +++ b/src/pythinker_code/tools/file/grep_local.py @@ -12,6 +12,7 @@ import stat import tarfile import tempfile +import time import zipfile from pathlib import Path from typing import override @@ -528,7 +529,17 @@ def _python_grep(params: Params, unavailable_reason: str) -> ToolReturnValue: matched_lines: list[str] = [] filtered_paths: list[str] = [] + # Bound the total wall-clock spent here, mirroring the ripgrep path's + # RG_TIMEOUT. This caps runaway many-file scans; it cannot interrupt a + # single catastrophic-backtracking regex mid-file (stdlib `re` has no + # timeout and the zero-dependency policy rules out a regex engine that does). + deadline = time.monotonic() + RG_TIMEOUT + timed_out = False + for file_path in _iter_python_search_files(params): + if time.monotonic() > deadline: + timed_out = True + break rel_path = _relative_output_path(file_path, search_base) if is_sensitive_file(rel_path): filtered_paths.append(rel_path) @@ -576,6 +587,8 @@ def _python_grep(params: Params, unavailable_reason: str) -> ToolReturnValue: matched_lines, pagination_message = _apply_python_pagination(matched_lines, params) messages = [f"ripgrep unavailable ({unavailable_reason}); used Python fallback."] + if timed_out: + messages.append(f"Search exceeded {RG_TIMEOUT}s; returning partial results.") if filtered_paths: messages.append(sensitive_file_warning(filtered_paths)) if pagination_message: diff --git a/src/pythinker_code/tools/web/fetch.py b/src/pythinker_code/tools/web/fetch.py index bdbe103e..fb06b748 100644 --- a/src/pythinker_code/tools/web/fetch.py +++ b/src/pythinker_code/tools/web/fetch.py @@ -2,7 +2,7 @@ import socket from pathlib import Path from typing import override -from urllib.parse import urlparse +from urllib.parse import urljoin, urlparse import aiohttp import trafilatura @@ -19,6 +19,8 @@ from pythinker_code.utils.logging import logger MAX_FETCH_BYTES = 5 * 1024 * 1024 +MAX_FETCH_REDIRECTS = 10 # matches aiohttp's default redirect cap +_REDIRECT_STATUSES = frozenset({301, 302, 303, 307, 308}) def _validate_fetch_url(url: str) -> str | None: @@ -65,6 +67,42 @@ async def _read_limited(response: aiohttp.ClientResponse, max_bytes: int) -> byt return b"".join(chunks) +class _FetchBlocked(Exception): + """Raised when a URL or one of its redirect targets fails SSRF validation.""" + + def __init__(self, reason: str) -> None: + super().__init__(reason) + self.reason = reason + + +async def _get_revalidating_redirects( + session: aiohttp.ClientSession, url: str, headers: dict[str, str] +) -> aiohttp.ClientResponse: + """GET ``url``, following redirects manually and re-validating every hop. + + aiohttp follows redirects internally without re-checking the destination, so + a public URL that 30x-redirects to a private/link-local address (e.g. a cloud + metadata endpoint at 169.254.169.254) would otherwise sail past + ``_validate_fetch_url``. We disable automatic redirects and validate each + ``Location`` before following it. + + Returns the final, open, non-redirect response (the caller owns closing it). + Raises ``_FetchBlocked`` if any hop is blocked or the redirect limit is hit. + """ + current = url + for _ in range(MAX_FETCH_REDIRECTS + 1): + if reason := _validate_fetch_url(current): + raise _FetchBlocked(reason) + response = await session.get(current, headers=headers, allow_redirects=False) + location = response.headers.get(aiohttp.hdrs.LOCATION) + if response.status in _REDIRECT_STATUSES and location: + await response.release() + current = urljoin(str(response.url), location) + continue + return response + raise _FetchBlocked("too many redirects") + + class Params(BaseModel): url: str = Field(description="The URL to fetch content from.") @@ -101,52 +139,53 @@ async def __call__(self, params: Params) -> ToolReturnValue: @staticmethod async def fetch_with_http_get(params: Params) -> ToolReturnValue: builder = ToolResultBuilder(max_line_length=None) - if reason := _validate_fetch_url(params.url): - return builder.error(f"Failed to fetch URL: {reason}", brief="URL blocked") + headers = { + "User-Agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36" + ), + } try: # Fetching arbitrary web pages can take a while on large/slow sites. fetch_timeout = aiohttp.ClientTimeout(total=180, sock_read=60, sock_connect=15) - async with ( - new_client_session(timeout=fetch_timeout) as session, - session.get( - params.url, - headers={ - "User-Agent": ( - "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " - "(KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36" - ), - }, - ) as response, - ): - if response.status >= 400: - logger.warning( - "FetchURL HTTP error: status={status}, url={url}", - status=response.status, - url=params.url, - ) - return builder.error( - ( - f"Failed to fetch URL. Status: {response.status}. " - f"This may indicate the page is not accessible or the server is down." - ), - brief=f"HTTP {response.status} error", - ) - + async with new_client_session(timeout=fetch_timeout) as session: try: - resp_text = (await _read_limited(response, MAX_FETCH_BYTES)).decode( - "utf-8", errors="replace" - ) - except ValueError: - max_mb = MAX_FETCH_BYTES // 1024 // 1024 + response = await _get_revalidating_redirects(session, params.url, headers) + except _FetchBlocked as blocked: return builder.error( - f"Failed to fetch URL: response exceeds {max_mb}MB.", - brief="Response too large", + f"Failed to fetch URL: {blocked.reason}", brief="URL blocked" ) + async with response: + if response.status >= 400: + logger.warning( + "FetchURL HTTP error: status={status}, url={url}", + status=response.status, + url=params.url, + ) + return builder.error( + ( + f"Failed to fetch URL. Status: {response.status}. " + "This may indicate the page is not accessible or " + "the server is down." + ), + brief=f"HTTP {response.status} error", + ) + + try: + resp_text = (await _read_limited(response, MAX_FETCH_BYTES)).decode( + "utf-8", errors="replace" + ) + except ValueError: + max_mb = MAX_FETCH_BYTES // 1024 // 1024 + return builder.error( + f"Failed to fetch URL: response exceeds {max_mb}MB.", + brief="Response too large", + ) - content_type = response.headers.get(aiohttp.hdrs.CONTENT_TYPE, "").lower() - if content_type.startswith(("text/plain", "text/markdown")): - builder.write(resp_text) - return builder.ok("The returned content is the full content of the page.") + content_type = response.headers.get(aiohttp.hdrs.CONTENT_TYPE, "").lower() + if content_type.startswith(("text/plain", "text/markdown")): + builder.write(resp_text) + return builder.ok("The returned content is the full content of the page.") except TimeoutError: logger.warning("FetchURL timed out: url={url}", url=params.url) return builder.error( diff --git a/src/pythinker_code/ui/shell/slash.py b/src/pythinker_code/ui/shell/slash.py index c15c0429..1f303879 100644 --- a/src/pythinker_code/ui/shell/slash.py +++ b/src/pythinker_code/ui/shell/slash.py @@ -580,15 +580,14 @@ def _feedback_destination(soul: PythinkerSoul) -> tuple[str, dict[str, str]] | N @shell_mode_registry.command def feedback(app: Shell, args: str): """Open a GitHub issue to submit feedback or report a bug""" - import webbrowser - from pythinker_code.ui.theme import get_tui_tokens as _get_tok_fb + from pythinker_code.utils.term import open_url_in_browser _t_fb = _get_tok_fb() ISSUE_URL = "https://github.com/TechMatrix-labs/pythinker-code/issues/new/choose" - if webbrowser.open(ISSUE_URL): + if open_url_in_browser(ISSUE_URL): console.print(f"[{_t_fb.success}]Opening GitHub issues in your browser...[/]") else: console.print(f"Please open: [underline]{ISSUE_URL}[/underline]") @@ -599,7 +598,6 @@ def feedback(app: Shell, args: str): async def report_error(app: Shell, args: str): """Submit a report about an error you hit, with a snapshot of recent failures.""" import platform - import webbrowser import aiohttp @@ -608,13 +606,14 @@ async def report_error(app: Shell, args: str): from pythinker_code.ui.shell.oauth import current_model_key from pythinker_code.ui.theme import get_tui_tokens as _get_tok_re from pythinker_code.utils.aiohttp import new_client_session + from pythinker_code.utils.term import open_url_in_browser _t_re = _get_tok_re() ISSUE_URL = "https://github.com/TechMatrix-labs/pythinker-code/issues" def _fallback_to_issues(): - if not webbrowser.open(ISSUE_URL): + if not open_url_in_browser(ISSUE_URL): console.print(f"Please file the report at [underline]{ISSUE_URL}[/underline].") soul = ensure_pythinker_soul(app) diff --git a/src/pythinker_code/ui/shell/visualize/_live_view.py b/src/pythinker_code/ui/shell/visualize/_live_view.py index e0a1b95f..b723185c 100644 --- a/src/pythinker_code/ui/shell/visualize/_live_view.py +++ b/src/pythinker_code/ui/shell/visualize/_live_view.py @@ -111,7 +111,6 @@ # running long enough that a quick turn won't flash it. _WORKING_TIP_MIN_ELAPSED_S = 4.0 _MAX_PINNED_TODO_ROWS = 5 -_ACTIVE_TODO_ACCENT = Style(color="#C9795A") def _todo_activity_label(label: str) -> str: @@ -576,8 +575,9 @@ def _todo_activity_line(self, label: str, *, elapsed_s: float, width: int) -> Te suffix = f" {metadata}" label_width = max(1, width - cell_width(prefix) - cell_width(suffix)) - line = Text(prefix, style=_ACTIVE_TODO_ACCENT) - line.append(truncate_to_width(label, label_width), style=_ACTIVE_TODO_ACCENT) + accent = tui_rich_style("warning") + line = Text(prefix, style=accent) + line.append(truncate_to_width(label, label_width), style=accent) line.append(suffix, style=tui_rich_style("muted")) return line @@ -672,7 +672,7 @@ def _pinned_todo_row( title = truncate_to_width(todo.title.strip(), title_budget) row = Text(prefix, style=tui_rich_style("muted")) if todo.status == "in_progress": - row.append(icon, style=_ACTIVE_TODO_ACCENT) + row.append(icon, style=tui_rich_style("warning")) row.append(" ") row.append(title, style=title_style) return row diff --git a/src/pythinker_code/utils/term.py b/src/pythinker_code/utils/term.py index c22d12e5..7baacb06 100644 --- a/src/pythinker_code/utils/term.py +++ b/src/pythinker_code/utils/term.py @@ -3,10 +3,39 @@ import contextlib import os import re +import subprocess import sys import time +def open_url_in_browser(url: str) -> bool: + """Open a URL in the default browser, detached from the terminal. + + Unlike webbrowser.open(), the launched process does not inherit + stdin/stdout/stderr, so browser startup chatter cannot corrupt the + terminal and the subprocess cannot consume key presses meant for the TUI. + """ + if sys.platform == "win32": + import webbrowser + + return webbrowser.open(url) + + cmd = ["open", url] if sys.platform == "darwin" else ["xdg-open", url] + try: + subprocess.Popen( + cmd, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + return True + except OSError: + import webbrowser + + return webbrowser.open(url) + + def ensure_new_line() -> None: """Ensure the next prompt starts at column 0 regardless of prior command output.""" diff --git a/tests/background/test_manager.py b/tests/background/test_manager.py index 9cf5d858..5817119f 100644 --- a/tests/background/test_manager.py +++ b/tests/background/test_manager.py @@ -2,6 +2,7 @@ import asyncio import contextlib +import signal import time import pytest @@ -13,7 +14,8 @@ ApprovalRuntimeEvent, ApprovalSource, ) -from pythinker_code.background import TaskRuntime, TaskSpec +from pythinker_code.background import TaskRuntime, TaskSpec, TaskStatus +from pythinker_code.background import manager as manager_module from pythinker_code.background.agent_runner import BackgroundAgentRunner from pythinker_code.notifications import NotificationDelivery, NotificationEvent, NotificationView from pythinker_code.soul.agent import Agent as SoulAgent @@ -1166,3 +1168,114 @@ async def test_manager_surfaces_timeout_failure(runtime): assert waited.runtime.interrupted is True assert waited.runtime.timed_out is True assert waited.runtime.failure_reason == "Command timed out after 1s" + + +def _make_kill_task(runtime, task_id: str, status: TaskStatus): + store = runtime.background_tasks.store + spec = TaskSpec( + id=task_id, + kind="bash", + session_id=runtime.session.id, + description="escalation target", + tool_call_id=f"tool-{task_id}", + command="true", + shell_name="bash", + shell_path="/bin/bash", + cwd=str(runtime.session.work_dir), + timeout_s=60, + ) + store.create_task(spec) + store.write_runtime(spec.id, TaskRuntime(status=status, updated_at=time.time())) + return spec.id + + +def test_escalate_sigkill_skips_when_task_terminal(runtime, monkeypatch): + """The delayed SIGKILL escalation must not fire once the task is terminal: + SIGTERM already worked, and the pid/pgid may have been recycled by the OS.""" + manager = runtime.background_tasks + task_id = _make_kill_task(runtime, "bkill001", "completed") + + calls: list = [] + monkeypatch.setattr(manager_module.os, "killpg", lambda *a: calls.append(a)) + monkeypatch.setattr(manager_module.os, "kill", lambda *a: calls.append(a)) + + manager._escalate_sigkill(task_id, pgid=999999) + manager._escalate_sigkill(task_id, pid=999999) + + assert calls == [] + + +def test_escalate_sigkill_fires_when_task_still_running(runtime, monkeypatch): + """If the task is still running after the grace delay, escalate to SIGKILL.""" + manager = runtime.background_tasks + task_id = _make_kill_task(runtime, "bkill002", "running") + + calls: list = [] + monkeypatch.setattr(manager_module.os, "killpg", lambda pgid, sig: calls.append((pgid, sig))) + + manager._escalate_sigkill(task_id, pgid=999999) + + assert calls == [(999999, signal.SIGKILL)] + + +def test_mark_task_completed_is_lock_protected(runtime, monkeypatch): + """The terminal-transition read-modify-write must run inside the + cross-process runtime lock and write via the unlocked writer. + + Without the lock, a concurrent recover()/heartbeat could read a stale + runtime between this method's read and write, clobbering the terminal + status. We assert the operation order lock_enter -> read -> write_unlocked + -> lock_exit, and that the *locking* wrapper write_runtime is never used + inside the lock (which would self-deadlock on the same-thread fcntl lock). + """ + manager = runtime.background_tasks + store = manager.store + task_id = _make_kill_task(runtime, "block001", "running") + + events: list[str] = [] + + real_lock = store._runtime_lock + real_read = store.read_runtime + real_write_unlocked = store._write_runtime_unlocked + + # _write_runtime_unlocked re-reads the runtime internally for its + # terminal-clobber guard. Suppress recording reads while inside the writer + # so the asserted sequence reflects only the manager's own read-modify-write. + in_writer = False + + @contextlib.contextmanager + def spy_lock(tid): + events.append("lock_enter") + with real_lock(tid): + try: + yield + finally: + events.append("lock_exit") + + def spy_read(tid): + if not in_writer: + events.append("read") + return real_read(tid) + + def spy_write_unlocked(tid, rt): + nonlocal in_writer + events.append("write_unlocked") + in_writer = True + try: + return real_write_unlocked(tid, rt) + finally: + in_writer = False + + def fail_write_runtime(*_a, **_k): + raise AssertionError("_mark_task_* must not use the locking write_runtime wrapper") + + monkeypatch.setattr(store, "_runtime_lock", spy_lock) + monkeypatch.setattr(store, "read_runtime", spy_read) + monkeypatch.setattr(store, "_write_runtime_unlocked", spy_write_unlocked) + monkeypatch.setattr(store, "write_runtime", fail_write_runtime) + + manager._mark_task_completed(task_id) + + assert events == ["lock_enter", "read", "write_unlocked", "lock_exit"] + # The mutation still landed: state reflects the terminal status. + assert real_read(task_id).status == "completed" diff --git a/tests/background/test_worker.py b/tests/background/test_worker.py index 0229c2b1..2172fb16 100644 --- a/tests/background/test_worker.py +++ b/tests/background/test_worker.py @@ -112,6 +112,49 @@ async def test_worker_marks_timeout_as_failed(runtime): assert view.runtime.failure_reason == "Command timed out after 1s" +@pytest.mark.asyncio +async def test_worker_bounds_output_log_growth(runtime): + store = BackgroundTaskStore(runtime.session.context_file.parent / "tasks") + spec = TaskSpec( + id="b6664444", + kind="bash", + session_id=runtime.session.id, + description="flood task", + tool_call_id="tool-7", + command="for i in $(seq 1 1000000); do echo aaaaaaaaaaaaaaaaaaaaaaaaaaaaaa; done", + shell_name="bash", + shell_path="/bin/bash", + cwd=str(runtime.session.work_dir), + timeout_s=60, + ) + store.create_task(spec) + + cap = 2000 + await run_background_task_worker( + store.task_dir(spec.id), + heartbeat_interval_ms=50, + control_poll_interval_ms=20, + kill_grace_period_ms=50, + max_output_bytes=cap, + ) + + view = store.merged_view(spec.id) + assert view.runtime.status == "failed" + assert view.runtime.interrupted is True + assert view.runtime.failure_reason is not None + assert "max_output_bytes" in view.runtime.failure_reason + + output_path = store.output_path(spec.id) + text = output_path.read_text(encoding="utf-8") + assert "output limit exceeded" in text + # The cap is enforced by a poll loop, not per-write, so the file overshoots + # the cap by whatever the child flushes between polls (and that window + # stretches under CPU contention). The guarantee is that growth is + # *bounded* far below the ~30 MB this command writes uncapped; a few MB of + # overshoot still proves the limiter fired and terminated the task. + assert output_path.stat().st_size <= 5 * 1024 * 1024 + + def test_terminate_process_tree_windows_uses_taskkill_tree(monkeypatch): calls: list[list[str]] = [] diff --git a/tests/tools/test_fetch_url.py b/tests/tools/test_fetch_url.py index 16674c8a..33b4f2a4 100644 --- a/tests/tools/test_fetch_url.py +++ b/tests/tools/test_fetch_url.py @@ -4,7 +4,7 @@ from __future__ import annotations -from collections.abc import AsyncIterator +from collections.abc import AsyncIterator, Awaitable, Callable from typing import Protocol import pytest @@ -307,3 +307,93 @@ async def service_handler(request: web.Request) -> web.Response: finally: await runner.cleanup() + + +@pytest_asyncio.fixture +async def serve_app() -> AsyncIterator[Callable[[web.Application], Awaitable[str]]]: + """Serve an arbitrary aiohttp app on a random loopback port and clean up.""" + runners: list[web.AppRunner] = [] + + async def _serve(app: web.Application) -> str: + runner = web.AppRunner(app) + await runner.setup() + site = web.TCPSite(runner, host="127.0.0.1", port=0) + await site.start() + runners.append(runner) + sockets = site._server.sockets # type: ignore[attr-defined] + assert sockets, "Server failed to bind to a port." + return f"http://127.0.0.1:{sockets[0].getsockname()[1]}" + + try: + yield _serve + finally: + for runner in runners: + await runner.cleanup() + + +async def test_fetch_url_redirect_to_blocked_target_is_rejected( + fetch_url_tool: FetchURL, + serve_app: Callable[[web.Application], Awaitable[str]], + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A 30x redirect whose target fails SSRF validation must be rejected, not + silently followed (aiohttp would otherwise follow it without re-checking).""" + + def _validator(url: str) -> str | None: + return "internal address blocked" if "blocked.invalid" in url else None + + monkeypatch.setattr(fetch_module, "_validate_fetch_url", _validator) + + async def handler(request: web.Request) -> web.Response: # noqa: ARG001 + return web.Response(status=302, headers={"Location": "http://blocked.invalid/secret"}) + + app = web.Application() + app.router.add_get("/", handler) + base = await serve_app(app) + + result = await fetch_url_tool(Params(url=base)) + + assert result.is_error + assert "internal address blocked" in result.message + + +async def test_fetch_url_follows_safe_redirect( + fetch_url_tool: FetchURL, + serve_app: Callable[[web.Application], Awaitable[str]], +) -> None: + """Legitimate redirects to allowed targets are still followed end to end.""" + + async def start(request: web.Request) -> web.Response: # noqa: ARG001 + return web.Response(status=302, headers={"Location": "/dest"}) + + async def dest(request: web.Request) -> web.Response: # noqa: ARG001 + return web.Response(text="redirected body", content_type="text/markdown") + + app = web.Application() + app.router.add_get("/start", start) + app.router.add_get("/dest", dest) + base = await serve_app(app) + + result = await fetch_url_tool(Params(url=f"{base}/start")) + + assert not result.is_error + assert "redirected body" in result.output + + +async def test_fetch_url_redirect_loop_is_capped( + fetch_url_tool: FetchURL, + serve_app: Callable[[web.Application], Awaitable[str]], +) -> None: + """A redirect loop terminates with an error instead of looping forever.""" + + async def loop(request: web.Request) -> web.Response: # noqa: ARG001 + return web.Response(status=302, headers={"Location": "/loop"}) + + app = web.Application() + app.router.add_get("/loop", loop) + base = await serve_app(app) + + result = await fetch_url_tool(Params(url=f"{base}/loop")) + + assert result.is_error + assert "too many redirects" in result.message diff --git a/tests/tools/test_grep.py b/tests/tools/test_grep.py index 3d1dd4cb..599fdcfe 100644 --- a/tests/tools/test_grep.py +++ b/tests/tools/test_grep.py @@ -9,11 +9,13 @@ import pytest_asyncio from inline_snapshot import snapshot +import pythinker_code.tools.file.grep_local as grep_module from pythinker_code.tools.file.grep_local import ( Grep, Params, _build_rg_args, _find_existing_rg, + _python_grep, _rg_binary_name, _strip_path_prefix, ) @@ -1037,3 +1039,26 @@ async def test_grep_allows_env_example(grep_tool: Grep): ) assert not result.is_error assert ".env.example" in result.output + + +def test_python_fallback_bounds_wall_clock(monkeypatch, tmp_path): + """The Python fallback caps total wall-clock like the ripgrep path, and + surfaces a partial-results notice when it does.""" + (tmp_path / "a.txt").write_text("needle\n") + (tmp_path / "b.txt").write_text("needle\n") + + # First monotonic() reading establishes the deadline; every later reading is + # far in the future, so the deadline is already blown on the first file. + readings = iter([0.0]) + + def _fake_monotonic() -> float: + return next(readings, 1e9) + + monkeypatch.setattr(grep_module.time, "monotonic", _fake_monotonic) + + result = _python_grep( + Params(pattern="needle", path=str(tmp_path), output_mode="files_with_matches"), + "forced fallback", + ) + + assert f"Search exceeded {grep_module.RG_TIMEOUT}s" in result.message diff --git a/tests/ui_and_conv/test_live_view_todos.py b/tests/ui_and_conv/test_live_view_todos.py index a394e467..230ff842 100644 --- a/tests/ui_and_conv/test_live_view_todos.py +++ b/tests/ui_and_conv/test_live_view_todos.py @@ -165,12 +165,14 @@ def test_finished_todos_move_to_bottom_of_menu(monkeypatch) -> None: assert rendered.index("✓ Finished first") < rendered.index("✓ Finished second") -def test_active_todo_activity_line_uses_coral_accent() -> None: +def test_active_todo_activity_line_uses_warning_accent() -> None: view = _LiveView(StatusUpdate(context_tokens=10_000)) line = view._todo_activity_line("Implement pinned todos", elapsed_s=0.88, width=100) - assert _span_colors_for(line, "Implement pinned todos") == {"#c9795a"} + assert _span_colors_for(line, "Implement pinned todos") == { + _color_hex(tui_rich_style("warning").color) + } def test_active_pinned_todo_row_uses_accent_icon_and_white_title() -> None: @@ -184,7 +186,7 @@ def test_active_pinned_todo_row_uses_accent_icon_and_white_title() -> None: ) title_style = _style_for(row, "Implement pinned todos") - assert _span_colors_for(row, "■") == {"#c9795a"} + assert _span_colors_for(row, "■") == {_color_hex(tui_rich_style("warning").color)} assert title_style.color == tui_rich_style("activity_label").color assert title_style.bold is True diff --git a/tests/ui_and_conv/test_shell_feedback_slash.py b/tests/ui_and_conv/test_shell_feedback_slash.py index 1a6a69e4..77f82f5a 100644 --- a/tests/ui_and_conv/test_shell_feedback_slash.py +++ b/tests/ui_and_conv/test_shell_feedback_slash.py @@ -24,7 +24,7 @@ def test_registered_in_shell_mode_registry(self) -> None: class TestFeedbackOpensIssue: def test_opens_new_issue_url(self, monkeypatch) -> None: open_mock = Mock(return_value=True) - monkeypatch.setattr("webbrowser.open", open_mock) + monkeypatch.setattr("pythinker_code.utils.term.open_url_in_browser", open_mock) monkeypatch.setattr(shell_slash.console, "print", Mock()) shell = Mock() @@ -37,7 +37,7 @@ def test_opens_new_issue_url(self, monkeypatch) -> None: assert "new" in url def test_prints_success_when_browser_opens(self, monkeypatch) -> None: - monkeypatch.setattr("webbrowser.open", Mock(return_value=True)) + monkeypatch.setattr("pythinker_code.utils.term.open_url_in_browser", Mock(return_value=True)) print_mock = Mock() monkeypatch.setattr(shell_slash.console, "print", print_mock) @@ -47,7 +47,7 @@ def test_prints_success_when_browser_opens(self, monkeypatch) -> None: assert "Opening" in output or "browser" in output.lower() def test_prints_url_when_browser_fails(self, monkeypatch) -> None: - monkeypatch.setattr("webbrowser.open", Mock(return_value=False)) + monkeypatch.setattr("pythinker_code.utils.term.open_url_in_browser", Mock(return_value=False)) print_mock = Mock() monkeypatch.setattr(shell_slash.console, "print", print_mock) From 9144a6dfd32e36c1ccc23b17d5bb7b81774cbd9f Mon Sep 17 00:00:00 2001 From: mohamed-elkholy95 Date: Fri, 29 May 2026 09:21:52 -0400 Subject: [PATCH 03/12] fix(background): crash-consistent agent-task status (H2) + prune aged terminal tasks (M9) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit H2: route every terminal agent-task update through one BackgroundTaskManager.finalize_agent_task() that writes the authoritative TaskRuntime first and the derived subagent record last, replacing the eight ad-hoc (update_instance, _mark_task_*) pairs in BackgroundAgentRunner whose ordering was inconsistent (run() wrote the record first, _run_core the task). recover() now reconciles a subagent record still stuck at running_background to the status implied by the authoritative TaskRuntime — for terminal tasks too — closing the crash/kill window that left TaskRuntime and AgentInstanceRecord divergent. A resumed running_foreground or already-terminal record is never clobbered. M9: prune terminal background-task directories older than config.background.task_retention_days (default 7) opportunistically during reconcile(); never removes non-terminal tasks or tasks whose worker is alive. Tests: crash/kill reconciliation, foreground no-clobber, finalize end-state, reconcile pruning, and cleanup_old_tasks unit coverage. --- src/pythinker_code/background/agent_runner.py | 44 +++-- src/pythinker_code/background/manager.py | 154 ++++++++++++++--- src/pythinker_code/background/store.py | 51 ++++++ src/pythinker_code/config.py | 3 + tests/background/test_manager.py | 158 ++++++++++++++++++ tests/background/test_store.py | 77 +++++++++ 6 files changed, 443 insertions(+), 44 deletions(-) diff --git a/src/pythinker_code/background/agent_runner.py b/src/pythinker_code/background/agent_runner.py index 601ef056..470b689b 100644 --- a/src/pythinker_code/background/agent_runner.py +++ b/src/pythinker_code/background/agent_runner.py @@ -99,9 +99,11 @@ async def run(self) -> None: id=self._task_id, t=self._timeout_s, ) - self._runtime.subagent_store.update_instance(self._agent_id, status="failed") - self._manager._mark_task_timed_out( - self._task_id, f"Agent task timed out after {self._timeout_s}s" + self._manager.finalize_agent_task( + self._task_id, + self._agent_id, + outcome="timed_out", + reason=f"Agent task timed out after {self._timeout_s}s", ) output.error( _timeout_recovery_message(timeout_s=self._timeout_s, agent_id=self._agent_id) @@ -109,25 +111,29 @@ async def run(self) -> None: else: # Internal timeout (e.g. aiohttp request) — treat as generic failure logger.exception("Background agent runner failed") - self._runtime.subagent_store.update_instance(self._agent_id, status="failed") - self._manager._mark_task_failed(self._task_id, str(exc)) + self._manager.finalize_agent_task( + self._task_id, self._agent_id, outcome="failed", reason=str(exc) + ) output.error(str(exc)) except asyncio.CancelledError: - self._runtime.subagent_store.update_instance(self._agent_id, status="killed") - self._manager._mark_task_killed(self._task_id, "Stopped by TaskStop") + self._manager.finalize_agent_task( + self._task_id, self._agent_id, outcome="killed", reason="Stopped by TaskStop" + ) output.stage("cancelled") raise except RunCancelled: # RunCancelled is Exception (not BaseException), so re-raising it from # an asyncio.create_task would trigger "Task exception was never retrieved". # Just mark killed and return — cleanup is already done. - self._runtime.subagent_store.update_instance(self._agent_id, status="killed") - self._manager._mark_task_killed(self._task_id, "Run was cancelled") + self._manager.finalize_agent_task( + self._task_id, self._agent_id, outcome="killed", reason="Run was cancelled" + ) output.stage("cancelled") except Exception as exc: logger.exception("Background agent runner failed") - self._runtime.subagent_store.update_instance(self._agent_id, status="failed") - self._manager._mark_task_failed(self._task_id, str(exc)) + self._manager.finalize_agent_task( + self._task_id, self._agent_id, outcome="failed", reason=str(exc) + ) output.error(str(exc)) finally: # Whatever happens in approval cleanup below, the dict pop must @@ -197,22 +203,24 @@ async def _ui_loop_fn(wire: Wire) -> None: ), ) if failure is not None: - self._manager._mark_task_failed(self._task_id, failure.message) - self._runtime.subagent_store.update_instance(self._agent_id, status="failed") + self._manager.finalize_agent_task( + self._task_id, self._agent_id, outcome="failed", reason=failure.message + ) output.stage(f"failed: {failure.brief}") return output.stage("run_soul_finished") if final_response is None: - self._manager._mark_task_failed( - self._task_id, "Agent completed but produced no output." + self._manager.finalize_agent_task( + self._task_id, + self._agent_id, + outcome="failed", + reason="Agent completed but produced no output.", ) - self._runtime.subagent_store.update_instance(self._agent_id, status="failed") output.stage("failed: empty output") return output.summary(final_response) - self._runtime.subagent_store.update_instance(self._agent_id, status="idle") - self._manager._mark_task_completed(self._task_id) + self._manager.finalize_agent_task(self._task_id, self._agent_id, outcome="completed") def _on_approval_runtime_event(self, event: ApprovalRuntimeEvent) -> None: request = event.request diff --git a/src/pythinker_code/background/manager.py b/src/pythinker_code/background/manager.py index be561464..a3764c65 100644 --- a/src/pythinker_code/background/manager.py +++ b/src/pythinker_code/background/manager.py @@ -8,7 +8,7 @@ import threading import time from pathlib import Path -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, Literal from pythinker_host.local import local_host @@ -19,6 +19,7 @@ if TYPE_CHECKING: from pythinker_code.soul.agent import Runtime + from pythinker_code.subagents.models import SubagentStatus from .ids import generate_task_id from .models import ( @@ -32,6 +33,25 @@ ) from .store import BackgroundTaskStore +AgentTaskOutcome = Literal["completed", "failed", "timed_out", "killed"] + + +def _subagent_status_for_task_status(status: TaskStatus) -> SubagentStatus | None: + """Map an authoritative agent ``TaskStatus`` to the subagent instance status + it implies, for crash-recovery reconciliation. + + Returns ``None`` for statuses that imply no reconciliation. ``recoverable`` + and ``completed`` park the instance at ``idle`` (resumable / done-clean); + ``killed`` and the failure family (``failed``/``lost``) map straight through. + """ + if status in ("completed", "recoverable"): + return "idle" + if status == "killed": + return "killed" + if status in ("failed", "lost"): + return "failed" + return None + class BackgroundTaskManager: def __init__( @@ -539,31 +559,15 @@ def recover(self) -> None: now = time.time() stale_after = self._config.worker_stale_after_ms / 1000 for view in self._store.list_views(): - if is_terminal_status(view.runtime.status): - continue if view.spec.kind == "agent": - if view.spec.id in self._live_agent_tasks: - continue - runtime = view.runtime.model_copy() - runtime.finished_at = now - runtime.updated_at = now - agent_id = (view.spec.kind_payload or {}).get("agent_id") - runtime.status = "recoverable" if isinstance(agent_id, str) else "lost" - runtime.failure_reason = ( - "In-process background agent is no longer running; resume the stored agent " - f"instance {agent_id} to continue." - if isinstance(agent_id, str) - else "In-process background agent is no longer running" - ) - self._store.write_runtime(view.spec.id, runtime) - if ( - isinstance(agent_id, str) - and self._runtime is not None - and self._runtime.subagent_store is not None - ): - record = self._runtime.subagent_store.get_instance(agent_id) - if record is not None and record.status == "running_background": - self._runtime.subagent_store.update_instance(agent_id, status="idle") + # Agent tasks are handled before the terminal-skip below so a + # terminal task whose subagent record is still stuck at + # running_background (crash after the authoritative TaskRuntime + # write, or after kill() but before the runner's cancel handler) + # still gets reconciled. + self._recover_agent_view(view, now=now) + continue + if is_terminal_status(view.runtime.status): continue last_progress_at = ( view.runtime.heartbeat_at @@ -606,9 +610,68 @@ def recover(self) -> None: ) self._store._write_runtime_unlocked(view.spec.id, runtime) # pyright: ignore[reportPrivateUsage] + def _recover_agent_view(self, view: TaskView, *, now: float) -> None: + """Recover an in-process agent task and reconcile its subagent record. + + An orphaned task (non-terminal, no live asyncio task) gets an + authoritative terminal status re-derived here. Either way the subagent + instance record is reconciled to agree with the authoritative + ``TaskRuntime`` — closing the window where a process crash between + :meth:`finalize_agent_task`'s two writes, or between :meth:`kill` and the + runner's cancellation handler, left the two records divergent. + """ + agent_id_raw = (view.spec.kind_payload or {}).get("agent_id") + agent_id = agent_id_raw if isinstance(agent_id_raw, str) else None + runtime_status: TaskStatus = view.runtime.status + if not is_terminal_status(runtime_status): + if view.spec.id in self._live_agent_tasks: + return + runtime = view.runtime.model_copy() + runtime.finished_at = now + runtime.updated_at = now + runtime.status = "recoverable" if agent_id is not None else "lost" + runtime.failure_reason = ( + "In-process background agent is no longer running; resume the stored agent " + f"instance {agent_id} to continue." + if agent_id is not None + else "In-process background agent is no longer running" + ) + self._store.write_runtime(view.spec.id, runtime) + runtime_status = runtime.status + self._reconcile_subagent_status(agent_id, runtime_status) + + def _reconcile_subagent_status(self, agent_id: str | None, runtime_status: TaskStatus) -> None: + """Bring a subagent instance record into agreement with the authoritative + agent ``TaskStatus``. + + Only a record still parked at ``running_background`` is touched, so a + record that has since been resumed in the foreground (``running_foreground``) + or already settled to a terminal status is never clobbered. This is safe + precisely because a background agent's record stays ``running_background`` + for the whole run, so an interrupted finalize always leaves it there. + """ + if agent_id is None or self._runtime is None or self._runtime.subagent_store is None: + return + target = _subagent_status_for_task_status(runtime_status) + if target is None: + return + record = self._runtime.subagent_store.get_instance(agent_id) + if record is None or record.status != "running_background": + return + self._runtime.subagent_store.update_instance(agent_id, status=target) + def reconcile(self, *, limit: int | None = None) -> list[str]: self.recover() - return self.publish_terminal_notifications(limit=limit) + published = self.publish_terminal_notifications(limit=limit) + # Opportunistic housekeeping: prune aged terminal task directories so the + # store does not grow unbounded. Best-effort and bounded — cleanup skips + # non-terminal tasks and any whose worker is still alive, and no-ops when + # task_retention_days == 0. A failure here must never break reconcile. + try: + self._store.cleanup_old_tasks(self._config.task_retention_days) + except Exception: + logger.warning("Background task cleanup failed during reconcile") + return published def publish_terminal_notifications(self, *, limit: int | None = None) -> list[str]: published: list[str] = [] @@ -684,6 +747,45 @@ def publish_terminal_notifications(self, *, limit: int | None = None) -> list[st break return published + def finalize_agent_task( + self, + task_id: str, + agent_id: str | None, + *, + outcome: AgentTaskOutcome, + reason: str | None = None, + ) -> None: + """Apply an in-process agent task's terminal outcome to both stores. + + The authoritative ``TaskRuntime`` is written FIRST and the derived + subagent instance record LAST, in one fixed order, so different call + sites can never interleave the pair in conflicting orders (the bug this + replaces: ``run()`` wrote the record first, ``_run_core`` wrote the task + first). A process crash between the two writes leaves the instance record + at ``running_background``; :meth:`recover` then reconciles it to the + status implied by the authoritative ``TaskRuntime`` + (see :func:`_subagent_status_for_task_status`). Callers must route every + terminal agent update through here. + """ + if outcome == "completed": + self._mark_task_completed(task_id) + elif outcome == "timed_out": + self._mark_task_timed_out(task_id, reason or "Agent task timed out") + elif outcome == "killed": + self._mark_task_killed(task_id, reason or "Agent task stopped") + else: + self._mark_task_failed(task_id, reason or "Agent task failed") + + if ( + agent_id is not None + and self._runtime is not None + and self._runtime.subagent_store is not None + ): + subagent_status: SubagentStatus = ( + "idle" if outcome == "completed" else "killed" if outcome == "killed" else "failed" + ) + self._runtime.subagent_store.update_instance(agent_id, status=subagent_status) + def _mark_task_running(self, task_id: str) -> None: with self._store._runtime_lock(task_id): # pyright: ignore[reportPrivateUsage] runtime = self._store.read_runtime(task_id) diff --git a/src/pythinker_code/background/store.py b/src/pythinker_code/background/store.py index 83904217..cf4397e5 100644 --- a/src/pythinker_code/background/store.py +++ b/src/pythinker_code/background/store.py @@ -4,6 +4,8 @@ import json import os import re +import shutil +import time from contextlib import contextmanager from pathlib import Path @@ -292,6 +294,55 @@ def tail_output(self, task_id: str, max_bytes: int, max_lines: int) -> str: lines = lines[-max_lines:] return "\n".join(lines) + def cleanup_old_tasks(self, max_age_days: int, *, now: float | None = None) -> list[str]: + """Prune terminal task directories older than ``max_age_days``. + + Returns the list of removed task ids. ``max_age_days <= 0`` disables + cleanup and returns ``[]``. Non-terminal tasks, and terminal tasks whose + recorded worker process is still alive, are never removed. + """ + if max_age_days <= 0: + return [] + + current = time.time() if now is None else now + max_age_seconds = max_age_days * 86_400.0 + removed: list[str] = [] + + for task_id in self.list_task_ids(): + runtime = self.read_runtime(task_id) + if not is_terminal_status(runtime.status): + continue + if runtime.worker_pid is not None and _pid_alive(runtime.worker_pid): + continue + reference = runtime.finished_at or runtime.updated_at + if current - reference < max_age_seconds: + continue + try: + shutil.rmtree(self.task_path(task_id), ignore_errors=False) + except OSError as exc: + logger.warning( + "Failed to remove old background task {task_id} at {path}: {error}", + task_id=task_id, + path=self.task_path(task_id), + error=exc, + ) + continue + removed.append(task_id) + + return removed + + +def _pid_alive(pid: int) -> bool: + """Return True if a process with ``pid`` appears to be running.""" + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except OSError: + # e.g. PermissionError: the process exists but we cannot signal it. + return True + return True + def _read_json_model[T: BaseModel](path: Path, model: type[T], *, fallback: T, artifact: str) -> T: try: diff --git a/src/pythinker_code/config.py b/src/pythinker_code/config.py index d700d90e..1432853a 100644 --- a/src/pythinker_code/config.py +++ b/src/pythinker_code/config.py @@ -109,6 +109,9 @@ class BackgroundConfig(BaseModel): """Background task runtime configuration.""" max_running_tasks: int = Field(default=4, ge=1) + task_retention_days: int = Field(default=7, ge=0) + """Terminal background tasks older than this many days are pruned on + reconcile. ``0`` disables cleanup.""" read_max_bytes: int = Field(default=30_000, ge=1024) notification_tail_lines: int = Field(default=20, ge=1) notification_tail_chars: int = Field(default=3_000, ge=256) diff --git a/tests/background/test_manager.py b/tests/background/test_manager.py index 5817119f..50634814 100644 --- a/tests/background/test_manager.py +++ b/tests/background/test_manager.py @@ -693,6 +693,164 @@ def test_recover_marks_stale_agent_task_lost_and_clears_instance_running_state(r assert instance.status == "idle" +def _seed_agent_task( + runtime, + *, + task_id: str, + agent_id: str, + task_status: TaskStatus, + instance_status: str = "running_background", + updated_offset: float = 0.0, +): + """Create a paired (subagent instance, agent TaskRuntime) for recovery tests.""" + store = runtime.background_tasks.store + runtime.subagent_store.create_instance( + agent_id=agent_id, + description="background agent", + launch_spec=AgentLaunchSpec( + agent_id=agent_id, + subagent_type="coder", + model_override=None, + effective_model=None, + ), + ) + runtime.subagent_store.update_instance(agent_id, status=instance_status) + spec = TaskSpec( + id=task_id, + kind="agent", + session_id=runtime.session.id, + description="background agent task", + tool_call_id=f"tool-{task_id}", + owner_role="root", + kind_payload={ + "agent_id": agent_id, + "subagent_type": "coder", + "prompt": "do work", + "model_override": None, + "launch_mode": "background", + }, + ) + store.create_task(spec) + now = time.time() - updated_offset + store.write_runtime( + spec.id, + TaskRuntime(status=task_status, updated_at=now, finished_at=now), + ) + return spec + + +def test_recover_reconciles_subagent_record_for_terminal_agent_task(runtime): + # Simulates a crash after finalize wrote the authoritative TaskRuntime + # (failed) but before the subagent record write: the instance is left + # running_background. recover() must bring the two into agreement. + manager = runtime.background_tasks + _seed_agent_task(runtime, task_id="acrashtask", agent_id="acrashagent", task_status="failed") + assert runtime.subagent_store.require_instance("acrashagent").status == "running_background" + + manager.recover() + + assert runtime.subagent_store.require_instance("acrashagent").status == "failed" + assert manager.store.merged_view("acrashtask").runtime.status == "failed" + + +def test_recover_reconciles_subagent_record_for_killed_agent_task(runtime): + # kill() marks TaskRuntime killed first, then cancels the asyncio task. A + # crash before the runner's cancel handler writes the instance leaves it + # running_background; recover() maps killed -> killed. + manager = runtime.background_tasks + _seed_agent_task(runtime, task_id="akilltask", agent_id="akillagent", task_status="killed") + + manager.recover() + + assert runtime.subagent_store.require_instance("akillagent").status == "killed" + + +def test_recover_does_not_clobber_resumed_foreground_instance(runtime): + # An old terminal task whose agent instance has since been resumed in the + # foreground must not be reset by recovery reconciliation. + manager = runtime.background_tasks + _seed_agent_task( + runtime, + task_id="aresumetask", + agent_id="aresumeagent", + task_status="recoverable", + instance_status="running_foreground", + ) + + manager.recover() + + assert runtime.subagent_store.require_instance("aresumeagent").status == "running_foreground" + + +def test_finalize_agent_task_updates_both_records_consistently(runtime): + manager = runtime.background_tasks + spec = _seed_agent_task( + runtime, task_id="afintask", agent_id="afinagent", task_status="running" + ) + + manager.finalize_agent_task("afintask", "afinagent", outcome="failed", reason="boom") + + assert manager.store.merged_view(spec.id).runtime.status == "failed" + assert manager.store.read_runtime(spec.id).failure_reason == "boom" + assert runtime.subagent_store.require_instance("afinagent").status == "failed" + + +def test_finalize_agent_task_completed_parks_instance_idle(runtime): + manager = runtime.background_tasks + _seed_agent_task(runtime, task_id="adonetask", agent_id="adoneagent", task_status="running") + + manager.finalize_agent_task("adonetask", "adoneagent", outcome="completed") + + assert manager.store.merged_view("adonetask").runtime.status == "completed" + assert runtime.subagent_store.require_instance("adoneagent").status == "idle" + + +def test_reconcile_prunes_aged_terminal_tasks(runtime): + manager = runtime.background_tasks + store = manager.store + runtime.config.background.task_retention_days = 1 + aged = time.time() - 3 * 86_400 + old_spec = TaskSpec( + id="aoldbash1", + kind="bash", + session_id=runtime.session.id, + description="aged completed task", + tool_call_id="tool-old", + command="echo hi", + shell_name="bash", + shell_path="/bin/bash", + cwd=str(runtime.session.work_dir), + timeout_s=10, + ) + store.create_task(old_spec) + store.write_runtime( + old_spec.id, TaskRuntime(status="completed", finished_at=aged, updated_at=aged) + ) + fresh_spec = TaskSpec( + id="afreshbash1", + kind="bash", + session_id=runtime.session.id, + description="fresh completed task", + tool_call_id="tool-fresh", + command="echo hi", + shell_name="bash", + shell_path="/bin/bash", + cwd=str(runtime.session.work_dir), + timeout_s=10, + ) + store.create_task(fresh_spec) + store.write_runtime( + fresh_spec.id, + TaskRuntime(status="completed", finished_at=time.time(), updated_at=time.time()), + ) + + manager.reconcile() + + task_ids = store.list_task_ids() + assert old_spec.id not in task_ids + assert fresh_spec.id in task_ids + + def test_mark_task_running_does_not_overwrite_terminal_state(runtime): manager = runtime.background_tasks store = manager.store diff --git a/tests/background/test_store.py b/tests/background/test_store.py index db8321d7..da6e9cce 100644 --- a/tests/background/test_store.py +++ b/tests/background/test_store.py @@ -206,3 +206,80 @@ def test_list_views_uses_spec_created_at_when_runtime_is_corrupted(runtime): views = store.list_views() assert [view.spec.id for view in views] == ["b9999995", "b9999994"] + + +def _make_spec(runtime, task_id: str) -> TaskSpec: + return TaskSpec( + id=task_id, + kind="bash", + session_id=runtime.session.id, + description="cleanup candidate", + tool_call_id=f"call-{task_id}", + command="echo ok", + shell_name="bash", + shell_path="/bin/bash", + cwd=str(runtime.session.work_dir), + timeout_s=60, + ) + + +def test_cleanup_old_tasks(runtime): + import os + + from pythinker_code.background.models import TaskRuntime + + store = BackgroundTaskStore(runtime.session.context_file.parent / "tasks") + now = 1_000_000.0 + day = 86_400.0 + + cases = { + # (a) terminal + old -> removed + "boldterm01": TaskRuntime( + status="completed", + updated_at=now - 30 * day, + finished_at=now - 30 * day, + ), + # (b) terminal + recent -> kept + "bnewterm01": TaskRuntime( + status="failed", + updated_at=now - 1 * day, + finished_at=now - 1 * day, + ), + # (c) non-terminal (running) + old -> kept + "boldrun001": TaskRuntime( + status="running", + updated_at=now - 30 * day, + ), + # (d) terminal + old but worker_pid is a live pid -> kept + "boldalive1": TaskRuntime( + status="killed", + updated_at=now - 30 * day, + finished_at=now - 30 * day, + worker_pid=os.getpid(), + ), + } + for task_id, rt in cases.items(): + store.create_task(_make_spec(runtime, task_id)) + store.write_runtime(task_id, rt) + + removed = store.cleanup_old_tasks(max_age_days=7, now=now) + + assert removed == ["boldterm01"] + assert not store.task_path("boldterm01").exists() + assert store.task_path("bnewterm01").exists() + assert store.task_path("boldrun001").exists() + assert store.task_path("boldalive1").exists() + + +def test_cleanup_old_tasks_disabled_returns_empty(runtime): + from pythinker_code.background.models import TaskRuntime + + store = BackgroundTaskStore(runtime.session.context_file.parent / "tasks") + store.create_task(_make_spec(runtime, "boldterm02")) + store.write_runtime( + "boldterm02", + TaskRuntime(status="completed", updated_at=0.0, finished_at=0.0), + ) + + assert store.cleanup_old_tasks(max_age_days=0, now=1_000_000.0) == [] + assert store.task_path("boldterm02").exists() From 3c3579f9c96cbc634741c4f2f74c7062b757debd Mon Sep 17 00:00:00 2001 From: mohamed-elkholy95 Date: Fri, 29 May 2026 09:30:13 -0400 Subject: [PATCH 04/12] fix(background): don't reconcile a live agent's record off a stale terminal task MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The H2 recover() reconciliation could corrupt a currently-running agent. When an agent_id is reused — a background resume mints a new task_id while the prior run's task stays terminal in the store — recover() saw the old terminal task alongside a running_background record and reset the live agent's record to the old task's terminal status. AgentInstanceRecord.last_task_id is not maintained, so gate the reconcile on the set of agent_ids owned by live in-process tasks and skip those. Regression test: old terminal task + live resumed task sharing one agent_id; the live agent's running_background record must survive recover(). --- src/pythinker_code/background/manager.py | 40 +++++++++++---- tests/background/test_manager.py | 65 ++++++++++++++++++++++++ 2 files changed, 94 insertions(+), 11 deletions(-) diff --git a/src/pythinker_code/background/manager.py b/src/pythinker_code/background/manager.py index a3764c65..13f34011 100644 --- a/src/pythinker_code/background/manager.py +++ b/src/pythinker_code/background/manager.py @@ -558,14 +558,24 @@ def kill_all_active(self, *, reason: str = "CLI session ended") -> list[str]: def recover(self) -> None: now = time.time() stale_after = self._config.worker_stale_after_ms / 1000 - for view in self._store.list_views(): + views = self._store.list_views() + # agent_ids owned by a live in-process task. A terminal task that reused + # the same agent_id (e.g. a prior run before a background resume) must + # not reconcile its status onto the *live* agent's record. + live_agent_ids = { + agent_id + for view in views + if view.spec.id in self._live_agent_tasks + and isinstance(agent_id := (view.spec.kind_payload or {}).get("agent_id"), str) + } + for view in views: if view.spec.kind == "agent": # Agent tasks are handled before the terminal-skip below so a # terminal task whose subagent record is still stuck at # running_background (crash after the authoritative TaskRuntime # write, or after kill() but before the runner's cancel handler) # still gets reconciled. - self._recover_agent_view(view, now=now) + self._recover_agent_view(view, now=now, live_agent_ids=live_agent_ids) continue if is_terminal_status(view.runtime.status): continue @@ -610,7 +620,7 @@ def recover(self) -> None: ) self._store._write_runtime_unlocked(view.spec.id, runtime) # pyright: ignore[reportPrivateUsage] - def _recover_agent_view(self, view: TaskView, *, now: float) -> None: + def _recover_agent_view(self, view: TaskView, *, now: float, live_agent_ids: set[str]) -> None: """Recover an in-process agent task and reconcile its subagent record. An orphaned task (non-terminal, no live asyncio task) gets an @@ -638,19 +648,27 @@ def _recover_agent_view(self, view: TaskView, *, now: float) -> None: ) self._store.write_runtime(view.spec.id, runtime) runtime_status = runtime.status - self._reconcile_subagent_status(agent_id, runtime_status) + self._reconcile_subagent_status(agent_id, runtime_status, live_agent_ids) - def _reconcile_subagent_status(self, agent_id: str | None, runtime_status: TaskStatus) -> None: + def _reconcile_subagent_status( + self, agent_id: str | None, runtime_status: TaskStatus, live_agent_ids: set[str] + ) -> None: """Bring a subagent instance record into agreement with the authoritative agent ``TaskStatus``. - Only a record still parked at ``running_background`` is touched, so a - record that has since been resumed in the foreground (``running_foreground``) - or already settled to a terminal status is never clobbered. This is safe - precisely because a background agent's record stays ``running_background`` - for the whole run, so an interrupted finalize always leaves it there. + Skipped when ``agent_id`` is owned by a live in-process task — the same + agent_id may have been reused by a newer, currently-running task (e.g. a + background resume), so the ``running_background`` record belongs to that + live run, not to this (older, terminal) view. Otherwise only a record + still parked at ``running_background`` is touched, so a record resumed in + the foreground (``running_foreground``) or already settled to a terminal + status is never clobbered. This is safe because a background agent's + record stays ``running_background`` for the whole run, so an interrupted + finalize always leaves it there. """ - if agent_id is None or self._runtime is None or self._runtime.subagent_store is None: + if agent_id is None or agent_id in live_agent_ids: + return + if self._runtime is None or self._runtime.subagent_store is None: return target = _subagent_status_for_task_status(runtime_status) if target is None: diff --git a/tests/background/test_manager.py b/tests/background/test_manager.py index 50634814..4dc29433 100644 --- a/tests/background/test_manager.py +++ b/tests/background/test_manager.py @@ -851,6 +851,71 @@ def test_reconcile_prunes_aged_terminal_tasks(runtime): assert fresh_spec.id in task_ids +@pytest.mark.asyncio +async def test_recover_does_not_clobber_resumed_background_instance(runtime): + # An agent_id can be reused: an old background task is terminal while the + # SAME agent_id has been resumed as a new, live background task whose record + # is (correctly) running_background. recover() must not reconcile the old + # terminal task's view onto the live agent's record. + manager = runtime.background_tasks + store = manager.store + runtime.subagent_store.create_instance( + agent_id="aresumebg", + description="resumed background agent", + launch_spec=AgentLaunchSpec( + agent_id="aresumebg", + subagent_type="coder", + model_override=None, + effective_model=None, + ), + ) + runtime.subagent_store.update_instance("aresumebg", status="running_background") + + def _agent_spec(task_id: str) -> TaskSpec: + return TaskSpec( + id=task_id, + kind="agent", + session_id=runtime.session.id, + description="agent run", + tool_call_id=f"tool-{task_id}", + owner_role="root", + kind_payload={ + "agent_id": "aresumebg", + "subagent_type": "coder", + "prompt": "p", + "model_override": None, + "launch_mode": "background", + }, + ) + + old_spec = _agent_spec("aoldresume1") + store.create_task(old_spec) + store.write_runtime( + old_spec.id, + TaskRuntime(status="failed", finished_at=time.time(), updated_at=time.time()), + ) + new_spec = _agent_spec("anewresume1") + store.create_task(new_spec) + store.write_runtime( + new_spec.id, + TaskRuntime(status="running", updated_at=time.time(), heartbeat_at=time.time()), + ) + + async def _alive() -> None: + await asyncio.sleep(3600) + + live = asyncio.create_task(_alive()) + manager._live_agent_tasks[new_spec.id] = live + try: + manager.recover() + assert runtime.subagent_store.require_instance("aresumebg").status == "running_background" + finally: + live.cancel() + with contextlib.suppress(asyncio.CancelledError): + await live + manager._live_agent_tasks.pop(new_spec.id, None) + + def test_mark_task_running_does_not_overwrite_terminal_state(runtime): manager = runtime.background_tasks store = manager.store From 6b20b55facc9fb2fcb93c6039591090a5da71626 Mon Sep 17 00:00:00 2001 From: mohamed-elkholy95 Date: Fri, 29 May 2026 11:34:21 -0400 Subject: [PATCH 05/12] fix(tui): stop rendering the todo list twice during an in-flight turn When a turn is in flight the pinned status tail already renders the todo list under its verb spinner. The background-task status line was reading the same `_latest_todos` and appending the rows again, so the list showed twice while the agent worked. Restrict the duplicated rows to the between-turns case (show_verb=True) where the background line is the only surface; suppress them when the pinned tail is active. --- src/pythinker_code/ui/shell/prompt.py | 11 ++++---- .../test_visualize_running_prompt.py | 27 +++++++++++++++++++ 2 files changed, 32 insertions(+), 6 deletions(-) diff --git a/src/pythinker_code/ui/shell/prompt.py b/src/pythinker_code/ui/shell/prompt.py index afe98e47..6d72fcee 100644 --- a/src/pythinker_code/ui/shell/prompt.py +++ b/src/pythinker_code/ui/shell/prompt.py @@ -2647,12 +2647,11 @@ def _render_background_working_status( detail_text = detail if _display_width(frame_text + detail_text) > columns: detail_text = _truncate_right(detail_text, columns - _display_width(frame_text)) - fragments = FormattedText([(muted_style, frame_text + detail_text)]) - todo_rows = self._render_background_todo_rows(columns) - if todo_rows: - ensure_prompt_newline(fragments) - fragments.extend(todo_rows) - return fragments + # ``show_verb=False`` means an in-flight turn's pinned status tail is already + # rendering the todo list under its verb spinner. Repeating the rows here + # would print the same todo list twice while the agent works, so this branch + # shows only the background-task count line. + return FormattedText([(muted_style, frame_text + detail_text)]) def _background_task_counts(self) -> BgTaskCounts: provider = getattr(self, "_background_task_count_provider", None) diff --git a/tests/ui_and_conv/test_visualize_running_prompt.py b/tests/ui_and_conv/test_visualize_running_prompt.py index fcfee032..0a174590 100644 --- a/tests/ui_and_conv/test_visualize_running_prompt.py +++ b/tests/ui_and_conv/test_visualize_running_prompt.py @@ -323,6 +323,33 @@ def render_pinned_status_tail(self, columns: int): # noqa: ARG002 assert "…" not in text +def test_background_status_omits_todos_when_verb_pinned() -> None: + """When the pinned status tail is active (show_verb=False) it already renders + the todo list under the verb spinner; the background-task line must NOT repeat + it, or the same todo list renders twice while the agent works.""" + session = object.__new__(CustomPromptSession) + session._background_task_count_provider = lambda: BgTaskCounts(agent=3) + session._status_block_provider = None + session._latest_todos = ( + TodoDisplayItem(title="Security vulnerability scan", status="in_progress"), + TodoDisplayItem(title="Code quality review", status="pending"), + ) + + # show_verb=False ⟺ an in-flight turn's pinned tail is already showing todos. + pinned = CustomPromptSession._render_background_working_status(session, 100, show_verb=False) + pinned_text = "".join(item[1] for item in pinned) + assert "3 background agents" in pinned_text + assert "Security vulnerability scan" not in pinned_text + assert "Code quality review" not in pinned_text + + # Between turns (no pinned tail) the background line is the only surface, so + # it must still carry the todos. + standalone = CustomPromptSession._render_background_working_status(session, 100, show_verb=True) + standalone_text = "".join(item[1] for item in standalone) + assert "Security vulnerability scan" in standalone_text + assert "Code quality review" in standalone_text + + def test_running_prompt_hides_placeholder() -> None: view = object.__new__(_PromptLiveView) view._turn_ended = False From 1105e115acad2ebb9cb3d5247df2d589fb72f9fc Mon Sep 17 00:00:00 2001 From: mohamed-elkholy95 Date: Fri, 29 May 2026 11:34:36 -0400 Subject: [PATCH 06/12] fix(background): steer away from blocking on one task when siblings run Blocking on a single task with TaskOutput(block=true) waits only for that task and freezes the turn until the slowest sibling finishes, so completion notifications for the others land with no listener. Add that guidance to the TaskOutput tool description, the Agent tool's next_step hints, and the idle-completion system-reminder (which now reports how many background tasks are still running). Steers the model to return control and rely on automatic re-wake instead. --- src/pythinker_code/tools/agent/__init__.py | 5 +++ src/pythinker_code/tools/background/output.md | 1 + src/pythinker_code/ui/shell/__init__.py | 34 +++++++++++++++---- tests/tools/test_tool_descriptions.py | 1 + .../test_background_idle_reminder.py | 25 ++++++++++++++ 5 files changed, 60 insertions(+), 6 deletions(-) create mode 100644 tests/ui_and_conv/test_background_idle_reminder.py diff --git a/src/pythinker_code/tools/agent/__init__.py b/src/pythinker_code/tools/agent/__init__.py index 640e727f..29b66b3f 100644 --- a/src/pythinker_code/tools/agent/__init__.py +++ b/src/pythinker_code/tools/agent/__init__.py @@ -430,6 +430,11 @@ async def _run_in_background(self, params: Params) -> ToolReturnValue: "next_step: Use TaskOutput with this task_id for a non-blocking status/output " "snapshot. Only set block=true when you intentionally want to wait." ), + ( + "next_step: If you launched several agents, do not block=true on any single " + "one — blocking waits only for that task and freezes the turn until the " + "slowest finishes. Return control and rely on the completion notifications." + ), f'resume_hint: Use Agent(resume="{agent_id}", prompt="...") to continue this ' "instance later.", ] diff --git a/src/pythinker_code/tools/background/output.md b/src/pythinker_code/tools/background/output.md index 8e772430..ec38e10f 100644 --- a/src/pythinker_code/tools/background/output.md +++ b/src/pythinker_code/tools/background/output.md @@ -6,6 +6,7 @@ Guidelines: - Prefer relying on automatic completion notifications. Use this tool only when you need task output before the automatic notification arrives. - By default this tool is non-blocking and returns a current status/output snapshot. - Use `block=true` only when you intentionally want to wait for completion or timeout. +- When several background tasks are running, do not `block=true` on a single one — blocking waits only for that task and freezes the turn until the slowest finishes. Return control and rely on the automatic completion notifications. - This tool returns structured task metadata, a fixed-size output preview, and an `output_path` for the full log. - When the preview is truncated, use `ReadFile` with the returned `output_path` to inspect the full log in pages. - This tool works with the generic background task system and should remain the primary read path for future task types, not just bash. diff --git a/src/pythinker_code/ui/shell/__init__.py b/src/pythinker_code/ui/shell/__init__.py index ee7478b4..cd77abfe 100644 --- a/src/pythinker_code/ui/shell/__init__.py +++ b/src/pythinker_code/ui/shell/__init__.py @@ -94,6 +94,25 @@ class _PromptEvent: """Explicit skill/flow prefixes that should remain visible in transcript.""" +def _background_idle_reminder(active_running: int) -> str: + """Build the system-reminder injected when background tasks finish while idle. + + When sibling tasks are still running, steer the model to return control and + rely on the automatic re-wake instead of blocking on a single task with + ``TaskOutput(block=true)`` — blocking on one freezes the turn until the + slowest of them finishes. + """ + body = "Background tasks completed while you were idle." + if active_running > 0: + noun = "task is" if active_running == 1 else "tasks are" + body += ( + f" {active_running} background {noun} still running. Do not block on a" + " single task with TaskOutput(block=true); return control now and you" + " will be automatically re-woken as each one finishes." + ) + return f"{body}" + + def _format_local_shell_output( *, stdout: str, stderr: str, returncode: int | None ) -> RenderableType | None: @@ -774,12 +793,15 @@ def _can_auto_trigger_pending() -> bool: deferred_bg_trigger = False logger.info("Background task completed while idle, triggering agent") resume_prompt.set() - ok = await self.run_soul_command( - "" - "Background tasks completed while you" - " were idle." - "" - ) + active_running = 0 + if isinstance(self.soul, PythinkerSoul): + with contextlib.suppress(Exception): + active_running = len( + list_task_views( + self.soul.runtime.background_tasks, active_only=True + ) + ) + ok = await self.run_soul_command(_background_idle_reminder(active_running)) console.print() if not ok: bg_auto_failures += 1 diff --git a/tests/tools/test_tool_descriptions.py b/tests/tools/test_tool_descriptions.py index 5c93e47f..74439b17 100644 --- a/tests/tools/test_tool_descriptions.py +++ b/tests/tools/test_tool_descriptions.py @@ -219,6 +219,7 @@ def test_task_output_description(task_output_tool: TaskOutput): - Prefer relying on automatic completion notifications. Use this tool only when you need task output before the automatic notification arrives. - By default this tool is non-blocking and returns a current status/output snapshot. - Use `block=true` only when you intentionally want to wait for completion or timeout. +- When several background tasks are running, do not `block=true` on a single one — blocking waits only for that task and freezes the turn until the slowest finishes. Return control and rely on the automatic completion notifications. - This tool returns structured task metadata, a fixed-size output preview, and an `output_path` for the full log. - When the preview is truncated, use `ReadFile` with the returned `output_path` to inspect the full log in pages. - This tool works with the generic background task system and should remain the primary read path for future task types, not just bash. diff --git a/tests/ui_and_conv/test_background_idle_reminder.py b/tests/ui_and_conv/test_background_idle_reminder.py new file mode 100644 index 00000000..cbd23aa1 --- /dev/null +++ b/tests/ui_and_conv/test_background_idle_reminder.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +from pythinker_code.ui.shell import _background_idle_reminder + + +def test_reminder_no_running_tasks_is_unchanged() -> None: + out = _background_idle_reminder(0) + assert out == ( + "Background tasks completed while you were idle." + ) + assert "block=true" not in out + + +def test_reminder_singular_steers_away_from_blocking() -> None: + out = _background_idle_reminder(1) + assert "1 background task is still running" in out + assert "Do not block on a single task with TaskOutput(block=true)" in out + assert out.startswith("") and out.endswith("") + + +def test_reminder_plural_steers_away_from_blocking() -> None: + out = _background_idle_reminder(3) + assert "3 background tasks are still running" in out + assert "return control now" in out + assert "re-woken" in out From 608c8cfd1a60a95fd7a49fb2fc79fe2dfcb9b77b Mon Sep 17 00:00:00 2001 From: mohamed-elkholy95 Date: Fri, 29 May 2026 11:34:36 -0400 Subject: [PATCH 07/12] fix(soul): nudge once when a turn ends on a bare statement of intent Models sometimes end a message with a transitional preamble ("Let me synthesize the findings into a unified report.") but attach no tool call and produce no result. The loop treats any tool-call-free message as the final answer, so the turn ends before the promised work is done. Detect that shape conservatively and inject a one-shot system-reminder asking the model to deliver the result or make the tool call. Capped at once per turn so a stubborn model can still finish. --- src/pythinker_code/soul/pythinkersoul.py | 67 ++++++++++++++++++++++ tests/core/test_unfinished_intent_nudge.py | 43 ++++++++++++++ 2 files changed, 110 insertions(+) create mode 100644 tests/core/test_unfinished_intent_nudge.py diff --git a/src/pythinker_code/soul/pythinkersoul.py b/src/pythinker_code/soul/pythinkersoul.py index 6524bb6e..f0a25493 100644 --- a/src/pythinker_code/soul/pythinkersoul.py +++ b/src/pythinker_code/soul/pythinkersoul.py @@ -225,6 +225,45 @@ def _malformed_empty_tool_call_summary( return "; ".join(missing_by_tool) +_UNFINISHED_INTENT_LEAD_RE = re.compile( + r"^(let me|let's|let us|now let me|now i'?ll|i'?ll|i will|i'?m going to|" + r"i am going to|first,?\s+i'?ll|next,?\s+i'?ll)\b", + re.IGNORECASE, +) +_UNFINISHED_INTENT_ACTION_RE = re.compile( + r"\b(synthesi|summari|writ|creat|compil|prepar|draft|put together|generat|" + r"produc|present|report|provid|run|check|cross-check|verif|read|search|" + r"analy|review|implement|fix|updat|build|gather|look|examin|continu|" + r"proceed|start|begin|assembl|consolidat|finaliz|outlin|map out|" + r"investigat|explor|dig into|pull together)\w*", + re.IGNORECASE, +) + + +def _looks_like_unfinished_intent(text: str) -> bool: + """Return True when *text* is essentially a statement of intent to act, with + no actual result delivered. + + Models sometimes end a message with a transitional preamble such as + "Let me synthesize the findings into a unified report." but attach no tool + call and produce no result. The agent loop treats any tool-call-free message + as the final answer, so the turn ends before the promised work is done. This + detects that shape (conservatively) so the loop can nudge one more step. + """ + text = text.strip() + if not text or len(text) > 400 or text.endswith("?"): + return False + sentences = [s.strip() for s in re.split(r"[.!\n]+", text) if s.strip()] + if not sentences: + return False + last = sentences[-1] + if "let me know" in last.lower(): + return False + return bool( + _UNFINISHED_INTENT_LEAD_RE.match(last) and _UNFINISHED_INTENT_ACTION_RE.search(last) + ) + + @dataclass(frozen=True, slots=True) class StepOutcome: stop_reason: StepStopReason @@ -1087,6 +1126,9 @@ async def _agent_loop(self) -> TurnOutcome: step_no = 0 self._current_step_no = 0 + # One-shot per turn: nudge at most once when a step ends on a bare + # statement of intent (see `_looks_like_unfinished_intent`). + self._intent_nudge_used = False while True: step_no += 1 if step_no > self._loop_control.max_steps_per_turn: @@ -1436,6 +1478,31 @@ async def _pythinker_core_step_with_retry() -> StepResult: if result.tool_calls: return None + + # A tool-call-free message normally ends the turn. If it is only a + # restatement of intent ("Let me synthesize the findings…") with no + # result, nudge the model to actually deliver — but at most once per + # turn so a stubborn model can still finish. + if not getattr(self, "_intent_nudge_used", False) and _looks_like_unfinished_intent( + result.message.extract_text(" ") + ): + self._intent_nudge_used = True + await self._context.append_message( + Message( + role="user", + content=[ + system_reminder( + "Your previous message stated an intention to act (for example " + "to produce a report or run a tool) but included no tool call and " + "no actual result, which would normally end your turn. Either " + "produce the promised result now, in full, or make the necessary " + "tool call. Do not reply with only a restatement of intent." + ) + ], + ) + ) + return None + return StepOutcome(stop_reason="no_tool_calls", assistant_message=result.message) async def _grow_context(self, result: StepResult, tool_results: list[ToolResult]): diff --git a/tests/core/test_unfinished_intent_nudge.py b/tests/core/test_unfinished_intent_nudge.py new file mode 100644 index 00000000..9add7e43 --- /dev/null +++ b/tests/core/test_unfinished_intent_nudge.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +import pytest + +from pythinker_code.soul.pythinkersoul import _looks_like_unfinished_intent + + +@pytest.mark.parametrize( + "text", + [ + # The exact message that ended the deep-scan turn with no report. + "All 4 agents completed. Let me synthesize the findings into a unified report.", + "Let me synthesize the findings into a unified report.", + "I'll now write the full report.", + "Let me cross-check the most critical findings before presenting the report.", + "Next, I'll compile the results into a summary.", + "Now let me gather the remaining details.", + ], +) +def test_detects_unfinished_intent(text: str) -> None: + assert _looks_like_unfinished_intent(text) is True + + +@pytest.mark.parametrize( + "text", + [ + "", + " ", + # Closing offer, not a promise of work this turn. + "Let me know if you need anything else.", + # No intent lead in the final sentence. + "Done. The bug is fixed and all tests pass.", + "The report is ready above.", + # A question hands control back to the user. + "Should I proceed with the refactor, or wait?", + # Intent lead but no action verb tied to producing work. + "I'll keep that in mind.", + # A real, substantive answer is long enough not to be a bare preamble. + "Let me explain the architecture. " + ("The system has many layers. " * 20), + ], +) +def test_ignores_non_intent_or_substantive(text: str) -> None: + assert _looks_like_unfinished_intent(text) is False From f1f4460f67a7323fc8334f5d38e1ad85c9dcc6f5 Mon Sep 17 00:00:00 2001 From: mohamed-elkholy95 Date: Fri, 29 May 2026 11:34:36 -0400 Subject: [PATCH 08/12] fix: address CodeRabbit review findings across subsystems MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - background/manager: re-read TaskControl inside the recovery lock so a kill landing between list_views() and lock acquisition is honored as killed, not mislabeled lost; derive the subagent record from the runtime that actually won the terminal race instead of the requested outcome. - cli: guard SIGQUIT registration behind hasattr — it is POSIX-only and was crashing CLI startup on Windows. - grep_local: make _iter_python_search_files lazy so the per-file timeout check fires during discovery instead of after the whole tree is walked. - web/fetch: drop the misleading await on the synchronous response.release(). - ui/shell/slash: add missing return type annotation. - tests: add a fail-closed validation test for SetTodoList params. --- src/pythinker_code/background/manager.py | 18 ++++++++++++------ src/pythinker_code/cli/__init__.py | 7 ++++++- src/pythinker_code/tools/file/grep_local.py | 11 +++++++---- src/pythinker_code/tools/web/fetch.py | 2 +- src/pythinker_code/ui/shell/slash.py | 2 +- tests/tools/test_todo.py | 7 +++++++ 6 files changed, 34 insertions(+), 13 deletions(-) diff --git a/src/pythinker_code/background/manager.py b/src/pythinker_code/background/manager.py index 13f34011..7c1e74d9 100644 --- a/src/pythinker_code/background/manager.py +++ b/src/pythinker_code/background/manager.py @@ -593,6 +593,7 @@ def recover(self) -> None: # and the status write. with self._store._runtime_lock(view.spec.id): # pyright: ignore[reportPrivateUsage] fresh_runtime = self._store.read_runtime(view.spec.id) + fresh_control = self._store.read_control(view.spec.id) if is_terminal_status(fresh_runtime.status): continue fresh_progress = ( @@ -607,10 +608,10 @@ def recover(self) -> None: runtime = fresh_runtime.model_copy() runtime.finished_at = now runtime.updated_at = now - if view.control.kill_requested_at is not None: + if fresh_control.kill_requested_at is not None: runtime.status = "killed" runtime.interrupted = True - runtime.failure_reason = view.control.kill_reason or "Killed during recovery" + runtime.failure_reason = fresh_control.kill_reason or "Killed during recovery" else: runtime.status = "lost" runtime.failure_reason = ( @@ -799,10 +800,15 @@ def finalize_agent_task( and self._runtime is not None and self._runtime.subagent_store is not None ): - subagent_status: SubagentStatus = ( - "idle" if outcome == "completed" else "killed" if outcome == "killed" else "failed" - ) - self._runtime.subagent_store.update_instance(agent_id, status=subagent_status) + # _mark_task_*() returns early when the task is already terminal, so a + # kill/timeout race can leave the authoritative TaskRuntime at a + # different terminal status than this call's `outcome`. Derive the + # subagent status from the runtime that actually won (matching + # recover()'s reconciliation) so the two records never diverge. + final_status = self._store.read_runtime(task_id).status + subagent_status = _subagent_status_for_task_status(final_status) + if subagent_status is not None: + self._runtime.subagent_store.update_instance(agent_id, status=subagent_status) def _mark_task_running(self, task_id: str) -> None: with self._store._runtime_lock(task_id): # pyright: ignore[reportPrivateUsage] diff --git a/src/pythinker_code/cli/__init__.py b/src/pythinker_code/cli/__init__.py index 83b82076..b72322a1 100644 --- a/src/pythinker_code/cli/__init__.py +++ b/src/pythinker_code/cli/__init__.py @@ -1057,7 +1057,12 @@ def _restore_term_and_exit(signum: int, frame: object) -> None: _signal.signal(signum, _signal.SIG_DFL) os.kill(os.getpid(), signum) - for _sig in (_signal.SIGTERM, _signal.SIGQUIT): + # SIGQUIT is POSIX-only; the signal module does not expose it on Windows. + _signals_to_trap = [_signal.SIGTERM] + if hasattr(_signal, "SIGQUIT"): + _signals_to_trap.append(_signal.SIGQUIT) + + for _sig in _signals_to_trap: with contextlib.suppress(OSError, ValueError): _signal.signal(_sig, _restore_term_and_exit) diff --git a/src/pythinker_code/tools/file/grep_local.py b/src/pythinker_code/tools/file/grep_local.py index ea0571d4..993f741f 100644 --- a/src/pythinker_code/tools/file/grep_local.py +++ b/src/pythinker_code/tools/file/grep_local.py @@ -14,6 +14,7 @@ import tempfile import time import zipfile +from collections.abc import Iterator from pathlib import Path from typing import override @@ -476,11 +477,14 @@ def _matches_python_type_filter(rel_path: str, file_type: str | None) -> bool: return any(fnmatch.fnmatch(rel_path, glob_pattern) for glob_pattern in globs) -def _iter_python_search_files(params: Params) -> list[Path]: +def _iter_python_search_files(params: Params) -> Iterator[Path]: + # Lazy by design: ``rglob`` is itself an iterator, so yielding candidates one + # at a time lets the consumer's per-file deadline check fire *during* + # discovery. Materializing the full list here would let a large tree blow the + # RG_TIMEOUT budget before the first timeout check could run. search_path = Path(os.path.expanduser(params.path)) search_base = search_path if search_path.is_dir() else search_path.parent candidates = [search_path] if search_path.is_file() else search_path.rglob("*") - files: list[Path] = [] excluded_vcs = {".git", ".svn", ".hg", ".bzr", ".jj", ".sl"} ignore_patterns = [] if params.include_ignored else _load_basic_ignore_patterns(search_base) for candidate in candidates: @@ -495,8 +499,7 @@ def _iter_python_search_files(params: Params) -> list[Path]: continue if not _matches_python_type_filter(rel_path, params.type): continue - files.append(candidate) - return files + yield candidate def _apply_python_pagination(lines: list[str], params: Params) -> tuple[list[str], str]: diff --git a/src/pythinker_code/tools/web/fetch.py b/src/pythinker_code/tools/web/fetch.py index fb06b748..202491f9 100644 --- a/src/pythinker_code/tools/web/fetch.py +++ b/src/pythinker_code/tools/web/fetch.py @@ -96,7 +96,7 @@ async def _get_revalidating_redirects( response = await session.get(current, headers=headers, allow_redirects=False) location = response.headers.get(aiohttp.hdrs.LOCATION) if response.status in _REDIRECT_STATUSES and location: - await response.release() + response.release() current = urljoin(str(response.url), location) continue return response diff --git a/src/pythinker_code/ui/shell/slash.py b/src/pythinker_code/ui/shell/slash.py index 1f303879..aab04033 100644 --- a/src/pythinker_code/ui/shell/slash.py +++ b/src/pythinker_code/ui/shell/slash.py @@ -612,7 +612,7 @@ async def report_error(app: Shell, args: str): ISSUE_URL = "https://github.com/TechMatrix-labs/pythinker-code/issues" - def _fallback_to_issues(): + def _fallback_to_issues() -> None: if not open_url_in_browser(ISSUE_URL): console.print(f"Please file the report at [underline]{ISSUE_URL}[/underline].") diff --git a/tests/tools/test_todo.py b/tests/tools/test_todo.py index 070e3473..a686d440 100644 --- a/tests/tools/test_todo.py +++ b/tests/tools/test_todo.py @@ -38,6 +38,13 @@ def test_todos_none_still_works(self): params = Params(todos=None) assert params.todos is None + def test_invalid_json_string_fails_validation(self): + """Invalid JSON strings must fail validation (fail-closed).""" + from pydantic import ValidationError + + with pytest.raises(ValidationError): + Params(todos="not valid json") # type: ignore[arg-type] + class TestSetTodoListOutputNotEmpty: """Regression test for issue #1710: SetTodoList storm. From e1fbe91635f0cf154f19d51e6d6555e6af112843 Mon Sep 17 00:00:00 2001 From: mohamed-elkholy95 Date: Fri, 29 May 2026 11:41:26 -0400 Subject: [PATCH 09/12] ci: satisfy spell-check and host formatting gates - typos: accept the verb stems (prepar/generat/provid/updat/examin/continu) used as word-prefix alternatives in the unfinished-intent detection regex. - ruff format packages/pythinker-host/tests/test_local_host.py, which was left unformatted and failed `make check-pythinker-host` across the matrix. --- packages/pythinker-host/tests/test_local_host.py | 4 +--- pyproject.toml | 9 +++++++++ 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/packages/pythinker-host/tests/test_local_host.py b/packages/pythinker-host/tests/test_local_host.py index 0c42bb36..62e02cd3 100644 --- a/packages/pythinker-host/tests/test_local_host.py +++ b/packages/pythinker-host/tests/test_local_host.py @@ -231,9 +231,7 @@ def _record_killpg(pgid: int, sig: int) -> None: @pytest.mark.skipif(os.name == "nt", reason="POSIX process-group signal path") -async def test_kill_signals_running_process( - local_host: LocalHost, monkeypatch: pytest.MonkeyPatch -): +async def test_kill_signals_running_process(local_host: LocalHost, monkeypatch: pytest.MonkeyPatch): """A still-running process is killed via its process group.""" import pythinker_host.local as local_module diff --git a/pyproject.toml b/pyproject.toml index 9c901108..c31a855b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -178,4 +178,13 @@ Encrypter = "Encrypter" # Hex session IDs (e.g. 06ba6c38) contain "ba". ba = "ba" uest = "uest" +# Verb stems in the unfinished-intent detection regex (soul/pythinkersoul.py) +# match word prefixes ("prepar" → prepare/preparing); kept as bare stems on +# purpose so they cover every inflection. +prepar = "prepar" +generat = "generat" +provid = "provid" +updat = "updat" +examin = "examin" +continu = "continu" From 6e515f1e9e438cd8ad9797a45c919c6e8379b50c Mon Sep 17 00:00:00 2001 From: mohamed-elkholy95 Date: Fri, 29 May 2026 11:46:30 -0400 Subject: [PATCH 10/12] ci: fix Windows host pyright and main-package formatting gates - test_local_host: route POSIX-only os.killpg / signal.SIGKILL through Any holders so the strict host pyright gate passes on the win32 platform stubs (the test is already skipped at runtime on Windows). Verified with `pyright --pythonplatform Windows`. - ruff format tests/ui_and_conv/test_shell_feedback_slash.py, which failed the main-package `ruff format --check` gate. --- packages/pythinker-host/tests/test_local_host.py | 10 ++++++++-- tests/ui_and_conv/test_shell_feedback_slash.py | 8 ++++++-- 2 files changed, 14 insertions(+), 4 deletions(-) diff --git a/packages/pythinker-host/tests/test_local_host.py b/packages/pythinker-host/tests/test_local_host.py index 62e02cd3..c1dc6911 100644 --- a/packages/pythinker-host/tests/test_local_host.py +++ b/packages/pythinker-host/tests/test_local_host.py @@ -6,6 +6,7 @@ import sys from collections.abc import Generator from pathlib import Path, PurePosixPath, PureWindowsPath +from typing import Any import pytest @@ -238,7 +239,12 @@ async def test_kill_signals_running_process(local_host: LocalHost, monkeypatch: process = await local_host.exec(*_python_code_args("import time; time.sleep(30)")) assert process.returncode is None - real_killpg = local_module.os.killpg + # os.killpg / signal.SIGKILL are POSIX-only; this test is skipped on Windows + # but pyright still type-checks the body, so route them through Any holders + # to keep the strict host gate green on the win32 platform stubs. + os_any: Any = local_module.os + signal_any: Any = signal + real_killpg = os_any.killpg sent: list[int] = [] def _record_killpg(pgid: int, sig: int) -> None: @@ -250,4 +256,4 @@ def _record_killpg(pgid: int, sig: int) -> None: await process.kill() await process.wait() - assert sent and sent[0] == signal.SIGKILL + assert sent and sent[0] == signal_any.SIGKILL diff --git a/tests/ui_and_conv/test_shell_feedback_slash.py b/tests/ui_and_conv/test_shell_feedback_slash.py index 77f82f5a..e5cfd132 100644 --- a/tests/ui_and_conv/test_shell_feedback_slash.py +++ b/tests/ui_and_conv/test_shell_feedback_slash.py @@ -37,7 +37,9 @@ def test_opens_new_issue_url(self, monkeypatch) -> None: assert "new" in url def test_prints_success_when_browser_opens(self, monkeypatch) -> None: - monkeypatch.setattr("pythinker_code.utils.term.open_url_in_browser", Mock(return_value=True)) + monkeypatch.setattr( + "pythinker_code.utils.term.open_url_in_browser", Mock(return_value=True) + ) print_mock = Mock() monkeypatch.setattr(shell_slash.console, "print", print_mock) @@ -47,7 +49,9 @@ def test_prints_success_when_browser_opens(self, monkeypatch) -> None: assert "Opening" in output or "browser" in output.lower() def test_prints_url_when_browser_fails(self, monkeypatch) -> None: - monkeypatch.setattr("pythinker_code.utils.term.open_url_in_browser", Mock(return_value=False)) + monkeypatch.setattr( + "pythinker_code.utils.term.open_url_in_browser", Mock(return_value=False) + ) print_mock = Mock() monkeypatch.setattr(shell_slash.console, "print", print_mock) From 2f7afee77b53f8a5aa9090a696734f4e91316ced Mon Sep 17 00:00:00 2001 From: mohamed-elkholy95 Date: Fri, 29 May 2026 11:53:28 -0400 Subject: [PATCH 11/12] test: refresh default-config snapshot for new background fields The default-config dump snapshot was stale: earlier commits added `task_retention_days` and `max_output_bytes` to BackgroundConfig without updating tests/core/test_config.py, failing test-pythinker-code across the matrix. Add both fields in dump order. --- tests/core/test_config.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/core/test_config.py b/tests/core/test_config.py index b3a1dee9..d8badffa 100644 --- a/tests/core/test_config.py +++ b/tests/core/test_config.py @@ -42,6 +42,7 @@ def test_default_config_dump(): }, "background": { "max_running_tasks": 4, + "task_retention_days": 7, "read_max_bytes": 30000, "notification_tail_lines": 20, "notification_tail_chars": 3000, @@ -49,6 +50,7 @@ def test_default_config_dump(): "worker_heartbeat_interval_ms": 5000, "worker_stale_after_ms": 15000, "kill_grace_period_ms": 2000, + "max_output_bytes": 52428800, "keep_alive_on_exit": False, "agent_task_timeout_s": 3600, "print_wait_ceiling_s": 3600, From a5b5acc38b9ca231eb3ffd5fbdfc67aa496c4fbf Mon Sep 17 00:00:00 2001 From: mohamed-elkholy95 Date: Fri, 29 May 2026 13:00:22 -0400 Subject: [PATCH 12/12] fix: log idle-reminder task-count failures; test grep fallback via public API Address two CodeRabbit review findings on PR #13: - ui/shell: replace contextlib.suppress(Exception) around the idle-reminder active-task count with a try/except that logs at debug (active_running still falls back to 0), so background-task introspection failures are no longer invisible. - tests/tools/test_grep: exercise the Python fallback's wall-clock bound through the public Grep() API with a forced ripgrep-unavailable path instead of calling _python_grep directly, and drop the now-unused import. --- src/pythinker_code/ui/shell/__init__.py | 8 +++++++- tests/tools/test_grep.py | 26 ++++++++++++++++--------- 2 files changed, 24 insertions(+), 10 deletions(-) diff --git a/src/pythinker_code/ui/shell/__init__.py b/src/pythinker_code/ui/shell/__init__.py index cd77abfe..f7aa7ff0 100644 --- a/src/pythinker_code/ui/shell/__init__.py +++ b/src/pythinker_code/ui/shell/__init__.py @@ -795,12 +795,18 @@ def _can_auto_trigger_pending() -> bool: resume_prompt.set() active_running = 0 if isinstance(self.soul, PythinkerSoul): - with contextlib.suppress(Exception): + try: active_running = len( list_task_views( self.soul.runtime.background_tasks, active_only=True ) ) + except Exception: + logger.debug( + "Failed to compute active background task count for " + "idle reminder", + exc_info=True, + ) ok = await self.run_soul_command(_background_idle_reminder(active_running)) console.print() if not ok: diff --git a/tests/tools/test_grep.py b/tests/tools/test_grep.py index 599fdcfe..1f3b2527 100644 --- a/tests/tools/test_grep.py +++ b/tests/tools/test_grep.py @@ -15,7 +15,6 @@ Params, _build_rg_args, _find_existing_rg, - _python_grep, _rg_binary_name, _strip_path_prefix, ) @@ -1041,24 +1040,33 @@ async def test_grep_allows_env_example(grep_tool: Grep): assert ".env.example" in result.output -def test_python_fallback_bounds_wall_clock(monkeypatch, tmp_path): +async def test_python_fallback_bounds_wall_clock(monkeypatch, tmp_path): """The Python fallback caps total wall-clock like the ripgrep path, and surfaces a partial-results notice when it does.""" + + async def fail_rg_path() -> str: + raise RuntimeError("Failed to download ripgrep binary") + + monkeypatch.setattr("pythinker_code.tools.file.grep_local._ensure_rg_path", fail_rg_path) + (tmp_path / "a.txt").write_text("needle\n") (tmp_path / "b.txt").write_text("needle\n") - # First monotonic() reading establishes the deadline; every later reading is - # far in the future, so the deadline is already blown on the first file. - readings = iter([0.0]) + # A strictly increasing clock: the first reading inside the fallback sets the + # deadline, and the next reading (during the file walk) is far enough ahead to + # blow it immediately. The +1e9 step keeps that true regardless of any other + # monotonic() calls made elsewhere on the public Grep() path. + ticks = [0.0] def _fake_monotonic() -> float: - return next(readings, 1e9) + value = ticks[0] + ticks[0] += 1e9 + return value monkeypatch.setattr(grep_module.time, "monotonic", _fake_monotonic) - result = _python_grep( - Params(pattern="needle", path=str(tmp_path), output_mode="files_with_matches"), - "forced fallback", + result = await Grep()( + Params(pattern="needle", path=str(tmp_path), output_mode="files_with_matches") ) assert f"Search exceeded {grep_module.RG_TIMEOUT}s" in result.message