kenneth/Auto-Trader
1
0 Fork 0
Code Issues Pull Requests Actions 16 Packages Projects Releases Wiki Activity

fix(agent): stall watchdog, bash tree-kill timeout, compaction verification ledger

Browse Source 
A 2026-08-20 INTC weekly-update run froze for 20+ minutes: a bash call at
19:20:55 never returned. On Windows, subprocess.run(shell=True, timeout=120)
killed only cmd.exe; the piped findstr grandchildren inherited the stdout/
stderr pipe handles, so the follow-up communicate() blocked forever. The run
made no forward progress (no LLM completion, no tool result), nothing failed
it, and no state.json was ever written — the UI kept showing it as "running".

Three fixes:

- bash_tool.py: spawn in a dedicated process group and on timeout kill the
  WHOLE tree (taskkill /F /T /PID on Windows, killpg on POSIX) instead of
  only the shell; return a structured tool_timeout error in bounded time.
- loop.py: add a daemon run-stall-watchdog that fails a run after
  VIBE_TRADING_RUN_STALL_TIMEOUT_SECONDS (default 1800s) of no LLM
  completion or tool result (heartbeats do not count — a hung tool keeps
  emitting them). It writes a failed end trace record + state.json, emits
  stall_warning/stalled events, and sets the cancel event; the terminal
  status honors _stall_reason instead of overwriting it with success.
