diff --git a/solstone/think/runner.py b/solstone/think/runner.py index efa01c2c3..e4bef7e97 100644 --- a/solstone/think/runner.py +++ b/solstone/think/runner.py @@ -33,6 +33,8 @@ from solstone.think.callosum import CallosumConnection from solstone.think.utils import CHRONICLE_DIR, get_journal, now_ms logger = logging.getLogger(__name__) +# Default wall-clock budget (30m) for a task run with no explicit cap. +DEFAULT_TASK_MAX_RUNTIME = 1800 KILL_REAP_GRACE_S = 0.5 @@ -529,7 +531,7 @@ def run_task( ref: str | None = None, callosum: CallosumConnection | None = None, day: str | None = None, -) -> tuple[bool, int, Path]: +) -> tuple[bool, int, Path, bool]: """Run a task to completion with automatic logging (blocking). Spawns process, waits for completion, cleans up resources. @@ -546,18 +548,19 @@ def run_task( in that day's health directory instead of today's. Returns: - (success, exit_code, log_path) tuple where success = (exit_code == 0) - and log_path points to the process output log file. + (success, exit_code, log_path, timed_out) tuple where success = + (exit_code == 0), log_path points to the process output log file, and + timed_out is True only when the wall-clock ``timeout`` was exceeded. Example: - success, code, log = run_task( + success, code, log, timed_out = run_task( ["sol", "generate", "20241101", "-f", "flow"], timeout=300, ) # Logs to: {JOURNAL}/{YYYYMMDD}/health/{ref}_generate.log # With explicit correlation ID: - success, code, log = run_task( + success, code, log, timed_out = run_task( ["journal", "indexer", "--rescan"], ref="1730476800000", ) @@ -565,10 +568,12 @@ def run_task( """ managed = ManagedProcess.spawn(cmd, env=env, ref=ref, callosum=callosum, day=day) log_path = managed.log_writer.path + timed_out = False try: exit_code = managed.wait(timeout=timeout) except subprocess.TimeoutExpired: logger.error(f"{managed.name} timed out after {timeout}s, terminating...") + timed_out = True exit_code = managed.terminate() finally: managed.cleanup() @@ -576,4 +581,4 @@ def run_task( if exit_code != 0: logger.warning(f"{managed.name} exited with code {exit_code}") - return (exit_code == 0, exit_code, log_path) + return (exit_code == 0, exit_code, log_path, timed_out) diff --git a/solstone/think/supervisor.py b/solstone/think/supervisor.py index 9e8d7cd56..9290b5a21 100644 --- a/solstone/think/supervisor.py +++ b/solstone/think/supervisor.py @@ -52,8 +52,8 @@ from solstone.think.processing import ( ) from solstone.think.providers.mlx_server import MLX_SERVER_PROCESS_NAME from solstone.think.readiness import START_TIME_TOLERANCE_S, clear_ready, signal_ready +from solstone.think.runner import DEFAULT_TASK_MAX_RUNTIME, _command_partition from solstone.think.runner import ManagedProcess as RunnerManagedProcess -from solstone.think.runner import _command_partition from solstone.think.sync_check import ( DEFAULT_INTERVAL_SECONDS, SyncCheckSnapshot, @@ -80,9 +80,6 @@ from solstone.think.utils import ( write_service_port, ) -DEFAULT_TASK_MAX_RUNTIME = ( - 1800 # 30m wall-clock default for any uncapped TaskQueue partition -) REACTIVE_TASK_CAPS = { "daily": 21600, # 6h daily/from-scratch reprocess "segment": 4500, # 75m per-segment think diff --git a/solstone/think/thinking.py b/solstone/think/thinking.py index aa44a90ba..bbb4f6e18 100644 --- a/solstone/think/thinking.py +++ b/solstone/think/thinking.py @@ -14,6 +14,7 @@ import argparse import fnmatch import json import logging +import subprocess import sys import threading import time @@ -53,7 +54,7 @@ from solstone.think.pipeline_health import ( read_daily_deterministic_failures, read_segment_progress, ) -from solstone.think.runner import run_task +from solstone.think.runner import DEFAULT_TASK_MAX_RUNTIME, run_task from solstone.think.sense_splitter import ( write_change_detection, write_idle_stubs, @@ -459,25 +460,42 @@ def _emit_periodic_status() -> None: logging.debug("Status emission failed", exc_info=True) -def run_command(cmd: list[str], day: str) -> bool: - """Run a shell command synchronously.""" +def run_bounded_phase( + cmd: list[str], day: str, timeout: float | None +) -> tuple[bool, bool]: + """Run a phase subprocess. + + Returns (ok, timed_out). timed_out is True only when the wall-clock budget + was exceeded (covers the terminate-re-raise edge). + """ logging.info("==> %s", " ".join(cmd)) cmd_name = cmd[1] if cmd[0] in ("sol", "journal") and len(cmd) > 1 else cmd[0] cmd_name = cmd_name.replace("-", "_") try: - success, exit_code, _log_path = run_task(cmd, day=day) + success, exit_code, _log_path, timed_out = run_task( + cmd, day=day, timeout=timeout + ) if not success: logging.error( "Command failed with exit code %s: %s", exit_code, " ".join(cmd) ) day_log(day, f"{cmd_name} error {exit_code}") - return False - return True + return (success, timed_out) + except subprocess.TimeoutExpired: + logging.error("Command timed out and could not be reaped: %s", " ".join(cmd)) + day_log(day, f"{cmd_name} timeout") + return (False, True) except Exception as e: logging.error("Command exception: %s: %s", e, " ".join(cmd)) day_log(day, f"{cmd_name} exception") - return False + return (False, False) + + +def run_command(cmd: list[str], day: str) -> bool: + """Run a shell command synchronously (unbounded).""" + ok, _timed_out = run_bounded_phase(cmd, day, timeout=None) + return ok def run_queued_command(cmd: list[str], day: str, timeout: int = 600) -> bool: @@ -3734,17 +3752,31 @@ def main() -> None: day_log(day, f"starting: {' '.join(cmd)}") _jsonl_log("phase.start", mode=_run_mode, day=day, phase="segment_think") _phase_start = time.time() - phase_ok = run_command(cmd, day) - _jsonl_log( - "phase.complete", - mode=_run_mode, - day=day, - phase="segment_think", - success=phase_ok, - duration_ms=int((time.time() - _phase_start) * 1000), + phase_ok, phase_timed_out = run_bounded_phase( + cmd, day, DEFAULT_TASK_MAX_RUNTIME ) + phase_complete = { + "mode": _run_mode, + "day": day, + "phase": "segment_think", + "success": phase_ok, + "duration_ms": int((time.time() - _phase_start) * 1000), + } + if phase_timed_out: + phase_complete.update( + reason_code="wall_clock_exceeded", + timeout_seconds=DEFAULT_TASK_MAX_RUNTIME, + bounded=True, + ) + _jsonl_log("phase.complete", **phase_complete) if not phase_ok: - logging.warning("Segment-think repair failed, continuing anyway") + if phase_timed_out: + logging.warning( + "Segment-think repair exceeded its %ss budget, continuing anyway", + DEFAULT_TASK_MAX_RUNTIME, + ) + else: + logging.warning("Segment-think repair failed, continuing anyway") # MAIN PHASE: Run prompts resolved_stream = args.stream diff --git a/tests/test_runner.py b/tests/test_runner.py index 2c1b4e191..d85a490b0 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -275,12 +275,13 @@ def test_run_task_emits_logs_tract_events(journal_path, mock_callosum): listener.start(callback=lambda msg: received.append(msg)) # Run task - success, exit_code, log_path = run_task(["echo", "run_task test"]) + success, exit_code, log_path, timed_out = run_task(["echo", "run_task test"]) # Verify success assert success is True assert exit_code == 0 assert log_path.exists() + assert timed_out is False # Verify events were emitted logs_events = [msg for msg in received if msg.get("tract") == "logs"] @@ -435,12 +436,15 @@ def test_process_day_override(journal_path, mock_callosum): def test_run_task_day_override(journal_path, mock_callosum): """Test that run_task passes day through to log placement.""" target_day = "20240201" - success, exit_code, log_path = run_task(["echo", "task day test"], day=target_day) + success, exit_code, log_path, timed_out = run_task( + ["echo", "task day test"], day=target_day + ) assert success assert exit_code == 0 assert target_day in str(log_path) assert log_path.exists() + assert timed_out is False @pytest.mark.parametrize( diff --git a/tests/test_think_full.py b/tests/test_think_full.py index 4af10ac87..3e64052fe 100644 --- a/tests/test_think_full.py +++ b/tests/test_think_full.py @@ -18,6 +18,10 @@ def test_main_runs_with_mocked_prompts(journal_copy, monkeypatch): commands_run.append(cmd) return True + def mock_run_bounded_phase(cmd, day, timeout=None): + commands_run.append(cmd) + return (True, False) + def mock_run_queued_command(cmd, day, timeout=600): commands_run.append(cmd) return True @@ -28,6 +32,7 @@ def test_main_runs_with_mocked_prompts(journal_copy, monkeypatch): return (5, 0, [], set()) # 5 success, 0 failures, no failed names monkeypatch.setattr(mod, "run_command", mock_run_command) + monkeypatch.setattr(mod, "run_bounded_phase", mock_run_bounded_phase) monkeypatch.setattr(mod, "run_queued_command", mock_run_queued_command) monkeypatch.setattr(mod, "run_daily_prompts", mock_run_daily_prompts) monkeypatch.setattr( @@ -62,6 +67,10 @@ def test_main_runs_segment_think_prephase_before_daily_synthesis( commands_run.append(cmd) return True + def mock_run_bounded_phase(cmd, day, timeout=None): + commands_run.append(cmd) + return (True, False) + def mock_run_queued_command(cmd, day, timeout=600): commands_run.append(cmd) return True @@ -71,6 +80,7 @@ def test_main_runs_segment_think_prephase_before_daily_synthesis( return (5, 0, [], set()) monkeypatch.setattr(mod, "run_command", mock_run_command) + monkeypatch.setattr(mod, "run_bounded_phase", mock_run_bounded_phase) monkeypatch.setattr(mod, "run_queued_command", mock_run_queued_command) monkeypatch.setattr(mod, "run_daily_prompts", mock_run_daily_prompts) monkeypatch.setattr( @@ -116,6 +126,10 @@ def test_from_scratch_refreshes_segment_think_prephase( commands_run.append(cmd) return True + def mock_run_bounded_phase(cmd, day, timeout=None): + commands_run.append(cmd) + return (True, False) + def mock_run_queued_command(cmd, day, timeout=600): commands_run.append(cmd) return True @@ -125,6 +139,7 @@ def test_from_scratch_refreshes_segment_think_prephase( return (5, 0, [], set()) monkeypatch.setattr(mod, "run_command", mock_run_command) + monkeypatch.setattr(mod, "run_bounded_phase", mock_run_bounded_phase) monkeypatch.setattr(mod, "run_queued_command", mock_run_queued_command) monkeypatch.setattr(mod, "run_daily_prompts", mock_run_daily_prompts) monkeypatch.setattr( @@ -155,10 +170,12 @@ def test_segment_think_prephase_failure_is_non_fatal( def mock_run_command(cmd, day): commands_run.append(cmd) - if len(cmd) >= 2 and cmd[:2] == ["journal", "think"] and "--segments" in cmd: - return False return True + def mock_run_bounded_phase(cmd, day, timeout=None): + commands_run.append(cmd) + return (False, False) + def mock_run_queued_command(cmd, day, timeout=600): commands_run.append(cmd) return True @@ -168,6 +185,7 @@ def test_segment_think_prephase_failure_is_non_fatal( return (5, 0, [], set()) monkeypatch.setattr(mod, "run_command", mock_run_command) + monkeypatch.setattr(mod, "run_bounded_phase", mock_run_bounded_phase) monkeypatch.setattr(mod, "run_queued_command", mock_run_queued_command) monkeypatch.setattr(mod, "run_daily_prompts", mock_run_daily_prompts) monkeypatch.setattr( diff --git a/tests/test_think_segment_prephase.py b/tests/test_think_segment_prephase.py index 3fed43c84..157642acb 100644 --- a/tests/test_think_segment_prephase.py +++ b/tests/test_think_segment_prephase.py @@ -133,6 +133,9 @@ def test_daily_health_log_keeps_segment_events_out(journal_copy, monkeypatch): _patch_main_runtime(monkeypatch) monkeypatch.setattr(mod, "run_command", mock_run_command) + monkeypatch.setattr( + mod, "run_bounded_phase", lambda cmd, day, timeout=None: (True, False) + ) monkeypatch.setattr(mod, "run_queued_command", mock_run_queued_command) monkeypatch.setattr(mod, "run_daily_prompts", mock_run_daily_prompts) monkeypatch.setattr("sys.argv", ["sol think", "--day", "20240101"]) @@ -156,6 +159,104 @@ def test_daily_health_log_keeps_segment_events_out(journal_copy, monkeypatch): ] +def test_daily_segment_prephase_timeout_is_nonfatal(journal_copy, monkeypatch): + mod = importlib.import_module("solstone.think.thinking") + bounded_calls = [] + command_calls = [] + daily_called = [] + + def fake_bounded(cmd, day, timeout=None): + bounded_calls.append((cmd, day, timeout)) + return (False, True) + + def fake_command(cmd, day): + command_calls.append(cmd) + return True + + def fake_daily(day, verbose, **kwargs): + daily_called.append(day) + return (5, 0, [], set()) + + _patch_main_runtime(monkeypatch) + monkeypatch.setattr(mod, "run_bounded_phase", fake_bounded) + monkeypatch.setattr(mod, "run_command", fake_command) + monkeypatch.setattr(mod, "run_queued_command", lambda cmd, day, timeout=600: True) + monkeypatch.setattr(mod, "run_daily_prompts", fake_daily) + monkeypatch.setattr("sys.argv", ["sol think", "--day", "20240101"]) + + mod.main() + + health_dir = journal_copy / "chronicle" / "20240101" / "health" + daily_files = sorted(health_dir.glob("*_daily.jsonl")) + assert len(daily_files) == 1 + events = _read_jsonl(daily_files[0]) + segment_completes = [ + event + for event in events + if _event_name(event) == "phase.complete" + and event.get("phase") == "segment_think" + ] + assert len(segment_completes) == 1 + complete = segment_completes[0] + assert complete["success"] is False + assert complete["reason_code"] == "wall_clock_exceeded" + assert complete["timeout_seconds"] == 1800 + assert complete["bounded"] is True + + assert daily_called + assert bounded_calls == [ + ( + ["journal", "think", "--segments", "--day", "20240101"], + "20240101", + mod.DEFAULT_TASK_MAX_RUNTIME, + ) + ] + assert mod.DEFAULT_TASK_MAX_RUNTIME == 1800 + assert ["journal", "sense", "--day", "20240101"] in command_calls + assert ["journal", "journal-stats"] in command_calls + assert ["journal", "think", "--segments", "--day", "20240101"] not in command_calls + + +def test_daily_segment_prephase_failure_has_no_timeout_reason( + journal_copy, monkeypatch +): + mod = importlib.import_module("solstone.think.thinking") + daily_called = [] + + def fake_daily(day, verbose, **kwargs): + daily_called.append(day) + return (5, 0, [], set()) + + _patch_main_runtime(monkeypatch) + monkeypatch.setattr( + mod, "run_bounded_phase", lambda cmd, day, timeout=None: (False, False) + ) + monkeypatch.setattr(mod, "run_command", lambda cmd, day: True) + monkeypatch.setattr(mod, "run_queued_command", lambda cmd, day, timeout=600: True) + monkeypatch.setattr(mod, "run_daily_prompts", fake_daily) + monkeypatch.setattr("sys.argv", ["sol think", "--day", "20240101"]) + + mod.main() + + health_dir = journal_copy / "chronicle" / "20240101" / "health" + daily_files = sorted(health_dir.glob("*_daily.jsonl")) + assert len(daily_files) == 1 + events = _read_jsonl(daily_files[0]) + segment_completes = [ + event + for event in events + if _event_name(event) == "phase.complete" + and event.get("phase") == "segment_think" + ] + assert len(segment_completes) == 1 + complete = segment_completes[0] + assert complete["success"] is False + assert "reason_code" not in complete + assert "timeout_seconds" not in complete + assert "bounded" not in complete + assert daily_called + + def test_segment_health_log_receives_segment_talent_events(tmp_path, monkeypatch): from solstone.think import thinking as think