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..c1dc6911 100644
--- a/packages/pythinker-host/tests/test_local_host.py
+++ b/packages/pythinker-host/tests/test_local_host.py
@@ -2,9 +2,11 @@
import asyncio
import os
+import signal
import sys
from collections.abc import Generator
from pathlib import Path, PurePosixPath, PureWindowsPath
+from typing import Any
import pytest
@@ -196,3 +198,62 @@ 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
+
+ # 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:
+ 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_any.SIGKILL
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"
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/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 2af7f51c..7c1e74d9 100644
--- a/src/pythinker_code/background/manager.py
+++ b/src/pythinker_code/background/manager.py
@@ -5,9 +5,10 @@
import signal
import subprocess
import sys
+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
@@ -18,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 (
@@ -31,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__(
@@ -158,6 +179,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 +195,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 +442,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 +460,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 +536,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]:
@@ -493,32 +558,26 @@ 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():
- if is_terminal_status(view.runtime.status):
- continue
+ 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":
- 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, live_agent_ids=live_agent_ids)
+ continue
+ if is_terminal_status(view.runtime.status):
continue
last_progress_at = (
view.runtime.heartbeat_at
@@ -529,38 +588,109 @@ 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
+ # 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)
+ fresh_control = self._store.read_control(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 = fresh_runtime.model_copy()
+ runtime.finished_at = now
+ runtime.updated_at = now
+ if fresh_control.kill_requested_at is not None:
+ runtime.status = "killed"
+ runtime.interrupted = True
+ runtime.failure_reason = fresh_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 _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
+ 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
- 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"
- )
+ 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, live_agent_ids)
+
+ 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``.
+
+ 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 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:
+ 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] = []
@@ -636,34 +766,81 @@ 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
+ ):
+ # _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:
- 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 +848,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 +869,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 +892,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..cf4397e5 100644
--- a/src/pythinker_code/background/store.py
+++ b/src/pythinker_code/background/store.py
@@ -1,8 +1,12 @@
from __future__ import annotations
+import contextlib
import json
import os
import re
+import shutil
+import time
+from contextlib import contextmanager
from pathlib import Path
from pydantic import BaseModel, ValidationError
@@ -108,7 +112,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)
@@ -262,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/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..b72322a1 100644
--- a/src/pythinker_code/cli/__init__.py
+++ b/src/pythinker_code/cli/__init__.py
@@ -1041,6 +1041,31 @@ 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)
+
+ # 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)
+
try:
switch_target, exit_code = asyncio.run(_reload_loop(session_id))
except (typer.BadParameter, typer.Exit):
@@ -1336,6 +1361,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 +1380,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..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)
@@ -116,6 +119,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/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/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/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/tools/file/grep_local.py b/src/pythinker_code/tools/file/grep_local.py
index bbc3bdfe..993f741f 100644
--- a/src/pythinker_code/tools/file/grep_local.py
+++ b/src/pythinker_code/tools/file/grep_local.py
@@ -12,7 +12,9 @@
import stat
import tarfile
import tempfile
+import time
import zipfile
+from collections.abc import Iterator
from pathlib import Path
from typing import override
@@ -475,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:
@@ -494,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]:
@@ -528,7 +532,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 +590,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/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/src/pythinker_code/tools/web/fetch.py b/src/pythinker_code/tools/web/fetch.py
index bdbe103e..202491f9 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:
+ 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/__init__.py b/src/pythinker_code/ui/shell/__init__.py
index ee7478b4..f7aa7ff0 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,21 @@ 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):
+ 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:
bg_auto_failures += 1
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/src/pythinker_code/ui/shell/slash.py b/src/pythinker_code/ui/shell/slash.py
index c15c0429..aab04033 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):
+ 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].")
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..4dc29433 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
@@ -691,6 +693,229 @@ 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
+
+
+@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
@@ -1166,3 +1391,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_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()
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/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,
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
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..1f3b2527 100644
--- a/tests/tools/test_grep.py
+++ b/tests/tools/test_grep.py
@@ -9,6 +9,7 @@
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,
@@ -1037,3 +1038,35 @@ async def test_grep_allows_env_example(grep_tool: Grep):
)
assert not result.is_error
assert ".env.example" in result.output
+
+
+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")
+
+ # 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:
+ value = ticks[0]
+ ticks[0] += 1e9
+ return value
+
+ monkeypatch.setattr(grep_module.time, "monotonic", _fake_monotonic)
+
+ 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
diff --git a/tests/tools/test_todo.py b/tests/tools/test_todo.py
index 9a5173ce..a686d440 100644
--- a/tests/tools/test_todo.py
+++ b/tests/tools/test_todo.py
@@ -15,6 +15,37 @@ 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
+
+ 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.
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
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..e5cfd132 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,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("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 +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("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)
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