- loop.py + tools.py + financial_rigor_tool.py: preserve verification
  evidence across compaction. _verification_ledger() re-attaches
  financial_rigor results to the compacted context ("already verified, do
  NOT re-run"), and deterministic identical calls are served from a
  _called_identical cache (tool_result_cached trace event). Stops the
  post-compact re-run loop that re-executed the same 7 calcs 5-9x each.

config: new vibe_trading_run_stall_timeout_seconds
tests: bash tool timeout/tree-kill/heredoc; verification ledger; stall
timeout override
This commit is contained in:
Will Liao committed 2026-08-20 22:24:20 -04:00
1 parent e897f5966b
commit 89c2a45387
8 files changed
+507 -27
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.gitignore
+1
View File
@@ -75,6 +75,7 @@ data/overnight*.log
data/*.log
data/test_write.txt
agent/.pytest_tmp/
agent/.pytest-tmp/
agent/vibe_trading.egg-info/
agent/vibe_trading_ai.egg-info/
agent/tests/e2e_backtest/
agent/src/agent/loop.py
+260 -1
View File
@@ -141,6 +141,23 @@ def _goal_max_continuations() -> int:
from src.config.accessor import get_env_config
return get_env_config().agent_tuning.vibe_trading_goal_max_continuations
def _stall_timeout_seconds() -> float:
"""Return the run-stall watchdog timeout in seconds (0/negative disables).
A run that makes no forward progress (no LLM completion, no tool result)
for this long is treated as a zombie and failed explicitly with a clear
reason instead of staying "running" forever with no state.json
(recurring 2026-08 zombie runs). Heartbeats do NOT count as progress: a
hung tool keeps emitting heartbeats, which is exactly the case the
watchdog must catch.
"""
ov = _override("STALL_TIMEOUT_SECONDS")
if ov is not None:
return ov
from src.config.accessor import get_env_config
return get_env_config().agent_tuning.vibe_trading_run_stall_timeout_seconds
logger = logging.getLogger(__name__)
@@ -324,6 +341,61 @@ def _summary_chunks(msgs: list, limit: int = SUMMARY_CHUNK_CHARS) -> list[str]:
return chunks or ["[]"]
def _verification_ledger(messages: list) -> str:
"""Extract terse deterministic-verification records from a message list.
Walks the conversation for successful financial_rigor tool results
(the deterministic calculator/verifier) and renders each as a short
"already verified" line. Re-attached to the compressed context after
auto-compact so the model does not re-run identical expressions it can
no longer see (2026-08-20 INTC run re-ran the same calcs 5-9x each
after compaction cleared the outputs).
Args:
messages: Message list to scan (tool results only).
Returns:
Newline-joined ledger lines, or an empty string when nothing found.
"""
lines: list[str] = []
for msg in messages:
if msg.get("role") != "tool":
continue
content = msg.get("content", "")
if not isinstance(content, str):
continue
try:
payload = json.loads(content)
except Exception: # noqa: BLE001 - non-JSON results are skipped
continue
if not isinstance(payload, dict) or payload.get("status") != "ok":
continue
command = payload.get("command")
if command == "calc" and payload.get("result_exact") is not None:
expr = payload.get("expr", "?")
lines.append(f"calc {expr} = {payload.get("result_exact")}")
elif command == "verify_market_cap" and payload.get("verdict") is not None:
lines.append(f"market_cap verdict={payload.get("verdict")} dev={payload.get("deviation_pct")}%")
elif command == "verify_valuation" and payload.get("metrics"):
metrics = payload["metrics"]
if isinstance(metrics, dict) and metrics:
summary = ", ".join(f"{k}={v}" for k, v in list(metrics.items())[:8])
lines.append(f"valuation {summary}")
elif command == "cross_validate" and payload.get("all_consistent") is not None:
lines.append(f"cross_validate field={payload.get("field", "?")} consistent={payload.get("all_consistent")}")
elif command == "benford" and payload.get("reliable") is not None:
lines.append(f"benford reliable={payload.get("reliable")} conformity={payload.get("conformity", "?")}")
# Deduplicate while preserving order; cap the ledger size.
seen: set[str] = set()
unique: list[str] = []
for line in lines:
if line not in seen:
seen.add(line)
unique.append(line)
if len(unique) >= 60:
break
return "\n".join(unique)
def _microcompact(messages: list) -> None:
"""Layer 1: silently prune old tool results, keeping the most recent N intact.
@@ -836,6 +908,16 @@ class AgentLoop:
self._released_fallback = False
self._released_fallback_reason: str | None = None
self._written_files: set[str] = set()
self._stall_reason: str | None = None
self._last_activity_wall: float = 0.0
self._run_done = threading.Event()
# Identical deterministic tool calls (e.g. financial_rigor calc with
# the same expression) are served from this cache instead of being
# re-executed. Regression: after auto-compact cleared earlier tool
# results, the model re-ran the same financial_rigor expressions
# 5-9x each (2026-08-20 INTC run) because it could no longer see its
# own verification records.
self._called_identical: dict[tuple[str, str], str] = {}
def cancel(self) -> None:
"""Cancel the current loop.
@@ -926,6 +1008,10 @@ class AgentLoop:
self._released_fallback = False
self._released_fallback_reason = None
self._written_files = set()
self._stall_reason = None
self._last_activity_wall = _time.time()
self._run_done = threading.Event()
self._called_identical = {}
run_started_wall = _time.time()
state_store = RunStateStore()
@@ -993,6 +1079,20 @@ class AgentLoop:
goal_last_progress: tuple[int, int] | None = None
wrap_up_at = max(1, int(self.max_iterations * 0.8))
# Zombie-run watchdog: fail a run that makes no forward progress
# (no LLM completion, no tool result) for the stall timeout instead
# of leaving it "running" forever with no state.json. Heartbeats do
# not count as progress - a hung tool keeps emitting them.
stall_timeout = _stall_timeout_seconds()
if stall_timeout > 0:
watchdog = threading.Thread(
target=self._stall_watchdog,
args=(trace, run_dir, state_store, stall_timeout),
name="run-stall-watchdog",
daemon=True,
)
watchdog.start()
try:
while iteration < self.max_iterations:
if self._cancel_event.is_set():
@@ -1184,6 +1284,9 @@ class AgentLoop:
if self._cancel_event.is_set():
break
# An LLM response arrived - real progress for the stall watchdog.
self._last_activity_wall = _time.time()
usage = getattr(response, "usage_metadata", None)
if getattr(response, "response_model", None):
last_response_model = response.response_model
@@ -1528,6 +1631,7 @@ class AgentLoop:
trace.write({"type": "end", "iter": self._run_iteration, "status": "error", "reason": str(exc), "iterations": iteration})
trace.close()
state_store.mark_failure(run_dir, str(exc))
self._run_done.set()
return {
"status": "failed",
"error_code": error_code,
@@ -1544,7 +1648,12 @@ class AgentLoop:
# returned dict so SessionService can surface a meaningful UI
# message instead of "Execution failed: unknown" (issue #114).
final_reason: str | None = None
if self._cancel_event.is_set():
if self._stall_reason is not None:
# The stall watchdog already wrote the failed state; keep this
# run's terminal status honest instead of overwriting it.
final_reason = self._stall_reason
final_status = "failed"
elif self._cancel_event.is_set():
final_reason = "cancelled by user"
state_store.mark_cancelled(run_dir, final_reason)
final_status = "cancelled"
@@ -1629,6 +1738,8 @@ class AgentLoop:
if final_reason is not None:
result["reason"] = final_reason
self._run_done.set()
cf_warnings = compute_content_filter_warnings(
content_filter_count, max(1, iteration),
)
@@ -1637,6 +1748,75 @@ class AgentLoop:
return result
def _stall_watchdog(
self,
trace: TraceWriter,
run_dir: Path,
state_store: Any,
stall_timeout: float,
) -> None:
"""Fail the run when no forward progress happens for the stall timeout.
A run can hang inside a tool or provider call without ever raising:
the bash subprocess stuck in ``communicate()`` (2026-08-20 INTC run,
19:20:55 -> never returned) is the canonical case. Such a run stays
"running" forever with no state.json - a zombie. This daemon thread
watches wall-clock progress (LLM completions and tool results only,
NOT heartbeats) and, once the stall timeout passes, writes a failed
end event and state, and sets the cancel event so the loop exits at
its next boundary.
Args:
trace: Trace writer for this run.
run_dir: Run directory for state.json.
state_store: RunStateStore.
stall_timeout: Stall threshold in seconds.
"""
# Capture the done-event locally: a reused AgentLoop replaces
# self._run_done when the next run starts, and this thread must
# not keep watching with the old run timeout.
run_done = self._run_done
warned = False
while not run_done.is_set():
wait_s = min(30.0, max(5.0, stall_timeout / 4))
if run_done.wait(timeout=wait_s):
return
idle = _time.time() - self._last_activity_wall
if idle < stall_timeout:
warned = False
continue
if not warned:
warned = True
self._emit(
"stall_warning",
{"idle_s": round(idle, 1), "timeout_s": stall_timeout},
)
continue
reason = (
"run stalled: no LLM completion or tool result for "
+ str(int(idle)) + "s (stall watchdog)"
)
self._stall_reason = reason
try:
trace.write(
{
"type": "end",
"iter": self._run_iteration,
"status": "failed",
"reason": reason,
"stalled": True,
}
)
except Exception: # noqa: BLE001 - best-effort
pass
try:
state_store.mark_failure(run_dir, reason)
except Exception: # noqa: BLE001 - best-effort
pass
self._emit("stalled", {"reason": reason})
self._cancel_event.set()
return
# -- Tool execution with read/write batching --------------------------------
def _process_tool_calls(
@@ -1698,6 +1878,36 @@ class AgentLoop:
react_trace.append({"type": "tool_skipped", "tool": tc.name})
continue
# Deterministic tools (e.g. financial_rigor calc) return the same
# result for the same args. After auto-compact cleared earlier
# tool outputs, the model used to re-run identical expressions
# 5-9x each (2026-08-20 INTC run) to re-verify numbers it could
# no longer see. Serve an identical prior call from cache instead.
if tool_def is not None and getattr(tool_def, "deterministic", False):
cache_key = self._identical_call_key(tc.name, tc.arguments)
if cache_key is not None and cache_key in self._called_identical:
cached = self._called_identical[cache_key]
messages.append(context.format_tool_result(tc.id, tc.name, cached))
trace.write({
"type": "tool_result_cached",
"iter": iteration,
"tool": tc.name,
"call_id": tc.id,
})
react_trace.append({"type": "tool_result_cached", "tool": tc.name})
self._emit(
"tool_result",
{
"tool": tc.name,
"status": "ok",
"elapsed_ms": 0,
"preview": cached[:200],
"call_id": tc.id,
"cached": True,
},
)
continue
if self._grounding is not None:
authorization = self._grounding.authorize_tool_call(
tc.name,
@@ -2219,6 +2429,24 @@ class AgentLoop:
"file write is a failure."
)
def _identical_call_key(self, tool_name: str, arguments: Mapping[str, Any]) -> tuple[str, str] | None:
"""Build a stable key identifying a deterministic tool invocation.
Args:
tool_name: Tool name.
arguments: Tool arguments.
Returns:
Tuple of (tool_name, canonical JSON of the args) usable as a
dict key, or None when the args cannot be serialized.
"""
try:
normalized = _normalize_tool_run_dir(dict(arguments or {}), self.memory.run_dir)
canonical = json.dumps(normalized, sort_keys=True, ensure_ascii=False, default=str)
except Exception: # noqa: BLE001 - un-serializable args are never cached
return None
return (tool_name, canonical)
def _finalize_tool_result(
self,
tc: Any,
@@ -2248,6 +2476,9 @@ class AgentLoop:
if update_memory:
self._update_memory(tc.name)
# A tool completed - real progress for the stall watchdog.
self._last_activity_wall = _time.time()
success = _is_tool_success(result)
if success:
self._called_ok.add(tc.name)
@@ -2277,6 +2508,19 @@ class AgentLoop:
}
)
# Cache successful deterministic results so an identical later call is
# served without re-execution (regression: repeated financial_rigor
# calcs after compaction, 2026-08-20 INTC run).
if success:
try:
tool_def = self.registry.get(tc.name)
except Exception: # noqa: BLE001
tool_def = None
if tool_def is not None and getattr(tool_def, "deterministic", False):
cache_key = self._identical_call_key(tc.name, tc.arguments)
if cache_key is not None:
self._called_identical[cache_key] = result
status = "ok" if success else "error"
truncated = truncate_tool_result(result)
messages.append(context.format_tool_result(tc.id, tc.name, truncated))
@@ -2449,6 +2693,14 @@ class AgentLoop:
)
self._previous_summary = summary
# Preserve deterministic verification records across the compaction.
# The LLM summary does not reliably retain exact calc results, so the
# model used to re-run the same financial_rigor expressions after a
# compact to "re-verify" numbers it could no longer see (2026-08-20
# INTC run: identical calcs re-ran 5-9x each). Re-attach a terse
# ledger of already-verified values so the model does not re-run them.
verification_ledger = _verification_ledger(head)
tokens_before = estimate_tokens(messages)
trace.write_text_entry(
{
@@ -2467,6 +2719,12 @@ class AgentLoop:
# Reconstruct: system + summary + acknowledge + preserved tail
state_summary = self.memory.to_summary()
compressed = f"[Conversation compressed — handoff summary. Transcript: {transcript_path}]\n\n{summary}"
if verification_ledger:
compressed += (
"\n\n[Verified tool results from the compressed turns - do NOT "
"re-run these tools, they are already verified]:\n"
+ verification_ledger
)
if state_summary and state_summary != "(empty state)":
compressed += f"\n\nCurrent agent state:\n{state_summary}"
@@ -2500,6 +2758,7 @@ _LEGACY_LAZY = {
"STREAM_RETRY_DELAY_S": _stream_retry_delay_s,
"TOOL_TIMEOUT_SECONDS": _tool_timeout_seconds,
"GOAL_MAX_CONTINUATIONS": _goal_max_continuations,
"STALL_TIMEOUT_SECONDS": _stall_timeout_seconds,
}
agent/src/agent/tools.py
+4
View File
@@ -25,6 +25,10 @@ class BaseTool(ABC):
parameters: Dict[str, Any] = {}
repeatable: bool = False
is_readonly: bool = True
# Pure, side-effect-free computation: identical args always produce the
# same result, so the loop may serve repeated identical calls from cache
# instead of re-executing them (financial_rigor calc/verify, etc.).
deterministic: bool = False
@classmethod
def check_available(cls) -> bool:
agent/src/config/env_schema.py
+3
View File
@@ -357,6 +357,9 @@ class AgentTuningConfig(_EnvBase):
vibe_trading_llm_timeout_seconds: float = Field(
alias="VIBE_TRADING_LLM_TIMEOUT_SECONDS", default=300.0,
)
vibe_trading_run_stall_timeout_seconds: float = Field(
alias="VIBE_TRADING_RUN_STALL_TIMEOUT_SECONDS", default=1800.0,
)
vibe_trading_goal_max_continuations: int = Field(
alias="VIBE_TRADING_GOAL_MAX_CONTINUATIONS", default=3,
)
agent/src/tools/bash_tool.py
+85 -16
View File
@@ -3,7 +3,9 @@
from __future__ import annotations
import json
import os
import re
import signal
import subprocess
from typing import Any
@@ -12,6 +14,47 @@ from src.tools._shell_safety import broad_python_kill_error
_OUTPUT_LIMIT = 50_000
_DEFAULT_TIMEOUT = 120
# How long to wait for output to drain after killing a timed-out process
# tree before giving up (the pipes are held by grandchildren).
_DRAIN_TIMEOUT = 5
def _kill_process_tree(proc: subprocess.Popen) -> None:
"""Kill a subprocess and its whole descendant tree.
On Windows, ``cmd.exe /c <command>`` spawns grandchildren (e.g. a piped
``findstr``) that inherit the stdout/stderr pipe write handles. Killing
only the shell leaves those grandchildren alive, so a later
``communicate()`` blocks forever waiting for EOF that never arrives -
the exact 2026-08-20 hang (bash call at 19:20:55 never returned, cmd.exe
+ findstr still alive 20 minutes later). ``taskkill /F /T`` kills the
tree; POSIX uses ``killpg`` on the process group.
Args:
proc: The Popen object whose tree should be killed.
"""
if proc.poll() is not None:
return
if os.name == "nt":
try:
subprocess.run(
["taskkill", "/F", "/T", "/PID", str(proc.pid)],
capture_output=True,
timeout=10,
)
except Exception:
try:
proc.kill()
except Exception:
pass
else:
try:
os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
except Exception:
try:
proc.kill()
except Exception:
pass
class BashTool(BaseTool):
@@ -73,33 +116,59 @@ class BashTool(BaseTool):
ensure_ascii=False,
)
# Spawn with a dedicated process group so a timeout can kill the
# WHOLE tree, not just cmd.exe. Without this, grandchildren that
# inherited the pipe handles keep communicate() blocked forever.
creationflags = 0
start_new_session = False
if os.name == "nt":
creationflags = subprocess.CREATE_NEW_PROCESS_GROUP
else:
start_new_session = True
try:
result = subprocess.run(
proc = subprocess.Popen(
command,
shell=True,
cwd=cwd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
timeout=_DEFAULT_TIMEOUT,
encoding="utf-8",
errors="replace",
creationflags=creationflags,
start_new_session=start_new_session,
)
stdout = result.stdout[:_OUTPUT_LIMIT] if len(result.stdout) > _OUTPUT_LIMIT else result.stdout
stderr = result.stderr[:_OUTPUT_LIMIT] if len(result.stderr) > _OUTPUT_LIMIT else result.stderr
return json.dumps({
"status": "ok" if result.returncode == 0 else "error",
"exit_code": result.returncode,
"stdout": stdout,
"stderr": stderr,
}, ensure_ascii=False)
except subprocess.TimeoutExpired:
return json.dumps({
"status": "error",
"error": f"Command timed out after {_DEFAULT_TIMEOUT}s",
}, ensure_ascii=False)
except Exception as exc:
return json.dumps({
"status": "error",
"error": str(exc),
"error": f"Could not start command: {exc}",
}, ensure_ascii=False)
try:
stdout, stderr = proc.communicate(timeout=_DEFAULT_TIMEOUT)
except subprocess.TimeoutExpired:
# Kill the tree, then drain with a bounded second wait. The tree
# kill closes the inherited pipe handles, so the drain returns.
_kill_process_tree(proc)
try:
stdout, stderr = proc.communicate(timeout=_DRAIN_TIMEOUT)
except Exception:
stdout, stderr = "", ""
stdout = (stdout or "")[:_OUTPUT_LIMIT]
stderr = (stderr or "")[:_OUTPUT_LIMIT]
return json.dumps({
"status": "error",
"error_code": "tool_timeout",
"error": f"Command timed out after {_DEFAULT_TIMEOUT}s",
"stdout": stdout,
"stderr": stderr,
}, ensure_ascii=False)
stdout = stdout[:_OUTPUT_LIMIT] if len(stdout) > _OUTPUT_LIMIT else stdout
stderr = stderr[:_OUTPUT_LIMIT] if len(stderr) > _OUTPUT_LIMIT else stderr
return json.dumps({
"status": "ok" if proc.returncode == 0 else "error",
"exit_code": proc.returncode,
"stdout": stdout,
"stderr": stderr,
}, ensure_ascii=False)
agent/src/tools/financial_rigor_tool.py
+1
View File
@@ -512,6 +512,7 @@ class FinancialRigorTool(BaseTool):
"required": ["command"],
}
is_readonly = True
deterministic = True # pure computation; identical args -> identical result
repeatable = True # loop.py dedups non-repeatable tools by name; users call
# different sub-commands / params in one session.
agent/tests/test_bash_tool.py
+86 -10
View File
@@ -3,12 +3,54 @@
from __future__ import annotations
import json
import subprocess
from unittest.mock import MagicMock
import pytest
from src.tools import bash_tool
from src.tools.bash_tool import BashTool
from src.tools.bash_tool import BashTool, _kill_process_tree
class _FakePopen:
"""Minimal Popen stand-in for the allowed-command path."""
def __init__(self, returncode: int = 0, stdout: str = "ok", stderr: str = "") -> None:
self.returncode = returncode
self._stdout = stdout
self._stderr = stderr
self.killed = False
def communicate(self, timeout: float | None = None):
del timeout
return self._stdout, self._stderr
def kill(self) -> None:
self.killed = True
def poll(self):
return self.returncode
class _HangingPopen:
"""Popen stand-in that always times out on communicate."""
def __init__(self) -> None:
self.returncode = None
self.killed = False
def communicate(self, timeout: float | None = None):
raise subprocess.TimeoutExpired(cmd="hang", timeout=timeout)
def kill(self) -> None:
self.killed = True
def poll(self):
return None
def _install_fake_popen(monkeypatch: pytest.MonkeyPatch, popen: _FakePopen) -> None:
monkeypatch.setattr(bash_tool.subprocess, "Popen", lambda *a, **k: popen)
@pytest.mark.parametrize(
@@ -32,14 +74,14 @@ def test_rejects_broad_python_process_termination(
command: str,
monkeypatch: pytest.MonkeyPatch,
) -> None:
run = MagicMock(side_effect=AssertionError("unsafe command reached subprocess"))
monkeypatch.setattr(bash_tool.subprocess, "run", run)
popen = MagicMock(side_effect=AssertionError("unsafe command reached subprocess"))
monkeypatch.setattr(bash_tool.subprocess, "Popen", popen)
result = json.loads(BashTool().execute(command=command))
assert result["status"] == "error"
assert "cancel_background" in result["error"]
run.assert_not_called()
popen.assert_not_called()
@pytest.mark.parametrize(
@@ -54,13 +96,47 @@ def test_allows_non_broad_process_commands(
command: str,
monkeypatch: pytest.MonkeyPatch,
) -> None:
run = MagicMock()
run.return_value.returncode = 0
run.return_value.stdout = "ok"
run.return_value.stderr = ""
monkeypatch.setattr(bash_tool.subprocess, "run", run)
fake = _FakePopen()
_install_fake_popen(monkeypatch, fake)
result = json.loads(BashTool().execute(command=command))
assert result["status"] == "ok"
run.assert_called_once()
assert result["exit_code"] == 0
def test_timeout_kills_process_tree(monkeypatch: pytest.MonkeyPatch) -> None:
"""A timed-out command must kill the tree and report tool_timeout.
Regression for the 2026-08-20 hang: on Windows, killing only cmd.exe
leaves grandchildren (e.g. piped findstr) holding the stdout/stderr
pipe handles, so a bare subprocess.run timeout never returns. The tool
must spawn a dedicated process group and kill the WHOLE tree, then
return a bounded tool_timeout error.
"""
hanging = _HangingPopen()
_install_fake_popen(monkeypatch, hanging)
kills: list[int] = []
def _fake_kill(proc) -> None:
kills.append(proc.pid if hasattr(proc, "pid") else 0)
monkeypatch.setattr(bash_tool, "_kill_process_tree", _fake_kill)
result = json.loads(BashTool().execute(command="echo hang"))
assert result["status"] == "error"
assert result["error_code"] == "tool_timeout"
assert "timed out" in result["error"]
assert len(kills) == 1
def test_heredoc_unsupported(monkeypatch: pytest.MonkeyPatch) -> None:
popen = MagicMock(side_effect=AssertionError("heredoc reached subprocess"))
monkeypatch.setattr(bash_tool.subprocess, "Popen", popen)
result = json.loads(BashTool().execute(command="cat <<EOF\nhello\nEOF"))
assert result["status"] == "error"
assert result["error_code"] == "heredoc_unsupported"
popen.assert_not_called()
agent/tests/test_loop_helpers.py
+67
View File
@@ -19,6 +19,8 @@ from src.agent.loop import (
_normalize_tool_run_dir,
_archive_backtest_result,
_llm_timeout_seconds,
_stall_timeout_seconds,
_verification_ledger,
)
@@ -531,3 +533,68 @@ def test_pending_write_directive_tracks_written_targets(tmp_path: Path) -> None:
assert "old_plan.md" in agent2._pending_write_directive(
f"please update {old}", now - 10,
)
# ---------------------------------------------------------------------------
# _verification_ledger
# ---------------------------------------------------------------------------
def test_verification_ledger_extracts_calc_results() -> None:
"""A successful financial_rigor calc result becomes a terse ledger line."""
messages = [
{"role": "tool", "name": "financial_rigor", "content": json.dumps({
"status": "ok", "command": "calc",
"expr": "92.13/101.65-1", "result": -0.0937, "result_exact": "-0.09365912",
})},
{"role": "tool", "name": "financial_rigor", "content": json.dumps({
"status": "ok", "command": "calc",
"expr": "85.4/108.8-1", "result": -0.2151, "result_exact": "-0.21507353",
})},
]
from src.agent.loop import _verification_ledger
ledger = _verification_ledger(messages)
assert "calc 92.13/101.65-1 = -0.09365912" in ledger
assert "calc 85.4/108.8-1 = -0.21507353" in ledger
def test_verification_ledger_skips_errors_and_other_tools() -> None:
"""Failed results, non-financial_rigor tools, and non-JSON are skipped."""
messages = [
{"role": "tool", "name": "financial_rigor", "content": json.dumps({
"status": "error", "command": "calc", "error": "malformed",
})},
{"role": "tool", "name": "get_market_data", "content": json.dumps({"status": "ok"})},
{"role": "tool", "name": "financial_rigor", "content": "not json at all"},
{"role": "user", "content": "hello"},
]
from src.agent.loop import _verification_ledger
assert _verification_ledger(messages) == ""
def test_verification_ledger_deduplicates_and_caps() -> None:
"""Duplicate ledger lines collapse; the ledger is capped."""
messages = [
{"role": "tool", "name": "financial_rigor", "content": json.dumps({
"status": "ok", "command": "calc",
"expr": "a/b", "result": 1.0, "result_exact": "1.0",
})},
{"role": "tool", "name": "financial_rigor", "content": json.dumps({
"status": "ok", "command": "calc",
"expr": "a/b", "result": 1.0, "result_exact": "1.0",
})},
]
from src.agent.loop import _verification_ledger
ledger = _verification_ledger(messages)
assert ledger.count("calc a/b") == 1
def test_stall_timeout_seconds_default_and_override(monkeypatch) -> None:
"""The stall watchdog timeout reads config and honors a module override."""
import src.agent.loop as loop_module
assert _stall_timeout_seconds() > 0
monkeypatch.setattr(loop_module, "STALL_TIMEOUT_SECONDS", 42.0, raising=False)
assert _stall_timeout_seconds() == 42.0
monkeypatch.delattr(loop_module, "STALL_TIMEOUT_SECONDS", raising=False)
assert _stall_timeout_seconds() > 0