diff --git a/.claude-plugin/plugin.json b/.claude-plugin/plugin.json index a3eadaa..127fbc8 100644 --- a/.claude-plugin/plugin.json +++ b/.claude-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "model-router", - "version": "0.39.0", + "version": "0.40.0", "description": "Route authorized software work through one durable, capacity-aware subscription policy.", "author": { "name": "toolboxmd", diff --git a/.codex-plugin/plugin.json b/.codex-plugin/plugin.json index b420c16..1e28f73 100644 --- a/.codex-plugin/plugin.json +++ b/.codex-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "model-router", - "version": "0.39.0", + "version": "0.40.0", "description": "Route authorized software work through one durable, capacity-aware subscription policy.", "author": { "name": "toolboxmd", diff --git a/.grok-plugin/plugin.json b/.grok-plugin/plugin.json index a3eadaa..127fbc8 100644 --- a/.grok-plugin/plugin.json +++ b/.grok-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "model-router", - "version": "0.39.0", + "version": "0.40.0", "description": "Route authorized software work through one durable, capacity-aware subscription policy.", "author": { "name": "toolboxmd", diff --git a/CHANGELOG.md b/CHANGELOG.md index 6249bd1..6fbfd0a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## [0.40.0] - 2026-09-28 + +### Changed + +- A job waiting for T3 starts Chromeria after 60 seconds without an answer, and again every 5 minutes (`launching_t3` in `status`; `T3_LAUNCH_CMD` overrides the command, empty turns it off) (#140) + ## [0.39.0] - 2026-09-28 ### Changed diff --git a/RUNNER.md b/RUNNER.md index 8053c56..35b45cb 100644 --- a/RUNNER.md +++ b/RUNNER.md @@ -142,7 +142,14 @@ terminal report tells the planner in its thread (and in `status` and `T3_UNREACHABLE_WAIT_SECS` (24 hours). Until then the job waits for T3 (Chromeria quit, restarting, or installing a rebuild), rereading with backoff (5, 10, 20, 40, then 60 seconds). The wait begins with a - `waiting_for_t3` event in `status` carrying `since` and `cause`. While + `waiting_for_t3` event in `status` carrying `since` and `cause`. After + 60 seconds without an answer the job starts T3 itself, and again every + 5 minutes while it stays down, each launch a `launching_t3` event in + `status` carrying `at`, `command` and `result` (`ok` or the failure; a + failed launch never ends the wait). The command is `T3_LAUNCH_CMD`, + run without a shell with a 15 second timeout; it defaults to + `/usr/bin/open -g -a Chromeria` on macOS and to off elsewhere, and an + empty value turns launching off. While it waits no thread starts and the last snapshot's routes are not reused; a read that recovers continues the job, and a cancel ends the wait at the next reread. An empty list in a readable snapshot blocks at diff --git a/VERSION b/VERSION index 4ef2eb0..9b0025a 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.39.0 +0.40.0 diff --git a/runner/controller.py b/runner/controller.py index 75bee9e..415979c 100644 --- a/runner/controller.py +++ b/runner/controller.py @@ -14,7 +14,9 @@ import os import signal import secrets +import shlex import subprocess +import sys import time from pathlib import Path @@ -4182,18 +4184,48 @@ def _advertise(state_dir: str, request_id: str, token: str) -> None: T3_UNREACHABLE_WAIT_SECS = 24 * 3600.0 PRISM_RETRY_FIRST_SECS = 5.0 PRISM_RETRY_MAX_SECS = 60.0 +# After this long unreachable the job starts T3 itself, and again every +# T3_RELAUNCH_EVERY_SECS while it stays down (#140). T3_LAUNCH_CMD +# overrides the command; an empty value turns launching off. +T3_LAUNCH_AFTER_SECS = 60.0 +T3_RELAUNCH_EVERY_SECS = 300.0 +T3_LAUNCH_TIMEOUT_SECS = 15.0 +T3_LAUNCH_DEFAULT = "/usr/bin/open -g -a Chromeria" _clock = time.monotonic _sleep = time.sleep -def _record_t3_wait(state_dir, request_id: str, cause: str) -> None: - """Ledger event ``waiting_for_t3`` so ``status`` shows the wait.""" +def _t3_launch_command() -> str: + """The command that starts T3, or "" when launching is off.""" + command = os.environ.get("T3_LAUNCH_CMD") + if command is None: + command = T3_LAUNCH_DEFAULT if sys.platform == "darwin" else "" + return command.strip() + + +def _launch_t3(command: str) -> str: + """Run ``command`` once without a shell; "ok" or why it failed.""" + try: + done = subprocess.run(shlex.split(command), stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, + timeout=T3_LAUNCH_TIMEOUT_SECS, check=False) + except (OSError, ValueError, subprocess.SubprocessError) as e: + return f"{type(e).__name__}: {e}"[:300] + if done.returncode: + err = done.stderr.decode(errors="replace").strip() + return f"exit {done.returncode}: {err}"[:300] + return "ok" + + +def _record_t3_wait(state_dir, request_id: str, cause: str, + kind: str = "waiting_for_t3", payload: dict | None = None) -> None: + """Ledger event ``waiting_for_t3`` (or ``kind``) so ``status`` shows the wait.""" con = store.connect(state_dir) try: con.execute("BEGIN IMMEDIATE") _lease_guard(con, request_id) - core._event(con, request_id, "waiting_for_t3", - {"since": core._utcnow(), "cause": cause[:300]}) + core._event(con, request_id, kind, + payload or {"since": core._utcnow(), "cause": cause[:300]}) con.execute("COMMIT") except Exception: try: @@ -4210,13 +4242,16 @@ def _await_prism(state_dir, request_id: str, token: str | None) -> str | None: While T3 is unreachable the snapshot is reread with backoff for up to ``T3_UNREACHABLE_WAIT_SECS``; no stage has routes meanwhile, so - nothing launches and the last snapshot is never reused. A cancel - ends the wait at the next reread. Any other unreadable snapshot, or - an empty list the job needs, blocks at once. + nothing launches and the last snapshot is never reused. After + ``T3_LAUNCH_AFTER_SECS`` it starts T3 itself, and again every + ``T3_RELAUNCH_EVERY_SECS``; a failed launch is recorded and the + wait goes on. A cancel ends the wait at the next reread. Any other + unreadable snapshot, or an empty list the job needs, blocks at once. """ job = core.get_job(state_dir, request_id) start = _clock() delay = PRISM_RETRY_FIRST_SECS + launched = None snap = t3snapshot.refresh_for_job(job) if snap is None and t3snapshot.unreachable(): _record_t3_wait(state_dir, request_id, t3snapshot._CACHE.get("error") or "") @@ -4226,6 +4261,14 @@ def _await_prism(state_dir, request_id: str, token: str | None) -> str | None: hours = int(T3_UNREACHABLE_WAIT_SECS // 3600) return (f"Prism unreadable: T3 unreachable for {hours}h " f"({t3snapshot._CACHE.get('error') or 'no answer'})") + command = _t3_launch_command() + if command and waited >= T3_LAUNCH_AFTER_SECS and ( + launched is None or waited - launched >= T3_RELAUNCH_EVERY_SECS): + launched = waited + result = _launch_t3(command) + _record_t3_wait(state_dir, request_id, "", "launching_t3", + {"at": core._utcnow(), "command": command[:300], + "result": result}) _sleep(min(delay, T3_UNREACHABLE_WAIT_SECS - waited)) delay = min(delay * 2, PRISM_RETRY_MAX_SECS) if core.get_job(state_dir, request_id)["cancel_requested"]: diff --git a/tests/fakes.py b/tests/fakes.py index 801ed9f..dc6c9c4 100644 --- a/tests/fakes.py +++ b/tests/fakes.py @@ -24,10 +24,12 @@ def isolate_t3_env(): Some tests reach T3 discovery without a fake client; inside a runner job ``T3_SERVER_URL``/``T3_SERVER_TOKEN`` name the live server, and without them the default URL and ``t3 auth`` do. Point both at a - closed port for the module and restore them afterwards. + closed port for the module and restore them afterwards, and turn off + starting T3 (#140). """ patcher = mock.patch.dict("os.environ", {"T3_SERVER_URL": "http://127.0.0.1:9", - "T3_SERVER_TOKEN": "test-token"}) + "T3_SERVER_TOKEN": "test-token", + "T3_LAUNCH_CMD": ""}) patcher.start() unittest.addModuleCleanup(patcher.stop) diff --git a/tests/test_issue133.py b/tests/test_issue133.py index 465ca4c..8aef643 100644 --- a/tests/test_issue133.py +++ b/tests/test_issue133.py @@ -44,7 +44,8 @@ def run_controller(self, rid, prism, reads=None, step=None): ``prism``; returns the fake. ``reads`` scripts the first snapshot reads (None fails like an unreachable server, an exception is raised as is); ``step`` replaces - the controller step. Waits run on a fake clock (``self.slept``).""" + the controller step. Waits run on a fake clock (``self.slept``); + T3 launches go to a fake (``self.launched``).""" fake = use_fake_t3(self, self.sd, rid) fake.prism = prism if reads is not None: @@ -72,6 +73,15 @@ def sleep(secs): patcher = mock.patch.object(controller, name, value) patcher.start() self.addCleanup(patcher.stop) + # Never start the real T3 (#140); record what would launch. + self.launched = [] + + def launch(command): + self.launched.append((now[0], command)) + return self.launch_result(command) if hasattr(self, "launch_result") else "ok" + patcher = mock.patch.object(controller, "_launch_t3", launch) + patcher.start() + self.addCleanup(patcher.stop) if step is not None: patcher = mock.patch.object(controller, "step", step) patcher.start() diff --git a/tests/test_issue140.py b/tests/test_issue140.py new file mode 100644 index 0000000..26489fd --- /dev/null +++ b/tests/test_issue140.py @@ -0,0 +1,123 @@ +"""Start T3 when it stays unreachable (toolboxmd/model-router#140). + +After 60 continuous seconds without an answer the job runs the launch +command once, then again every 5 minutes while T3 stays down. A failed +launch is recorded and the wait goes on; an empty ``T3_LAUNCH_CMD`` +turns launching off. +""" +import json +import sys +import unittest +from pathlib import Path +from unittest import mock + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from runner import controller, core # noqa: E402 +from tests.test_issue133 import Base # noqa: E402 +from tests.fakes import default_prism, isolate_t3_env # noqa: E402 + + +def setUpModule(): + isolate_t3_env() + + +def launch_events(sd, rid): + return [json.loads(e["payload_json"]) for e in core.status_view(sd, rid)["recent_events"] + if e["kind"] == "launching_t3"] + + +def stop_step(steps): + def step(state_dir, request_id, token=None): + steps.append(1) + return {"action": "blocked", "reason": "test_stop"} + return step + + +class Launch(Base): + def setUp(self): + super().setUp() + patcher = mock.patch.dict("os.environ", {"T3_LAUNCH_CMD": "start-t3 --quiet"}) + patcher.start() + self.addCleanup(patcher.stop) + + def test_no_launch_before_60_seconds(self): + self.submit() + steps = [] + # Rereads at 5, 15, 35 seconds; back at 35. + self.run_controller("j", default_prism(), reads=[None] * 4, step=stop_step(steps)) + self.assertEqual(self.slept, [5.0, 10.0, 20.0, 40.0]) + self.assertEqual(self.launched, []) + self.assertEqual(launch_events(self.sd, "j"), []) + self.assertEqual(steps, [1]) + + def test_launch_after_60_seconds_then_continue_once_back(self): + self.submit() + steps = [] + self.run_controller("j", default_prism(), reads=[None] * 6, step=stop_step(steps)) + # Unreachable at 0, 5, 15, 35, 75: one launch at 75, back at 135. + self.assertEqual(self.launched, [(75.0, "start-t3 --quiet")]) + self.assertEqual(steps, [1]) + self.assertNotIn("Prism", core.get_job(self.sd, "j")["block_reason"] or "") + events = launch_events(self.sd, "j") + self.assertEqual(len(events), 1) + self.assertEqual(events[0]["command"], "start-t3 --quiet") + self.assertEqual(events[0]["result"], "ok") + self.assertTrue(events[0]["at"]) + + def test_relaunch_every_5_minutes_while_down(self): + self.submit() + steps = [] + self.run_controller("j", default_prism(), reads=[None] * 20, step=stop_step(steps)) + times = [t for t, _ in self.launched] + self.assertEqual(times, [75.0, 375.0, 675.0, 975.0]) + self.assertEqual(steps, [1]) + + def test_a_failed_launch_is_recorded_and_the_wait_goes_on(self): + self.submit() + steps = [] + self.launch_result = lambda command: "FileNotFoundError: start-t3" + self.run_controller("j", default_prism(), reads=[None] * 12, step=stop_step(steps)) + self.assertEqual([t for t, _ in self.launched], [75.0, 375.0]) + self.assertEqual([e["result"] for e in launch_events(self.sd, "j")], + ["FileNotFoundError: start-t3"] * 2) + self.assertEqual(steps, [1]) + self.assertNotIn("Prism", core.get_job(self.sd, "j")["block_reason"] or "") + + def test_empty_command_turns_launching_off(self): + self.submit() + steps = [] + with mock.patch.dict("os.environ", {"T3_LAUNCH_CMD": " "}): + self.run_controller("j", default_prism(), reads=[None] * 20, step=stop_step(steps)) + self.assertEqual(self.launched, []) + self.assertEqual(launch_events(self.sd, "j"), []) + self.assertEqual(steps, [1]) + + +class Command(unittest.TestCase): + def test_default_is_chromeria_on_macos_and_off_elsewhere(self): + with mock.patch.dict("os.environ", {}, clear=True): + with mock.patch.object(controller.sys, "platform", "darwin"): + self.assertEqual(controller._t3_launch_command(), "/usr/bin/open -g -a Chromeria") + with mock.patch.object(controller.sys, "platform", "linux"): + self.assertEqual(controller._t3_launch_command(), "") + with mock.patch.dict("os.environ", {"T3_LAUNCH_CMD": "x y"}): + self.assertEqual(controller._t3_launch_command(), "x y") + + def test_launch_runs_without_a_shell_and_reports_failures(self): + self.assertEqual(controller._launch_t3(f"{sys.executable} -c pass"), "ok") + self.assertTrue(controller._launch_t3( + f"{sys.executable} -c 'import sys; sys.exit(3)'").startswith("exit 3")) + self.assertTrue(controller._launch_t3("/nonexistent/start-t3").startswith( + "FileNotFoundError")) + self.assertTrue(controller._launch_t3("'unbalanced").startswith("ValueError")) + # No shell: a metacharacter is an argument, not a command separator. + self.assertEqual(controller._launch_t3(f"{sys.executable} -c pass ; false"), "ok") + with mock.patch.object(controller, "T3_LAUNCH_TIMEOUT_SECS", 0.2): + self.assertTrue(controller._launch_t3( + f"{sys.executable} -c 'import time; time.sleep(5)'").startswith("TimeoutExpired")) + + +if __name__ == "__main__": + unittest.main()