diff --git a/.env.example b/.env.example index ffa9799..2dcd3e7 100644 --- a/.env.example +++ b/.env.example @@ -24,15 +24,25 @@ CASE_LOCAL=1 # Per-turn input-token ceiling for Drive chat. Default 2000000. # CASE_TURN_TOKENS=2000000 +# OpenAI context compaction threshold, in tokens. 0 = off. Default 200000. +# CASE_COMPACT_AT=200000 + # Desktop resolution for new/woken computers (WxH or WxHxDEPTH). # DESK_RESOLUTION=1280x800x24 # Browser Drive still takes the key per request (x-openai-key / x-anthropic-key). -# The box key below is only for phone chat (Telegram or ntfy). +# The box key below is used by phone chat and schedules. # CASE_DRIVE_PROVIDER=openai # CASE_DRIVE_API_KEY= # CASE_DRIVE_MODEL= +# Schedule brain. CASE_BRAIN_CMD (a command template that contains {prompt}) +# wins if set. Otherwise compose POSTs to CASE_BRAIN_URL, which uses the box +# key above. Outside compose, unset both and the scheduler looks for `claude` +# on PATH. +# CASE_BRAIN_URL= +# CASE_BRAIN_CMD= + # Phone chat over Telegram: make a bot with @BotFather, paste its token, start # the ui, send /start to the bot and it replies with the chat id to put here. # CASE_TELEGRAM_TOKEN= diff --git a/README.md b/README.md index 12628a7..ead692a 100644 --- a/README.md +++ b/README.md @@ -75,7 +75,9 @@ in Drive. Cursor: to help through Drive, Telegram, or an Assist link. - **Skills:** the agent saves a completed task as a `SKILL.md` on the computer and follows it next time. The file survives restarts. -- **Schedules:** recurring runs use the computer's saved identity. +- **Schedules:** recurring runs use the computer's saved identity. They use + `CASE_DRIVE_API_KEY` from `.env` (the same box key as phone chat). Set + `CASE_BRAIN_CMD` if you want a different harness. - **Phone chat:** send tasks and answer handoffs through Telegram or ntfy. See [phone setup](#phone-chat). @@ -117,9 +119,10 @@ Drive connects out to the service, so phone chat works without exposing a local port. Your host and Docker must stay running to receive messages and run tasks. Phone tasks use a shared thread named `Phone`. They need a provider key on the -server because there is no browser tab to supply one. If you do not have a `.env` -file yet, copy `.env.example` to `.env` next to `compose.yaml`. Add these settings -to that file, replacing the placeholder with your key: +server because there is no browser tab to supply one. Schedules use the same +key. If you do not have a `.env` file yet, copy `.env.example` to `.env` next to +`compose.yaml`. Add these settings to that file, replacing the placeholder with +your key: ```dotenv CASE_DRIVE_PROVIDER=openai # or anthropic diff --git a/compose.yaml b/compose.yaml index 09e7e47..683a98b 100644 --- a/compose.yaml +++ b/compose.yaml @@ -41,6 +41,8 @@ services: CASE_BIND: "0.0.0.0" CASE_IMAGE: ${CASE_IMAGE:-case-desk:0.1} CASE_TOKEN: ${CASE_TOKEN:-} + CASE_BRAIN_URL: ${CASE_BRAIN_URL:-http://ui:4174/api/brain} + CASE_BRAIN_CMD: ${CASE_BRAIN_CMD:-} CASE_MAX_RUNNING: ${CASE_MAX_RUNNING:-4} # Total RAM the awake desktops may hold. Unset = 75% of what the engine sees, # which on a Mac is the Docker VM, not the Mac. @@ -121,6 +123,8 @@ services: CASE_DRIVE_PROVIDER: ${CASE_DRIVE_PROVIDER:-} CASE_DRIVE_API_KEY: ${CASE_DRIVE_API_KEY:-} CASE_DRIVE_MODEL: ${CASE_DRIVE_MODEL:-} + CASE_TURN_TOKENS: ${CASE_TURN_TOKENS:-} + CASE_COMPACT_AT: ${CASE_COMPACT_AT:-} CASE_TELEGRAM_TOKEN: ${CASE_TELEGRAM_TOKEN:-} CASE_TELEGRAM_CHAT_ID: ${CASE_TELEGRAM_CHAT_ID:-} CASE_PUBLIC_HOST: ${CASE_PUBLIC_HOST:-} diff --git a/control-plane/config.py b/control-plane/config.py index 7c4eefa..51d9797 100644 --- a/control-plane/config.py +++ b/control-plane/config.py @@ -67,6 +67,9 @@ def _desk_resolution(): # contain {prompt}. Unset = stock claude. The template carries no --allowedTools clamp: # a template harness runs with the box's full privileges, only use one you trust. BRAIN_CMD = os.environ.get("CASE_BRAIN_CMD", "") +# HTTP brain (compose default): POST {computer_id, prompt} to Drive. Used only +# when BRAIN_CMD is empty. Precedence: CASE_BRAIN_CMD > CASE_BRAIN_URL > claude. +BRAIN_URL = os.environ.get("CASE_BRAIN_URL", "") MCP_CONFIG = os.environ.get( "CASE_MCP_CONFIG", os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "case-mcp.json")) diff --git a/control-plane/scheduler.py b/control-plane/scheduler.py index 01a0a24..7abbabe 100644 --- a/control-plane/scheduler.py +++ b/control-plane/scheduler.py @@ -16,8 +16,11 @@ import subprocess import threading from datetime import datetime, timedelta, timezone +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError -from config import (BRAIN_BIN, BRAIN_CMD, BRAIN_TIMEOUT, MAX_RUNNING, MCP_CONFIG, +import requests + +from config import (BRAIN_BIN, BRAIN_CMD, BRAIN_TIMEOUT, BRAIN_URL, MAX_RUNNING, MCP_CONFIG, RUNS_DIR, log) from deskclient import desk_json, screenshot_bytes from errors import ApiError @@ -31,20 +34,47 @@ _LOCK = threading.Lock() # guards the check-then-add on SCHED_RUNNING (sweeper vs run-now) -def compute_next(kind, spec, jitter_s): - """Next fire time as UTC ISO. Lexicographic order == chronological (zero-padded, Z).""" +def _wall_exists(t, hh, mm): + """False when t is a spring-forward gap (that clock time never happened).""" + back = t.astimezone(timezone.utc).astimezone(t.tzinfo) + return (back.hour, back.minute) == (hh, mm) + + +def compute_next(kind, spec, jitter_s, tz=None, *, now=None): + """Next fire time as UTC ISO. Lexicographic order == chronological (zero-padded, Z). + Daily HH:MM is wall clock in tz (IANA). Empty tz = box local (MCP / old callers). + `now` overrides the clock for tests; for daily it is the wall clock in that zone.""" j = random.randint(0, int(jitter_s or 0)) if kind == "interval": if int(spec) < 60: raise ApiError(400, "bad_request", "interval must be at least 60 seconds") - nxt = datetime.now(timezone.utc) + timedelta(seconds=int(spec) + j) + nxt = (now or datetime.now(timezone.utc)) + timedelta(seconds=int(spec) + j) elif kind == "daily": - local = datetime.now() + name = str(tz).strip() if tz else "" + if name: + try: + local = now or datetime.now(ZoneInfo(name)) + except (ZoneInfoNotFoundError, ValueError): + raise ApiError(400, "bad_tz", f"unknown timezone {name}") + else: + # Naive on purpose: astimezone() below then picks the offset of the fire + # date, so a schedule set before a DST switch still fires at HH:MM after it. + # datetime.now().astimezone() would freeze today's offset into tomorrow. + local = now or datetime.now() hh, mm = (int(x) for x in str(spec).split(":")) t = local.replace(hour=hh, minute=mm, second=0, microsecond=0) if t <= local: t += timedelta(days=1) - nxt = (t + timedelta(seconds=j)).astimezone(timezone.utc) # naive→aware picks that date's offset + # Spring-forward gap: this HH:MM never happened that day. replace() + # still builds the invalid wall time and astimezone() shifts it (02:30 + # → 03:30). Skip to the next day that actually has that clock time. + # Re-check after jitter: 01:45 + 45m can land in the same hole. + if not _wall_exists(t, hh, mm): + t += timedelta(days=1) + fired = t + timedelta(seconds=j) + if not _wall_exists(fired, fired.hour, fired.minute): + fired = t + timedelta(days=1) + timedelta(seconds=j) + nxt = fired.astimezone(timezone.utc) else: raise ApiError(400, "bad_kind", "kind must be 'interval' or 'daily'") return nxt.strftime("%Y-%m-%dT%H:%M:%SZ") @@ -67,8 +97,43 @@ def brain_argv(full_prompt): "--allowedTools", "mcp__case__*"] -def run_brain(cid, prompt): - """Invoke the headless brain against this computer via Case MCP. Returns (code, summary).""" +def _run_brain_url(cid, prompt, name=""): + """POST {computer_id, prompt, name} to Drive. Returns the same (code, summary) as + the argv path, clipped the same way. `name` only titles the Drive thread.""" + token = (os.environ.get("CASE_TOKEN") or "").strip() + headers = {"Authorization": f"Bearer {token}"} if token else {} + try: + r = requests.post(BRAIN_URL, json={"computer_id": cid, "prompt": prompt, "name": name}, + headers=headers, timeout=BRAIN_TIMEOUT) + except requests.ConnectionError: # before Timeout: ConnectTimeout is both + return 127, (f"schedule brain unreachable at {BRAIN_URL} — start the ui service " + "or set CASE_BRAIN_CMD") + except requests.Timeout: + return -1, "brain run timed out" + except requests.RequestException as e: + return 127, f"schedule brain request failed: {e}" + try: + body = r.json() if r.content else {} + except ValueError: + body = {} + if not isinstance(body, dict): + body = {} + if r.status_code == 503: + return 2, str(body.get("error") or "schedule brain unavailable") + if r.status_code == 401: + return 1, "schedule brain rejected the token — CASE_TOKEN must match between cased and ui" + if body.get("ok"): + return (0 if body.get("finished") else 3), str(body.get("text") or "").strip()[-800:] + if body.get("error"): + return 1, str(body["error"])[-800:] + return 1, (r.text or f"HTTP {r.status_code}")[-800:] + + +def run_brain(cid, prompt, name=""): + """Invoke the headless brain against this computer. Returns (code, summary). + Precedence: CASE_BRAIN_CMD > CASE_BRAIN_URL > stock claude on PATH.""" + if not BRAIN_CMD and BRAIN_URL: + return _run_brain_url(cid, prompt, name) try: argv = brain_argv(f"On Case computer {cid}: {prompt}") except ValueError as e: @@ -134,8 +199,10 @@ def run_schedule(sid): s = store.get_schedule(sid, enabled_only=True) if not s: return + # sqlite3.Row has no dict.get — index like every other column. + tz = s["tz"] if "tz" in s.keys() else None # Reschedule FIRST so a hung/crashed run never wedges the slot. - store.set_schedule_next(sid, compute_next(s["kind"], s["spec"], s["jitter_s"])) + store.set_schedule_next(sid, compute_next(s["kind"], s["spec"], s["jitter_s"], tz)) cid, rid, started = s["computer_id"], new_id("run"), now() code, summary, status, artifact = -1, "", "fail", None # Only the run that woke an asleep box may put it back, never borrow a live session @@ -145,7 +212,7 @@ def run_schedule(sid): was_asleep = get_computer(cid)["state"] == "asleep" do_wake(cid) woke_for_run = was_asleep - code, summary = run_brain(cid, s["prompt"]) + code, summary = run_brain(cid, s["prompt"], name=s["name"]) status = "ok" if code == 0 else "fail" artifact = capture_run_artifacts(cid, rid, s["name"], status, summary, started) # awake except ApiError as e: @@ -154,6 +221,9 @@ def run_schedule(sid): if e.code == "too_many_running": status = "skipped" summary = f"another computer is running (max {MAX_RUNNING} on this box)" + elif e.code == "not_enough_ram": + status = "skipped" + summary = f"not enough free RAM on this box ({e.message})" else: summary = f"{e.code}: {e.message}" log.exception("schedule %s run failed", sid) @@ -182,9 +252,11 @@ def fire_due_schedules(spawn): def schedule_json(row): - return {k: row[k] for k in ("id", "computer_id", "name", "prompt", "kind", "spec", - "jitter_s", "enabled", "next_run_at", "last_run_at", - "last_status", "created_at")} + out = {k: row[k] for k in ("id", "computer_id", "name", "prompt", "kind", "spec", + "jitter_s", "enabled", "next_run_at", "last_run_at", + "last_status", "created_at")} + out["tz"] = row["tz"] if "tz" in row.keys() else None + return out def create_schedule(cid, body): @@ -192,16 +264,17 @@ def create_schedule(cid, body): if "prompt" not in body or "spec" not in body: raise ApiError(400, "bad_request", "prompt and spec are required") kind = body.get("kind", "daily") + tz = str(body.get("tz") or "").strip() or None try: jitter = int(body.get("jitter_s", 300)) - nxt = compute_next(kind, body["spec"], jitter) # also validates kind/spec + nxt = compute_next(kind, body["spec"], jitter, tz) # also validates kind/spec/tz except (TypeError, ValueError): raise ApiError(400, "bad_request", "spec must be seconds (interval) or HH:MM (daily); " "jitter_s must be an integer") sid = new_id("sch") store.insert_schedule(sid, cid, str(body.get("name") or sid), body["prompt"], - kind, str(body["spec"]), jitter, nxt) + kind, str(body["spec"]), jitter, nxt, tz) return schedule_json(store.get_schedule(sid)) diff --git a/control-plane/store.py b/control-plane/store.py index 302beec..c8e4525 100644 --- a/control-plane/store.py +++ b/control-plane/store.py @@ -66,7 +66,8 @@ CREATE TABLE IF NOT EXISTS schedules ( id TEXT PRIMARY KEY, computer_id TEXT, name TEXT, prompt TEXT, kind TEXT, spec TEXT, jitter_s INTEGER, enabled INTEGER, - next_run_at TEXT, last_run_at TEXT, last_status TEXT, created_at TEXT + next_run_at TEXT, last_run_at TEXT, last_status TEXT, created_at TEXT, + tz TEXT ); CREATE TABLE IF NOT EXISTS runs ( id TEXT PRIMARY KEY, schedule_id TEXT, computer_id TEXT, @@ -131,6 +132,7 @@ def __init__(self, home=None): ("credentials", "probe_url", "TEXT"), ("credentials", "proof_spec", "TEXT"), ("credentials", "verification_hosts", "TEXT"), + ("schedules", "tz", "TEXT"), ] # Active (non-terminal) auth-attempt statuses, kept here so the partial unique @@ -541,10 +543,10 @@ def prune_expired_assist_tokens(self): (ts, ts)).rowcount # ---- schedules ---- - def insert_schedule(self, sid, cid, name, prompt, kind, spec, jitter_s, next_run_at): + def insert_schedule(self, sid, cid, name, prompt, kind, spec, jitter_s, next_run_at, tz=None): self.q("INSERT INTO schedules (id,computer_id,name,prompt,kind,spec,jitter_s,enabled," - "next_run_at,last_run_at,last_status,created_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", - (sid, cid, name, prompt, kind, spec, jitter_s, 1, next_run_at, None, None, now())) + "next_run_at,last_run_at,last_status,created_at,tz) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", + (sid, cid, name, prompt, kind, spec, jitter_s, 1, next_run_at, None, None, now(), tz)) def get_schedule(self, sid, enabled_only=False): if enabled_only: diff --git a/mcp/case_mcp.py b/mcp/case_mcp.py index 5256ffa..f646712 100644 --- a/mcp/case_mcp.py +++ b/mcp/case_mcp.py @@ -601,12 +601,16 @@ def handoff_get(handoff_id: str) -> dict: if os.environ.get("CASE_MCP_SCHEDULES") == "1": @mcp.tool() def schedule_create(computer_id: str, prompt: str, kind: str = "daily", - spec: str = "09:00", name: str = "", jitter_s: int = 300) -> dict: - """Create a recurring schedule on a computer. kind=daily (spec HH:MM local) or - interval (spec seconds as string). Fires unattended using the host brain credential.""" + spec: str = "09:00", name: str = "", jitter_s: int = 300, + tz: str = "") -> dict: + """Create a recurring schedule on a computer. kind=daily (spec HH:MM in tz, + or box local if tz is empty) or interval (spec seconds as string). Fires + unattended using the host brain credential.""" body = {"prompt": prompt, "kind": kind, "spec": spec, "jitter_s": jitter_s} if name: body["name"] = name + if tz: + body["tz"] = tz return call("POST", f"/computers/{computer_id}/schedules", json=body).json() diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index 00abcb7..fcd9d95 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -44,6 +44,98 @@ def test_daily_fires_at_requested_local_time(): assert f"{local.hour:02d}:{local.minute:02d}" == spec, (spec, local.isoformat()) +def test_daily_kolkata_is_0330_utc(): + nxt = _dt(compute_next("daily", "09:00", 0, "Asia/Kolkata")) + assert nxt.hour == 3 and nxt.minute == 30, nxt.isoformat() + + +def test_daily_box_local_survives_dst_switch(): + # Set the night before clocks go back (London, 2026-10-25 01:00 UTC). 09:00 local + # tomorrow is 09:00Z, not the 08:00Z a frozen BST offset would give. + import time + old = os.environ.get("TZ") + os.environ["TZ"] = "Europe/London" + time.tzset() + try: + nxt = compute_next("daily", "09:00", 0, now=datetime(2026, 10, 24, 23, 0)) + assert nxt == "2026-10-25T09:00:00Z", nxt + finally: + if old is None: + del os.environ["TZ"] + else: + os.environ["TZ"] = old + time.tzset() + + +def test_daily_tz_survives_dst_switch(): + from zoneinfo import ZoneInfo + now = datetime(2026, 10, 24, 23, 0, tzinfo=ZoneInfo("Europe/London")) + nxt = compute_next("daily", "09:00", 0, "Europe/London", now=now) + assert nxt == "2026-10-25T09:00:00Z", nxt + + +def test_daily_skips_spring_forward_gap(): + # America/New_York 2026-03-08: 02:00 → 03:00, so 02:30 never happens. + # Fire the next day at 02:30 EDT (06:30Z), not 03:30 that morning. + from zoneinfo import ZoneInfo + now = datetime(2026, 3, 8, 0, 30, tzinfo=ZoneInfo("America/New_York")) + nxt = compute_next("daily", "02:30", 0, "America/New_York", now=now) + assert nxt == "2026-03-09T06:30:00Z", nxt + + +def test_daily_jitter_does_not_land_in_spring_forward_gap(): + # 01:45 + 45m = 02:30, which does not exist on 2026-03-08 in New York. + # Skip to the next day, keep the same jitter: 2026-03-09 02:30 EDT = 06:30Z. + from zoneinfo import ZoneInfo + import unittest.mock as mock + now = datetime(2026, 3, 8, 0, 30, tzinfo=ZoneInfo("America/New_York")) + with mock.patch("scheduler.random.randint", return_value=2700): + nxt = compute_next("daily", "01:45", 2700, "America/New_York", now=now) + assert nxt == "2026-03-09T06:30:00Z", nxt + + +def test_daily_box_local_skips_spring_forward_gap(): + import time + old = os.environ.get("TZ") + os.environ["TZ"] = "America/New_York" + time.tzset() + try: + nxt = compute_next("daily", "02:30", 0, now=datetime(2026, 3, 8, 0, 30)) + assert nxt == "2026-03-09T06:30:00Z", nxt + finally: + if old is None: + del os.environ["TZ"] + else: + os.environ["TZ"] = old + time.tzset() + + +def test_bad_tz_raises(): + from errors import ApiError + try: + compute_next("daily", "09:00", 0, "Not/AZone") + assert False, "expected bad_tz" + except ApiError as e: + assert e.code == "bad_tz", e + + +def test_sqlite_row_has_no_get_but_tz_index_works(): + store.q("DELETE FROM schedules") + store.insert_schedule("sch_tz", "c_1", "n", "p", "daily", "09:00", 0, + "2026-08-30T03:30:00Z", "Asia/Kolkata") + s = store.get_schedule("sch_tz") + assert not hasattr(s, "get"), type(s) + tz = s["tz"] if "tz" in s.keys() else None + assert tz == "Asia/Kolkata", tz + nxt = compute_next(s["kind"], s["spec"], s["jitter_s"], tz) + assert nxt[11:16] == "03:30", nxt + + +def test_schedules_tz_column_exists(): + cols = [r["name"] for r in store.db.execute("PRAGMA table_info(schedules)")] + assert "tz" in cols + + def test_jitter_stays_bounded(): base = datetime.now(timezone.utc) for _ in range(20): @@ -237,7 +329,7 @@ def active_attempt_exists(self, cid): scheduler.get_computer = lambda cid: {"id": cid, "state": "running"} scheduler.do_wake = lambda cid: None scheduler.do_sleep = lambda cid: slept.append(cid) - scheduler.run_brain = lambda cid, p: (0, "done") + scheduler.run_brain = lambda cid, p, name="": (0, "done") scheduler.capture_run_artifacts = lambda *a, **k: None scheduler.notifier = type("N", (), {"push": lambda self, m: None})() scheduler.emit = lambda *a, **k: None @@ -276,7 +368,7 @@ def active_attempt_exists(self, cid): scheduler.get_computer = lambda cid: {"id": cid, "state": "asleep"} scheduler.do_wake = lambda cid: None scheduler.do_sleep = lambda cid: slept.append(cid) - scheduler.run_brain = lambda cid, p: (0, "done") + scheduler.run_brain = lambda cid, p, name="": (0, "done") scheduler.capture_run_artifacts = lambda *a, **k: None scheduler.notifier = type("N", (), {"push": lambda self, m: None})() scheduler.emit = lambda *a, **k: None @@ -293,6 +385,177 @@ def active_attempt_exists(self, cid): assert rec.get("status") == "ok", rec +def test_ram_tight_box_is_a_skip_too(): + import scheduler + from errors import ApiError + rec = {} + + class _Store: + def get_schedule(self, sid, enabled_only=False): + return {"id": sid, "computer_id": "c_1", "name": "nightly", "prompt": "go", + "kind": "interval", "spec": "3600", "jitter_s": 0} + def set_schedule_next(self, *a): pass + def insert_run(self, rid, sid, cid, started, ended, code, summary, artifact, status): + rec["summary"] = summary + def set_schedule_result(self, sid, at, status): + rec["status"] = status + + def _tight(cid): + raise ApiError(409, "not_enough_ram", "3072 MB in use of 4096") + + old = (scheduler.store, scheduler.do_wake, scheduler.do_sleep, scheduler.get_computer, + scheduler.notifier, scheduler.emit) + try: + scheduler.store = _Store() + scheduler.get_computer = lambda cid: {"id": cid, "state": "asleep"} + scheduler.do_wake = _tight + scheduler.do_sleep = lambda cid: None + scheduler.notifier = type("N", (), {"push": lambda self, m: None})() + scheduler.emit = lambda *a, **k: None + scheduler.run_schedule("sch_x") + finally: + (scheduler.store, scheduler.do_wake, scheduler.do_sleep, scheduler.get_computer, + scheduler.notifier, scheduler.emit) = old + assert rec["status"] == "skipped", rec + assert "not enough free RAM" in rec["summary"], rec + assert "ApiError" not in rec["summary"], rec + + +def test_run_brain_url_finished(): + import unittest.mock as mock + import scheduler + old_cmd, old_url = scheduler.BRAIN_CMD, scheduler.BRAIN_URL + try: + scheduler.BRAIN_CMD = "" + scheduler.BRAIN_URL = "http://ui:4174/api/brain" + resp = mock.Mock(status_code=200, content=b'{"ok":true}', text="ok") + resp.json.return_value = {"ok": True, "finished": True, "text": "done"} + with mock.patch("scheduler.requests.post", return_value=resp) as post: + code, text = scheduler.run_brain("c_1", "hello", name="nightly") + assert (code, text) == (0, "done") + assert post.call_args.args[0] == "http://ui:4174/api/brain" + assert post.call_args.kwargs["json"] == {"computer_id": "c_1", "prompt": "hello", + "name": "nightly"} + # same clip as the argv path, so runs.summary and the logbook stay bounded + resp.json.return_value = {"ok": True, "finished": True, "text": "x" * 5000 + "END"} + with mock.patch("scheduler.requests.post", return_value=resp): + _, text = scheduler.run_brain("c_1", "hello") + assert len(text) == 800 and text.endswith("END"), len(text) + finally: + scheduler.BRAIN_CMD, scheduler.BRAIN_URL = old_cmd, old_url + + +def test_run_brain_url_unfinished(): + import unittest.mock as mock + import scheduler + old_cmd, old_url = scheduler.BRAIN_CMD, scheduler.BRAIN_URL + try: + scheduler.BRAIN_CMD = "" + scheduler.BRAIN_URL = "http://ui:4174/api/brain" + resp = mock.Mock(status_code=200, content=b'{"ok":true}', text="") + resp.json.return_value = {"ok": True, "finished": False, "text": "stopped mid-task"} + with mock.patch("scheduler.requests.post", return_value=resp): + code, text = scheduler.run_brain("c_1", "hello") + assert code == 3 and text == "stopped mid-task" + finally: + scheduler.BRAIN_CMD, scheduler.BRAIN_URL = old_cmd, old_url + + +def test_run_brain_url_503(): + import unittest.mock as mock + import scheduler + old_cmd, old_url = scheduler.BRAIN_CMD, scheduler.BRAIN_URL + try: + scheduler.BRAIN_CMD = "" + scheduler.BRAIN_URL = "http://ui:4174/api/brain" + resp = mock.Mock(status_code=503, content=b'{"error":"no key"}', text="") + resp.json.return_value = {"error": "set CASE_DRIVE_API_KEY in .env"} + with mock.patch("scheduler.requests.post", return_value=resp): + code, text = scheduler.run_brain("c_1", "hello") + assert code == 2 + assert "CASE_DRIVE_API_KEY" in text + finally: + scheduler.BRAIN_CMD, scheduler.BRAIN_URL = old_cmd, old_url + + +def test_run_brain_url_connection_error(): + import unittest.mock as mock + import scheduler + old_cmd, old_url = scheduler.BRAIN_CMD, scheduler.BRAIN_URL + try: + scheduler.BRAIN_CMD = "" + scheduler.BRAIN_URL = "http://ui:4174/api/brain" + # ConnectTimeout subclasses both ConnectionError and Timeout: it is "unreachable" + for exc in (scheduler.requests.ConnectionError(), scheduler.requests.ConnectTimeout()): + with mock.patch("scheduler.requests.post", side_effect=exc): + code, text = scheduler.run_brain("c_1", "hello") + assert code == 127, (exc, code) + assert "http://ui:4174/api/brain" in text and "CASE_BRAIN_CMD" in text, text + finally: + scheduler.BRAIN_CMD, scheduler.BRAIN_URL = old_cmd, old_url + + +def test_run_brain_url_sends_bearer_only_when_token_set(): + import unittest.mock as mock + import scheduler + old_cmd, old_url = scheduler.BRAIN_CMD, scheduler.BRAIN_URL + old_tok = os.environ.get("CASE_TOKEN") + try: + scheduler.BRAIN_CMD = "" + scheduler.BRAIN_URL = "http://ui:4174/api/brain" + resp = mock.Mock(status_code=200, content=b"{}", text="") + resp.json.return_value = {"ok": True, "finished": True, "text": ""} + os.environ["CASE_TOKEN"] = "tok" + with mock.patch("scheduler.requests.post", return_value=resp) as post: + scheduler.run_brain("c_1", "hello") + assert post.call_args.kwargs["headers"] == {"Authorization": "Bearer tok"} + os.environ["CASE_TOKEN"] = "" + with mock.patch("scheduler.requests.post", return_value=resp) as post: + scheduler.run_brain("c_1", "hello") + assert post.call_args.kwargs["headers"] == {} + resp401 = mock.Mock(status_code=401, content=b'{"error":"unauthorized"}', text="") + resp401.json.return_value = {"error": "unauthorized"} + with mock.patch("scheduler.requests.post", return_value=resp401): + code, text = scheduler.run_brain("c_1", "hello") + assert code == 1 and "CASE_TOKEN" in text, (code, text) + finally: + scheduler.BRAIN_CMD, scheduler.BRAIN_URL = old_cmd, old_url + if old_tok is None: + os.environ.pop("CASE_TOKEN", None) + else: + os.environ["CASE_TOKEN"] = old_tok + + +def test_run_brain_url_timeout(): + import unittest.mock as mock + import scheduler + old_cmd, old_url = scheduler.BRAIN_CMD, scheduler.BRAIN_URL + try: + scheduler.BRAIN_CMD = "" + scheduler.BRAIN_URL = "http://ui:4174/api/brain" + with mock.patch("scheduler.requests.post", side_effect=scheduler.requests.Timeout()): + code, text = scheduler.run_brain("c_1", "hello") + assert code == -1 + assert "timed out" in text + finally: + scheduler.BRAIN_CMD, scheduler.BRAIN_URL = old_cmd, old_url + + +def test_run_brain_cmd_wins_over_url(): + import unittest.mock as mock + import scheduler + old_cmd, old_url = scheduler.BRAIN_CMD, scheduler.BRAIN_URL + try: + scheduler.BRAIN_CMD = "definitely-not-a-brain-bin {prompt}" + scheduler.BRAIN_URL = "http://ui:4174/api/brain" + with mock.patch("scheduler.requests.post") as post: + code, _ = scheduler.run_brain("c_1", "hello") + assert post.call_count == 0 + assert code == 127 + finally: + scheduler.BRAIN_CMD, scheduler.BRAIN_URL = old_cmd, old_url + + if __name__ == "__main__": for name, fn in sorted(globals().items()): if name.startswith("test_"): diff --git a/web/package.json b/web/package.json index 4840ec4..3a13020 100644 --- a/web/package.json +++ b/web/package.json @@ -4,7 +4,7 @@ "type": "module", "scripts": { "start": "node web-ui/serve.mjs", - "test": "node web-ui/test_serve.mjs && node web-ui/test_http.mjs && node web-ui/test_phone.mjs && node web-ui/test_ntfy.mjs && node web-ui/test_telegram.mjs && node web-ui/test_nav.mjs && node web-ui/test_deploy.mjs" + "test": "node web-ui/test_serve.mjs && node web-ui/test_rate_retry.mjs && node web-ui/test_http.mjs && node web-ui/test_phone.mjs && node web-ui/test_ntfy.mjs && node web-ui/test_telegram.mjs && node web-ui/test_nav.mjs && node web-ui/test_deploy.mjs" }, "dependencies": { "@anthropic-ai/sdk": "^0.117.1", diff --git a/web/web-ui/case-tools.mjs b/web/web-ui/case-tools.mjs index 6f09a26..c5a8a15 100644 --- a/web/web-ui/case-tools.mjs +++ b/web/web-ui/case-tools.mjs @@ -372,9 +372,29 @@ function clipJson(v, n = 8000) { return s.length > n ? s.slice(0, n) + '…' : s; } +/** Is this the provider saying "too fast" rather than "bad request"? Callers with + * their own error fallbacks (unsupported summary/effort) must ask first: a 429 + * misread as an unsupported-param error retries with no backoff and degrades the + * request for nothing. */ +export function isRateLimited(err) { + const status = err?.status ?? err?.response?.status; + return status === 429 || status === 529 + || /rate limit|overloaded/i.test(err?.message || ''); +} + +/** Seconds to wait before attempt `a`: the server's own hint ("try again in Xs" + * or retry-after) if it gave one, else exponential. Padded, clamped to 1..60s. */ +export function rateWaitS(err, a) { + const m = /try again in ([\d.]+)s/i.exec(err?.message || ''); + const hdr = Number(err?.headers?.['retry-after'] + ?? err?.response?.headers?.get?.('retry-after')); + const wait = m ? Number(m[1]) : Number.isFinite(hdr) && hdr > 0 ? hdr : 2 ** a; + return Math.min(Math.max(wait + 0.5, 1), 60); +} + /** Retry a provider round on rate limits (429/529), honoring the server's - * suggested wait ("try again in Xs" / retry-after), capped at 60s. History is - * only mutated after a round completes, so replaying a failed round is safe. */ + * suggested wait. History is only mutated after a round completes, so replaying + * a failed round is safe. `signal` cancels the backoff sleep on STOP. */ function abortError(signal) { if (signal?.reason instanceof Error) return signal.reason; const err = new Error(signal?.reason ? String(signal.reason) : 'stopped by user'); @@ -407,16 +427,10 @@ export async function withRateRetry(fn, emit, tries = 5, signal) { try { return await fn(); } catch (err) { if (signal?.aborted) throw err; - const status = err?.status ?? err?.response?.status; - const limited = status === 429 || status === 529 - || /rate limit|overloaded/i.test(err?.message || ''); - if (!limited || a >= tries - 1) throw err; - const m = /try again in ([\d.]+)s/i.exec(err?.message || ''); - const hdr = Number(err?.headers?.['retry-after'] - ?? err?.response?.headers?.get?.('retry-after')); - let wait = m ? Number(m[1]) : Number.isFinite(hdr) && hdr > 0 ? hdr : 2 ** a; - wait = Math.min(Math.max(wait + 0.5, 1), 60); - emit?.({ type: 'think', text: `rate limited — retrying in ${Math.ceil(wait)}s` }); + if (!isRateLimited(err) || a >= tries - 1) throw err; + const wait = rateWaitS(err, a); + // `rate: true` so a non-UI consumer can pick the wait out of the think stream. + emit?.({ type: 'think', rate: true, text: `rate limited — retrying in ${Math.ceil(wait)}s` }); await abortableDelay(wait * 1000, signal); } } @@ -479,7 +493,9 @@ export async function anthropicToolLoop({ result = await withRateRetry(() => round(params), emit, 5, signal); } catch (err) { if (signal?.aborted) throw err; - if (!params.output_config) throw err; + // This fallback is for models that reject output_config — not for a rate + // limit whose retries already ran dry, which would only buy 5 more waits. + if (!params.output_config || isRateLimited(err)) throw err; const rest = { ...params }; delete rest.output_config; result = await withRateRetry(() => round(rest), emit, 5, signal); diff --git a/web/web-ui/index.html b/web/web-ui/index.html index f54ce31..00c8e46 100644 --- a/web/web-ui/index.html +++ b/web/web-ui/index.html @@ -471,6 +471,34 @@ .modal-card input:focus{border-color:var(--accent)} .modal-actions{display:flex;gap:8px;margin-top:12px;justify-content:flex-end} .modal-head{display:flex;align-items:stretch;gap:8px;margin-bottom:14px} +.mbar{display:none;align-items:center;gap:10px;padding:9px 12px;background:var(--field);border-bottom:1px solid var(--soft)} +.mbar #schedBtnM{margin-left:auto} +.mbar #keyBtnM{margin-left:6px} +.sched-card{width:min(520px,100%)} +.sched-card h3{margin:0} +.sched-list{margin:0 0 16px;max-height:40vh;overflow:auto} +.sched-row{ + display:grid;grid-template-columns:1fr auto auto;gap:8px;align-items:start; + padding:10px 0;border-bottom:1px solid var(--soft); +} +.sched-row .sn{font-size:13px;font-weight:600} +.sched-row .sm{font-size:11px;color:var(--faint);margin-top:2px} +.sched-row .sp{font-size:12px;color:var(--mut);margin-top:4px;white-space:pre-wrap} +.sched-empty{padding:12px 0;font-size:13px;color:var(--faint)} +.sched-card label{display:block;font-size:11px;color:var(--faint);margin:10px 0 4px} +.sched-card textarea,.sched-card input[type=text],.sched-card input[type=time],.sched-card input[type=number]{ + width:100%;border:1px solid var(--soft);background:var(--field);padding:8px 10px; + font-family:var(--sans);font-size:13px;outline:none; +} +.sched-card textarea:focus,.sched-card input:focus{border-color:var(--accent);background:var(--paper)} +.sched-when{display:flex;flex-wrap:wrap;gap:10px;align-items:center;margin-top:8px;font-size:13px} +.sched-when label{margin:0;color:var(--ink);font-size:13px} +.sched-when input[type=time],.sched-when input[type=number]{width:auto;max-width:8rem} +.sched-when select{border:1px solid var(--soft);background:var(--field);padding:7px 8px;font-size:12px;max-width:16rem} +.sched-when.interval #schedTz,.sched-when.interval #schedDaily{opacity:.4;pointer-events:none} +.sched-err{font-size:12px;color:var(--warn);min-height:14px;margin-top:8px} +#schedFuel{font-size:12px;color:var(--warn);margin:0 0 10px} +#schedFuel[hidden]{display:none} .key-tabs{display:flex;flex:1;border:1px solid var(--ink);min-width:0} .key-tab{ flex:1;border:0;background:transparent;padding:8px; @@ -489,6 +517,7 @@ @media(max-width:900px){ body{grid-template-columns:1fr} .side,#sideSplit{display:none} + .mbar{display:flex} .stage{flex-direction:column} #railSplit{flex-basis:8px;cursor:row-resize} #railSplit::before{inset:3px 0} @@ -508,12 +537,14 @@
no computer
+
+
@@ -636,6 +667,35 @@
+ +