diff --git a/.aipass/tier1_navmap.md b/.aipass/tier1_navmap.md index a3e3d62a..8cfae959 100644 --- a/.aipass/tier1_navmap.md +++ b/.aipass/tier1_navmap.md @@ -65,6 +65,7 @@ drone @ai_mail inbox # check mail → view drone @flow create . "Subject" [dplan] # new plan (default FPLAN) drone @seedgo audit aipass @branch # standards audit (drop @branch = all) drone @seedgo checklist # quick standards check +drone @trigger medic mute @ # BEFORE build/edit work — auto-expires 24h drone @git status / diff / log # read-only git awareness drone @memory search "query" # recall archived context ``` diff --git a/CHANGELOG.md b/CHANGELOG.md index 0bab8d3d..fd55fdbf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,38 @@ PyPI version — not the changelog header. ### Added +- **Medic is back on — and the loop is proven live.** Off since 2026-05-10 (a + pytest fixture storm flooded the error registry; the off switch was pulled to + stop the noise and forgotten for 65 days). Three fixes made re-enable safe: + (1) @prax: pytest logging routes to a temp dir when `PYTEST_CURRENT_TEST` is + set — test fixtures can never pollute production `logs/` again (the storm + class that caused the shutdown); (2) @trigger: circuit breaker self-heals — + open breakers half-open on read, close on a successful probe, cooldown decays + to base (previously `half_open` was a terminal trap and only manual reset + recovered); (3) @trigger: **TTL mutes** — `medic mute @branch` and `medic off` + now auto-expire after 24h by default (`--for 48h/7d` custom, `--forever` + explicit kill switch; temp `off` keeps detection running). Agents doing build + work mute themselves and never have to remember to unmute — the permanent + switch that got medic forgotten no longer exists. Breadcrumbs shipped: ai_mail + footer + navmap tell every agent to mute before build work. Live-fire proof: + a planted commons SQL bug was detected, dispatched, and fixed byte-identical + by @commons in 105 seconds (15/15 tests green); a real TG poll error was + correctly triaged NOT ACTIONABLE; organic instance-lock noise was correctly + triaged LOW/expected. @skills/@api on 7-day mutes until the TG poll-level fix + lands. 993 prax + 603 trigger tests green. + +- **Prax monitor: concurrent viewers — laptop and Telegram mirror side by + side.** Patrick's ruling after being locked out of his own monitor three + times: *processes are not agents; display processes must never be + single-instance.* The instance lock is gone from the display path — any + number of `monitor run` viewers start and render concurrently. The lock is + scoped to the one true single-writer responsibility: the Telegram relay + (`relay.pid`, held by `prax-monitor.service`); extra instances run + viewer-only, so no TG double-sends. The misleading "kill the existing + process" error is dead. 998 prax tests green, 3 new concurrent-viewer tests; + live-verified: interactive Mission Control rendering while the TG relay + service runs untouched. + - **Telegram user-comment mirror: the TG chat now shows the whole conversation, whichever door you speak through.** Patrick's spec from the live cross-door drill: his own messages typed in the terminal or claude.ai remote never diff --git a/src/aipass/ai_mail/apps/handlers/email/footer.py b/src/aipass/ai_mail/apps/handlers/email/footer.py index 6263ac0c..4b3db78b 100644 --- a/src/aipass/ai_mail/apps/handlers/email/footer.py +++ b/src/aipass/ai_mail/apps/handlers/email/footer.py @@ -1,9 +1,9 @@ # =================== AIPass ==================== # Name: footer.py # Description: Email Footer Handler -# Version: 1.0.0 +# Version: 1.1.0 # Created: 2026-01-29 -# Modified: 2026-01-29 +# Modified: 2026-07-14 # ============================================= """ @@ -19,6 +19,8 @@ from aipass.ai_mail.apps.handlers.json import json_handler # Standard footer for all outgoing emails STANDARD_FOOTER = """ --- +▶ BUILD/EDIT task? First: drone @trigger medic mute @ — auto-expires 24h, no unmute. + ⚠️ TASK CHECKLIST (before marking complete): □ SEEDGO CHECK → drone @seedgo audit @branch (80%+) □ UPDATE MEMORIES → Your .trinity/local.json records this work diff --git a/src/aipass/prax/CLOSED_PLANS.local.json b/src/aipass/prax/CLOSED_PLANS.local.json index b8f76701..304a6cad 100644 --- a/src/aipass/prax/CLOSED_PLANS.local.json +++ b/src/aipass/prax/CLOSED_PLANS.local.json @@ -69,6 +69,20 @@ "subject": "STATUS decommission — prax dashboard todo_count + engine dormant", "date_closed": "2026-06-08", "location": "prax" + }, + { + "plan_id": "FPLAN-0321", + "type": "FPLAN", + "subject": "Pytest log routing: detect test context, route away from production logs", + "date_closed": "2026-07-14", + "location": "prax" + }, + { + "plan_id": "FPLAN-0323", + "type": "FPLAN", + "subject": "Remove single-instance lock from monitor display path — concurrent viewers", + "date_closed": "2026-07-14", + "location": "prax" } ], "document_metadata": { diff --git a/src/aipass/prax/apps/handlers/config/load.py b/src/aipass/prax/apps/handlers/config/load.py index 32df2313..18fce093 100755 --- a/src/aipass/prax/apps/handlers/config/load.py +++ b/src/aipass/prax/apps/handlers/config/load.py @@ -31,13 +31,14 @@ import inspect import json import logging import os - -logger = logging.getLogger(__name__) +import tempfile from pathlib import Path from typing import Dict, Any, Optional from aipass.prax.apps.handlers.json import json_handler +logger = logging.getLogger(__name__) + # ============================================= # CONFIGURATION # ============================================= @@ -75,6 +76,10 @@ def get_system_logs_dir() -> Path: p = Path(test_log_dir) / "system" p.mkdir(parents=True, exist_ok=True) return p + if os.environ.get("PYTEST_CURRENT_TEST"): + p = Path(tempfile.gettempdir()) / "aipass_test_logs" / "system" + p.mkdir(parents=True, exist_ok=True) + return p global _system_logs_dir_cache if _system_logs_dir_cache is None: repo_root = _find_repo_root() @@ -133,6 +138,10 @@ def get_module_logs_dir(module_name: Optional[str] = None) -> Path: p = Path(test_log_dir) / module_name p.mkdir(parents=True, exist_ok=True) return p + if os.environ.get("PYTEST_CURRENT_TEST"): + p = Path(tempfile.gettempdir()) / "aipass_test_logs" / module_name + p.mkdir(parents=True, exist_ok=True) + return p # Standard: src/aipass/{module}/logs branch_dir = ECOSYSTEM_ROOT / module_name diff --git a/src/aipass/prax/apps/handlers/monitoring/instance_lock.py b/src/aipass/prax/apps/handlers/monitoring/instance_lock.py index 7c66ddd8..a3a90a80 100644 --- a/src/aipass/prax/apps/handlers/monitoring/instance_lock.py +++ b/src/aipass/prax/apps/handlers/monitoring/instance_lock.py @@ -6,13 +6,14 @@ # Modified: 2026-07-10 # ============================================= -"""Single-instance lock for the prax monitor. +"""Relay-scoped lock for the prax monitor Telegram relay. -Prevents duplicate monitor processes from running concurrently (and -double-sending Telegram relay messages). Uses a pidfile with liveness -check — cross-platform (Linux / macOS / Windows). +Prevents duplicate Telegram sends when multiple monitor viewers run +concurrently. The display path is never blocked — only the TG relay +acquires this lock, so interactive viewers always start. -Lock file lives in prax_json/monitor.pid (outside system_logs/ to avoid +Uses a pidfile with liveness check — cross-platform (Linux / macOS / Windows). +Lock file lives in prax_json/relay.pid (outside system_logs/ to avoid the tailed-directory feedback loop). """ @@ -82,14 +83,14 @@ def _is_pid_alive(pid: int) -> bool: def get_lock_path() -> Path: - """Return the path for the monitor single-instance lock file.""" + """Return the path for the relay lock file.""" if _lock_path_override is not None: return _lock_path_override - return Path(__file__).resolve().parent.parent.parent / "prax_json" / "monitor.pid" + return Path(__file__).resolve().parent.parent.parent / "prax_json" / "relay.pid" -def acquire(error_fn=None) -> None: - """Acquire single-instance lock. Raises SystemExit(1) if another live instance holds it.""" +def try_acquire() -> bool: + """Try to acquire the relay lock. Returns True if acquired, False if held by a live process.""" global _held_lock lock_path = get_lock_path() json_handler.log_operation("instance_lock_acquire", {"pid": os.getpid()}) @@ -99,11 +100,8 @@ def acquire(error_fn=None) -> None: data = _json.loads(lock_path.read_text(encoding="utf-8")) existing_pid = data.get("pid", 0) if existing_pid and _is_pid_alive(existing_pid): - msg = f"Monitor already running (PID {existing_pid}). Kill the existing process or remove {lock_path}" - if error_fn: - error_fn(msg) - logger.error("[instance_lock] %s", msg) - raise SystemExit(1) + logger.info("[instance_lock] Relay lock held by PID %d — skipping TG relay", existing_pid) + return False logger.info("[instance_lock] Reclaiming stale lock (PID %d is dead)", existing_pid) except (ValueError, OSError) as exc: logger.info("[instance_lock] Removing corrupt lock file: %s", exc) @@ -111,7 +109,8 @@ def acquire(error_fn=None) -> None: lock_path.parent.mkdir(parents=True, exist_ok=True) lock_path.write_text(_json.dumps({"pid": os.getpid()}), encoding="utf-8") _held_lock = lock_path - logger.info("[instance_lock] Acquired (PID %d)", os.getpid()) + logger.info("[instance_lock] Acquired relay lock (PID %d)", os.getpid()) + return True def release() -> None: diff --git a/src/aipass/prax/apps/handlers/monitoring/telegram_relay.py b/src/aipass/prax/apps/handlers/monitoring/telegram_relay.py index 67372456..e09bd091 100644 --- a/src/aipass/prax/apps/handlers/monitoring/telegram_relay.py +++ b/src/aipass/prax/apps/handlers/monitoring/telegram_relay.py @@ -26,6 +26,7 @@ from urllib.request import urlopen as _http_fetch from aipass.prax.apps.modules.logger import get_direct_logger from aipass.prax.apps.handlers.json import json_handler +from aipass.prax.apps.handlers.monitoring import instance_lock logger = get_direct_logger() @@ -72,6 +73,10 @@ def init_relay(enabled: bool, config: Optional[dict] = None) -> None: logger.info("[telegram_relay] Incomplete config (missing bot_token or chat_id) — relay inactive") return + if not instance_lock.try_acquire(): + logger.info("[telegram_relay] Another process owns the TG relay — viewer-only mode") + return + _bot_token = token _chat_id = int(chat) _RELAY_ACTIVE = True @@ -111,6 +116,7 @@ def stop_relay() -> None: _thread.join(timeout=BATCH_INTERVAL + 2) _thread = None + instance_lock.release() json_handler.log_operation("relay_stopped", {}) logger.info("[telegram_relay] Relay stopped") diff --git a/src/aipass/prax/apps/modules/monitor.py b/src/aipass/prax/apps/modules/monitor.py index f70c27af..bcdb6d13 100755 --- a/src/aipass/prax/apps/modules/monitor.py +++ b/src/aipass/prax/apps/modules/monitor.py @@ -48,7 +48,6 @@ from aipass.prax.apps.handlers.monitoring.telegram_relay import ( is_relay_enabled_by_env, ) from aipass.prax.apps.handlers.monitoring.pid_cache import get_pid_for_branch as _get_pid_for_branch -from aipass.prax.apps.handlers.monitoring import instance_lock import json as _json @@ -177,8 +176,6 @@ def _run_monitor(args: List[str]) -> bool: global _event_queue, _module_tracker global _display_thread, _file_watcher_thread, _log_watcher_thread - instance_lock.acquire(error_fn=error) - json_handler.log_operation("monitor_started", {"args": args}) logger.info(f"Starting unified monitoring (args: {args})") @@ -258,7 +255,6 @@ def _stop_threads(): if t is not None and t.is_alive(): t.join(timeout=2.0) - instance_lock.release() logger.info("All monitoring threads stopped") diff --git a/src/aipass/prax/tests/test_config.py b/src/aipass/prax/tests/test_config.py index bab000d3..a56975c4 100644 --- a/src/aipass/prax/tests/test_config.py +++ b/src/aipass/prax/tests/test_config.py @@ -47,8 +47,9 @@ def _fresh_import_load(monkeypatch, tmp_path): monkeypatch.setattr(load_mod, "PRAX_LOGGER_CONFIG_FILE", prax_json_dir / "prax_logger_config.json") # Reset the lazy cache so get_system_logs_dir() re-resolves monkeypatch.setattr(load_mod, "_system_logs_dir_cache", None) - # Clear test log redirect so tests exercise real path resolution + # Clear test log redirects so tests exercise real path resolution monkeypatch.delenv("AIPASS_TEST_LOG_DIR", raising=False) + monkeypatch.delenv("PYTEST_CURRENT_TEST", raising=False) return load_mod @@ -181,6 +182,50 @@ class TestGetModuleLogsDir: assert not (tmp_path / "polyglot").exists() +# ============================================= +# TESTS: PYTEST_CURRENT_TEST routing +# ============================================= + + +class TestPytestCurrentTestRouting: + """PYTEST_CURRENT_TEST env var routes logs to temp dir, not production.""" + + def test_system_logs_routed_to_temp(self, mock_prax_infrastructure, monkeypatch, tmp_path): + load_mod = _fresh_import_load(monkeypatch, tmp_path) + monkeypatch.setenv("PYTEST_CURRENT_TEST", "tests/test_foo.py::test_bar (call)") + result = load_mod.get_system_logs_dir() + assert "aipass_test_logs" in str(result) + assert result.name == "system" + assert result.exists() + assert not (tmp_path / "system_logs").exists() + + def test_module_logs_routed_to_temp(self, mock_prax_infrastructure, monkeypatch, tmp_path): + load_mod = _fresh_import_load(monkeypatch, tmp_path) + monkeypatch.setenv("PYTEST_CURRENT_TEST", "tests/test_foo.py::test_bar (call)") + (tmp_path / "flow").mkdir() + result = load_mod.get_module_logs_dir("flow") + assert "aipass_test_logs" in str(result) + assert result.name == "flow" + assert result.exists() + assert not (tmp_path / "flow" / "logs").exists() + + def test_aipass_test_log_dir_takes_precedence(self, mock_prax_infrastructure, monkeypatch, tmp_path): + load_mod = _fresh_import_load(monkeypatch, tmp_path) + override = tmp_path / "custom_test_logs" + override.mkdir() + monkeypatch.setenv("AIPASS_TEST_LOG_DIR", str(override)) + monkeypatch.setenv("PYTEST_CURRENT_TEST", "tests/test_foo.py::test_bar (call)") + result = load_mod.get_system_logs_dir() + assert result == override / "system" + + def test_no_pytest_env_uses_production_path(self, mock_prax_infrastructure, monkeypatch, tmp_path): + load_mod = _fresh_import_load(monkeypatch, tmp_path) + monkeypatch.delenv("PYTEST_CURRENT_TEST", raising=False) + monkeypatch.setattr(load_mod, "_find_repo_root", lambda: tmp_path) + result = load_mod.get_system_logs_dir() + assert result == tmp_path / "system_logs" + + # ============================================= # TESTS: lines_to_bytes # ============================================= diff --git a/src/aipass/prax/tests/test_instance_lock.py b/src/aipass/prax/tests/test_instance_lock.py index 77ce1de3..44d18ccf 100644 --- a/src/aipass/prax/tests/test_instance_lock.py +++ b/src/aipass/prax/tests/test_instance_lock.py @@ -10,8 +10,9 @@ Covers: - _is_pid_alive() cross-platform liveness check -- acquire() creates lock, refuses live duplicate, reclaims stale +- try_acquire() creates lock, returns False for live duplicate, reclaims stale - release() removes lock file on clean shutdown +- Concurrent viewer: relay lock scoped to TG sends, never blocks display """ import json @@ -94,86 +95,78 @@ class TestIsPidAlive: assert mod._is_pid_alive(1234) is True -class TestAcquire: - """Test single-instance lock acquisition.""" +class TestTryAcquire: + """Test relay lock acquisition.""" def test_creates_lock_file(self, tmp_path): - """acquire() creates a lock file with the current PID.""" + """try_acquire() creates a lock file with the current PID.""" mod = _import_lock() - lock_path = tmp_path / "monitor.pid" + lock_path = tmp_path / "relay.pid" setattr(mod, "_lock_path_override", lock_path) - mod.acquire() + assert mod.try_acquire() is True assert lock_path.exists() data = json.loads(lock_path.read_text(encoding="utf-8")) assert data["pid"] == os.getpid() - def test_refuses_when_live_instance_holds_lock(self, tmp_path): - """acquire() exits with SystemExit(1) when another live process holds the lock.""" + def test_returns_false_when_live_holder(self, tmp_path): + """try_acquire() returns False when another live process holds the lock.""" mod = _import_lock() - lock_path = tmp_path / "monitor.pid" + lock_path = tmp_path / "relay.pid" setattr(mod, "_lock_path_override", lock_path) lock_path.write_text(json.dumps({"pid": os.getpid()}), encoding="utf-8") - import pytest - - mock_error = MagicMock() - with pytest.raises(SystemExit) as exc_info: - mod.acquire(error_fn=mock_error) - assert exc_info.value.code == 1 - mock_error.assert_called_once() - assert str(os.getpid()) in mock_error.call_args[0][0] + assert mod.try_acquire() is False def test_reclaims_stale_lock(self, tmp_path): - """acquire() reclaims the lock when the recorded PID is dead.""" + """try_acquire() reclaims the lock when the recorded PID is dead.""" mod = _import_lock() - lock_path = tmp_path / "monitor.pid" + lock_path = tmp_path / "relay.pid" setattr(mod, "_lock_path_override", lock_path) lock_path.write_text(json.dumps({"pid": 99999999}), encoding="utf-8") with patch.object(mod, "_is_pid_alive", return_value=False): - mod.acquire() + assert mod.try_acquire() is True data = json.loads(lock_path.read_text(encoding="utf-8")) assert data["pid"] == os.getpid() def test_reclaims_corrupt_lock_file(self, tmp_path): - """acquire() overwrites a corrupt lock file.""" + """try_acquire() overwrites a corrupt lock file.""" mod = _import_lock() - lock_path = tmp_path / "monitor.pid" + lock_path = tmp_path / "relay.pid" setattr(mod, "_lock_path_override", lock_path) lock_path.write_text("{corrupt json", encoding="utf-8") - mod.acquire() + assert mod.try_acquire() is True data = json.loads(lock_path.read_text(encoding="utf-8")) assert data["pid"] == os.getpid() def test_creates_parent_directories(self, tmp_path): - """acquire() creates parent directories if they don't exist.""" + """try_acquire() creates parent directories if they don't exist.""" mod = _import_lock() - lock_path = tmp_path / "nested" / "dir" / "monitor.pid" + lock_path = tmp_path / "nested" / "dir" / "relay.pid" setattr(mod, "_lock_path_override", lock_path) - mod.acquire() - + assert mod.try_acquire() is True assert lock_path.exists() class TestRelease: - """Test single-instance lock release.""" + """Test relay lock release.""" def test_removes_lock_file(self, tmp_path): """release() removes the lock file.""" mod = _import_lock() - lock_path = tmp_path / "monitor.pid" + lock_path = tmp_path / "relay.pid" setattr(mod, "_lock_path_override", lock_path) - mod.acquire() + mod.try_acquire() assert lock_path.exists() mod.release() @@ -182,10 +175,10 @@ class TestRelease: def test_clears_held_lock_state(self, tmp_path): """release() clears the _held_lock global.""" mod = _import_lock() - lock_path = tmp_path / "monitor.pid" + lock_path = tmp_path / "relay.pid" setattr(mod, "_lock_path_override", lock_path) - mod.acquire() + mod.try_acquire() mod.release() assert mod._held_lock is None @@ -198,10 +191,39 @@ class TestRelease: def test_release_handles_already_deleted_file(self, tmp_path): """release() handles the case where the lock file was already deleted.""" mod = _import_lock() - lock_path = tmp_path / "monitor.pid" + lock_path = tmp_path / "relay.pid" setattr(mod, "_lock_path_override", lock_path) - mod.acquire() + mod.try_acquire() lock_path.unlink() mod.release() assert mod._held_lock is None + + +class TestConcurrentViewers: + """Concurrent monitor viewers: relay lock scoped, display never blocked.""" + + def test_second_acquire_returns_false(self, tmp_path): + """Second try_acquire() returns False when first holds the lock.""" + mod = _import_lock() + lock_path = tmp_path / "relay.pid" + setattr(mod, "_lock_path_override", lock_path) + + assert mod.try_acquire() is True + assert mod.try_acquire() is False + + def test_release_then_reacquire(self, tmp_path): + """After release(), another process can acquire the relay lock.""" + mod = _import_lock() + lock_path = tmp_path / "relay.pid" + setattr(mod, "_lock_path_override", lock_path) + + assert mod.try_acquire() is True + mod.release() + assert mod.try_acquire() is True + + def test_lock_path_is_relay_pid(self): + """Default lock file is relay.pid, not monitor.pid.""" + mod = _import_lock() + setattr(mod, "_lock_path_override", None) + assert mod.get_lock_path().name == "relay.pid" diff --git a/src/aipass/trigger/.seedgo/bypass.json b/src/aipass/trigger/.seedgo/bypass.json index 18ad35fa..2900bf23 100644 --- a/src/aipass/trigger/.seedgo/bypass.json +++ b/src/aipass/trigger/.seedgo/bypass.json @@ -551,6 +551,12 @@ "lines": [40], "pattern": "handler imported directly", "reason": "Test helper _import_module() must import the handler module directly to test it. All trigger test files follow this pattern." + }, + { + "file": "apps/handlers/medic_state.py", + "standard": "unused_function", + "pattern": "get_muted_branches", + "reason": "Public API returning List[str] of active muted branches. Used by 15+ existing tests and part of the medic_state interface. get_muted_branches_detail() supplements it for status display — this is the simple accessor." } ], "notes": { diff --git a/src/aipass/trigger/apps/handlers/error_registry.py b/src/aipass/trigger/apps/handlers/error_registry.py index 3a356a7d..bf97ac90 100644 --- a/src/aipass/trigger/apps/handlers/error_registry.py +++ b/src/aipass/trigger/apps/handlers/error_registry.py @@ -227,6 +227,25 @@ _circuit_breaker = _load_circuit_breaker_state() # --------------------------------------------------------------------------- +def _evaluate_state() -> None: + """Evaluate circuit breaker state transitions on read. + + If state is 'open' and cooldown has expired, transition to 'half_open' + with the probe slot available. Called by both circuit_breaker_allows() + and get_circuit_breaker_status() so the breaker self-heals even when + medic is off and no dispatches are running. + """ + global _circuit_breaker + if _circuit_breaker.state != "open": + return + elapsed = time.time() - _circuit_breaker.opened_at + if elapsed >= _circuit_breaker.cooldown_seconds: + _circuit_breaker.state = "half_open" + _circuit_breaker.half_open_allow = True + _circuit_breaker.summary_sent = False + _save_circuit_breaker_state() + + def circuit_breaker_allows() -> bool: """Check if the circuit breaker allows dispatch. @@ -234,34 +253,25 @@ def circuit_breaker_allows() -> bool: - Closed (normal): All dispatches allowed. Records are checked against trip_threshold to determine if breaker should open. - Open (paused): No dispatches. Transitions to half_open after cooldown - period expires. - - Half-Open (testing): Allow ONE dispatch to test recovery. If it - resolves, caller should reset to Closed. If another error comes, - circuit_breaker_record_error() will re-open with doubled cooldown. + period expires (evaluated on read via _evaluate_state). + - Half-Open (testing): Allow ONE dispatch to test recovery. On success + the caller must call circuit_breaker_probe_succeeded() to close. + If another error comes, circuit_breaker_record_error() re-opens + with doubled cooldown. Returns: True if dispatch is allowed, False if breaker is blocking """ global _circuit_breaker - now = time.time() + _evaluate_state() if _circuit_breaker.state == "closed": return True - if _circuit_breaker.state == "open": - elapsed = now - _circuit_breaker.opened_at - if elapsed >= _circuit_breaker.cooldown_seconds: - # Cooldown expired - transition to half_open - # This call IS the probe dispatch, so mark probe as used - _circuit_breaker.state = "half_open" - _circuit_breaker.half_open_allow = False - _circuit_breaker.summary_sent = False - return True - return False - if _circuit_breaker.state == "half_open": if _circuit_breaker.half_open_allow: _circuit_breaker.half_open_allow = False + _save_circuit_breaker_state() return True return False @@ -319,6 +329,25 @@ def circuit_breaker_trip(reason: str = "") -> None: _save_circuit_breaker_state() +def circuit_breaker_probe_succeeded() -> None: + """Close the breaker after a successful dispatch during half_open probe. + + Transitions half_open -> closed and resets cooldown to base_cooldown + so future trips start with the short cooldown again. No-op if the + breaker is not in half_open state. + """ + global _circuit_breaker + if _circuit_breaker.state != "half_open": + return + _circuit_breaker.state = "closed" + _circuit_breaker.opened_at = 0.0 + _circuit_breaker.cooldown_seconds = _circuit_breaker.base_cooldown + _circuit_breaker.recent_errors = [] + _circuit_breaker.summary_sent = False + _circuit_breaker.half_open_allow = True + _clear_circuit_breaker_state() + + def circuit_breaker_reset() -> None: """Reset circuit breaker to closed state. @@ -339,16 +368,24 @@ def circuit_breaker_reset() -> None: def get_circuit_breaker_status() -> dict: """Get current circuit breaker state as a dictionary. + Evaluates state transitions first so the returned state is always + up-to-date (e.g. an expired open breaker will report as half_open). + Returns: Dict with keys: state, opened_at, cooldown_seconds, - recent_error_count, summary_sent + recent_error_count, summary_sent, remaining_seconds """ + _evaluate_state() + remaining = 0 + if _circuit_breaker.state == "open": + remaining = max(0, int(_circuit_breaker.cooldown_seconds - (time.time() - _circuit_breaker.opened_at))) return { "state": _circuit_breaker.state, "opened_at": _circuit_breaker.opened_at, "cooldown_seconds": _circuit_breaker.cooldown_seconds, "recent_error_count": len(_circuit_breaker.recent_errors), "summary_sent": _circuit_breaker.summary_sent, + "remaining_seconds": remaining, } diff --git a/src/aipass/trigger/apps/handlers/events/error_detected.py b/src/aipass/trigger/apps/handlers/events/error_detected.py index 616fd0ac..3f0caca7 100644 --- a/src/aipass/trigger/apps/handlers/events/error_detected.py +++ b/src/aipass/trigger/apps/handlers/events/error_detected.py @@ -81,6 +81,7 @@ try: from aipass.trigger.apps.handlers.error_registry import ( circuit_breaker_allows, circuit_breaker_record_error, + circuit_breaker_probe_succeeded, should_dispatch as registry_should_dispatch, record_dispatch as registry_record_dispatch, ) @@ -97,6 +98,10 @@ except ImportError: """Fallback no-op error recording when error_registry is unavailable.""" pass + def circuit_breaker_probe_succeeded() -> None: + """Fallback no-op when error_registry is unavailable.""" + pass + def registry_should_dispatch(fingerprint: str) -> bool: """Fallback dispatch check that always allows dispatch for any fingerprint.""" return True @@ -116,44 +121,70 @@ def _is_medic_enabled() -> bool: """ Check if medic (auto-healing dispatch) is enabled. - Reads medic_enabled from trigger_config.json. + Reads medic_enabled from trigger_config.json. If disabled with a TTL + (medic_disabled_until timestamp), treats an expired TTL as enabled. Defaults to True if config is missing or unreadable. Returns: True if medic dispatch is enabled """ try: - if TRIGGER_CONFIG_FILE.exists(): - data = json.loads(TRIGGER_CONFIG_FILE.read_text(encoding="utf-8")) - return bool(data.get("config", {}).get("medic_enabled", True)) + if not TRIGGER_CONFIG_FILE.exists(): + return True + data = json.loads(TRIGGER_CONFIG_FILE.read_text(encoding="utf-8")) + config = data.get("config", {}) + enabled = bool(config.get("medic_enabled", True)) + if enabled: + return True + disabled_until = config.get("medic_disabled_until") + if disabled_until and datetime.fromisoformat(disabled_until) <= datetime.now(): + return True + return False except Exception as exc: _log_warning(f"_is_medic_enabled config read failed: {exc}") - return True # Default to enabled on read failure - return True + return True + + +def _mute_entry_matches(entry, branch_lower: str, now: datetime) -> bool: + """Check if a single mute entry matches the branch and is still active.""" + if isinstance(entry, str): + return entry.lower() == branch_lower + if not isinstance(entry, dict): + return False + if entry.get("name", "").lower() != branch_lower: + return False + expires_at = entry.get("expires_at") + if expires_at is None: + return True + return datetime.fromisoformat(expires_at) > now def _is_branch_muted(branch_name: str) -> bool: """ Check if a specific branch is muted for medic dispatch. - Reads muted_branches list from trigger_config.json. - Muted branches have errors detected but NOT dispatched. + Reads muted_branches list from trigger_config.json. Supports both + legacy plain-string entries (permanent) and new dict entries with + optional expires_at timestamp. Expired TTL mutes are treated as + unmuted. Args: branch_name: Branch name (case-insensitive) Returns: - True if branch is in the muted list + True if branch is actively muted """ try: - if TRIGGER_CONFIG_FILE.exists(): - data = json.loads(TRIGGER_CONFIG_FILE.read_text(encoding="utf-8")) - muted = data.get("config", {}).get("muted_branches", []) - return branch_name.lower() in [b.lower() for b in muted] + if not TRIGGER_CONFIG_FILE.exists(): + return False + data = json.loads(TRIGGER_CONFIG_FILE.read_text(encoding="utf-8")) + muted = data.get("config", {}).get("muted_branches", []) + branch_lower = branch_name.lower() + now = datetime.now() + return any(_mute_entry_matches(e, branch_lower, now) for e in muted) except Exception as exc: _log_warning(f"_is_branch_muted config read failed: {exc}") return False - return False def set_send_email_callback(callback: Callable[..., bool]) -> None: @@ -550,6 +581,7 @@ def handle_error_detected( if _REGISTRY_DISPATCH_AVAILABLE and fingerprint: # Medic v2: per-fingerprint dispatch tracking registry_record_dispatch(fingerprint) + circuit_breaker_probe_succeeded() else: # Legacy: per-branch rate limiting _record_dispatch(recipient) diff --git a/src/aipass/trigger/apps/handlers/medic_state.py b/src/aipass/trigger/apps/handlers/medic_state.py index 333b59ff..ea9db1c6 100644 --- a/src/aipass/trigger/apps/handlers/medic_state.py +++ b/src/aipass/trigger/apps/handlers/medic_state.py @@ -17,9 +17,10 @@ Architecture: """ import json -from datetime import datetime +import re +from datetime import datetime, timedelta from pathlib import Path -from typing import Any, Dict, List +from typing import Any, Dict, List, Optional from aipass.prax.apps.modules.logger import get_direct_logger from aipass.trigger.apps.config import TRIGGER_ROOT, atomic_write_json, json_file_lock @@ -31,6 +32,51 @@ TRIGGER_CONFIG_FILE = TRIGGER_ROOT / "trigger_json" / "trigger_config.json" MEDIC_SUPPRESSED_LOG = TRIGGER_ROOT / "logs" / "medic_suppressed.jsonl" RATE_LIMITED_LOG = TRIGGER_ROOT / "logs" / "rate_limited.jsonl" +_DURATION_RE = re.compile(r"^(\d+)(h|d)$") + +DEFAULT_MUTE_SECONDS = 86400 # 24 hours +DEFAULT_OFF_SECONDS = 86400 # 24 hours + + +def parse_duration(duration_str: str) -> Optional[float]: + """Parse a duration string like '24h', '48h', '7d' into seconds. + + Args: + duration_str: Duration with unit suffix (h=hours, d=days) + + Returns: + Seconds as float, or None if unparseable + """ + m = _DURATION_RE.match(duration_str.strip()) + if not m: + return None + value, unit = int(m.group(1)), m.group(2) + if unit == "h": + return float(value * 3600) + return float(value * 86400) + + +def _is_mute_active(entry, now: datetime) -> bool: + """Check if a single mute entry is still active.""" + if isinstance(entry, str): + return True + if not isinstance(entry, dict): + return False + expires_at = entry.get("expires_at") + if expires_at is None: + return True + return datetime.fromisoformat(expires_at) > now + + +def _clean_expired_mutes(data: dict) -> None: + """Remove expired mute entries from config data in-place.""" + config = data.get("config", {}) + muted = config.get("muted_branches", []) + if not muted: + return + now = datetime.now() + config["muted_branches"] = [e for e in muted if _is_mute_active(e, now)] + def read_config() -> dict: """ @@ -50,7 +96,7 @@ def read_config() -> dict: def write_config(data: dict) -> bool: """ - Write trigger_config.json. + Write trigger_config.json. Cleans expired mute entries before writing. Args: data: Config dict to persist @@ -59,6 +105,7 @@ def write_config(data: dict) -> bool: True on success, False on failure """ try: + _clean_expired_mutes(data) atomic_write_json(TRIGGER_CONFIG_FILE, data) return True except Exception as exc: @@ -70,19 +117,43 @@ def is_enabled() -> bool: """ Check if Medic is currently enabled. + If disabled with a TTL (medic_disabled_until), treats an expired + TTL as enabled — evaluate on read, no timers. + Returns: - True if medic_enabled is True in config (defaults to True) + True if medic_enabled is True or its TTL has expired """ data = read_config() - return bool(data.get("config", {}).get("medic_enabled", True)) + config = data.get("config", {}) + enabled = bool(config.get("medic_enabled", True)) + if not enabled: + disabled_until = config.get("medic_disabled_until") + if disabled_until: + if datetime.fromisoformat(disabled_until) <= datetime.now(): + return True + return enabled -def set_enabled(enabled: bool) -> bool: +def get_disabled_until() -> Optional[str]: + """Get the medic_disabled_until timestamp if set. + + Returns: + ISO timestamp string, or None if not set or permanent off + """ + data = read_config() + return data.get("config", {}).get("medic_disabled_until") + + +def set_enabled(enabled: bool, duration_seconds: Optional[float] = None) -> bool: """ Set medic_enabled flag in config. + When disabling with a duration, stores medic_disabled_until so the + off state auto-expires. When enabling, clears any stored expiry. + Args: enabled: True to enable, False to disable + duration_seconds: TTL in seconds for disable (None = permanent) Returns: True on success @@ -92,6 +163,11 @@ def set_enabled(enabled: bool) -> bool: if "config" not in data: data["config"] = {} data["config"]["medic_enabled"] = enabled + if not enabled and duration_seconds is not None: + expires = datetime.now() + timedelta(seconds=duration_seconds) + data["config"]["medic_disabled_until"] = expires.isoformat() + else: + data["config"].pop("medic_disabled_until", None) data["timestamp"] = datetime.now().strftime("%Y-%m-%d") if write_config(data): @@ -118,25 +194,74 @@ def _normalize_branch_name(name: str) -> str: def get_muted_branches() -> List[str]: """ - Get list of muted branch names. + Get list of currently active muted branch names. + + Evaluates TTL expiry on read — expired mutes are filtered out. Returns: List of muted branch names (lowercase, e.g., ['speakeasy', 'api']) """ data = read_config() raw = data.get("config", {}).get("muted_branches", []) - return [_normalize_branch_name(b) for b in raw] + now = datetime.now() + result = [] + for entry in raw: + if isinstance(entry, str): + result.append(_normalize_branch_name(entry)) + elif isinstance(entry, dict): + expires_at = entry.get("expires_at") + if expires_at is None or datetime.fromisoformat(expires_at) > now: + result.append(_normalize_branch_name(entry.get("name", ""))) + return result -def mute_branch(branch_name: str) -> bool: +def get_muted_branches_detail() -> List[Dict[str, Any]]: """ - Add a branch to the muted list. + Get muted branches with expiry info for status display. + + Returns active mutes only (expired ones filtered out). + + Returns: + List of dicts with 'name' and 'expires_at' (None = permanent) + """ + data = read_config() + raw = data.get("config", {}).get("muted_branches", []) + now = datetime.now() + result = [] + for entry in raw: + if isinstance(entry, str): + result.append({"name": _normalize_branch_name(entry), "expires_at": None}) + elif isinstance(entry, dict): + expires_at = entry.get("expires_at") + if expires_at is None or datetime.fromisoformat(expires_at) > now: + result.append( + { + "name": _normalize_branch_name(entry.get("name", "")), + "expires_at": expires_at, + } + ) + return result + + +def _mute_entry_name(entry) -> str: + """Extract the normalized branch name from a mute entry (string or dict).""" + if isinstance(entry, str): + return _normalize_branch_name(entry) + if isinstance(entry, dict): + return _normalize_branch_name(entry.get("name", "")) + return "" + + +def mute_branch(branch_name: str, duration_seconds: Optional[float] = None) -> bool: + """ + Add a branch to the muted list with optional TTL. Muted branches will have errors detected but NOT dispatched. Persists in trigger_config.json. Args: branch_name: Branch name (with or without @) + duration_seconds: TTL in seconds (None = permanent/forever) Returns: True on success @@ -146,10 +271,14 @@ def mute_branch(branch_name: str) -> bool: data = read_config() if "config" not in data: data["config"] = {} - muted = [_normalize_branch_name(b) for b in data["config"].get("muted_branches", [])] - if clean not in muted: - muted.append(clean) - data["config"]["muted_branches"] = muted + raw_muted = data["config"].get("muted_branches", []) + new_muted = [e for e in raw_muted if _mute_entry_name(e) != clean] + if duration_seconds is not None: + expires = datetime.now() + timedelta(seconds=duration_seconds) + new_muted.append({"name": clean, "expires_at": expires.isoformat()}) + else: + new_muted.append({"name": clean, "expires_at": None}) + data["config"]["muted_branches"] = new_muted data["timestamp"] = datetime.now().strftime("%Y-%m-%d") return write_config(data) @@ -169,9 +298,8 @@ def unmute_branch(branch_name: str) -> bool: data = read_config() if "config" not in data: data["config"] = {} - muted = [_normalize_branch_name(b) for b in data["config"].get("muted_branches", [])] - muted = [b for b in muted if b != clean] - data["config"]["muted_branches"] = muted + raw_muted = data["config"].get("muted_branches", []) + data["config"]["muted_branches"] = [e for e in raw_muted if _mute_entry_name(e) != clean] data["timestamp"] = datetime.now().strftime("%Y-%m-%d") return write_config(data) diff --git a/src/aipass/trigger/apps/modules/errors.py b/src/aipass/trigger/apps/modules/errors.py index 209eaea2..0a7b2c11 100644 --- a/src/aipass/trigger/apps/modules/errors.py +++ b/src/aipass/trigger/apps/modules/errors.py @@ -22,7 +22,6 @@ Architecture: Module orchestrates, error_registry handler manages data import json import os import sys -import time from typing import Optional @@ -465,13 +464,8 @@ def _cmd_circuit_breaker(console, args: list) -> bool: if cb_st == "closed": console.print(" [dim]Normal operation - all dispatch allowed[/dim]") elif cb_st == "open": - opened_at = cb.get("opened_at", 0) - cooldown = cb.get("cooldown_seconds", 0) - if opened_at > 0: - remaining = max(0, cooldown - int(time.time() - opened_at)) - error(f"Dispatch paused - {remaining}s remaining until half-open") - else: - error("Dispatch paused") + remaining = cb.get("remaining_seconds", 0) + error(f"Dispatch paused - {remaining}s remaining until half-open") console.print() console.print(" [dim]Run 'drone @trigger errors circuit-breaker reset' to force close[/dim]") elif cb_st == "half_open": diff --git a/src/aipass/trigger/apps/modules/medic.py b/src/aipass/trigger/apps/modules/medic.py index 018fb1a6..eb2f2d4d 100644 --- a/src/aipass/trigger/apps/modules/medic.py +++ b/src/aipass/trigger/apps/modules/medic.py @@ -30,11 +30,15 @@ from aipass.trigger.apps.handlers.json import json_handler from aipass.trigger.apps.handlers.medic_state import ( is_enabled, set_enabled, - get_muted_branches, + get_muted_branches_detail, + get_disabled_until, mute_branch, unmute_branch, get_suppression_stats, get_rate_limit_stats, + parse_duration, + DEFAULT_MUTE_SECONDS, + DEFAULT_OFF_SECONDS, ) if sys.platform == "win32": @@ -188,21 +192,24 @@ def print_help() -> None: console.rule("COMMANDS") console.print() console.print(" [bold]on[/bold] Enable error dispatch (starts log watcher if needed)") - console.print(" [bold]off[/bold] Disable error dispatch globally (errors still logged)") + console.print(" [bold]off[/bold] Disable dispatch for 24h (detection continues)") + console.print(" [bold]off --forever[/bold] Disable dispatch permanently (stops log watcher)") console.print(" [bold]status[/bold] Show current state, muted branches, and statistics") - console.print(" [bold]mute[/bold] @branch Suppress dispatch for a specific branch") + console.print(" [bold]mute[/bold] @branch Suppress dispatch for 24h (default)") + console.print(" [bold]mute[/bold] @branch --for 48h Custom TTL (e.g. 48h, 7d)") + console.print(" [bold]mute[/bold] @branch --forever Permanent mute") console.print(" [bold]unmute[/bold] @branch Resume dispatch for a muted branch") console.print(" [bold]help[/bold] Show this help") console.print() console.rule("OFF vs MUTE") console.print() - console.print(" [yellow]off[/yellow] Global kill switch. ALL error dispatch stops. No branch") - console.print(" receives auto-healing emails. Errors still logged to") - console.print(" medic_suppressed.jsonl for review.") + console.print(" [yellow]off[/yellow] 24h dispatch suppression (detection continues).") + console.print(" Auto-resumes after 24 hours. Use --forever for") + console.print(" permanent disable (stops log watcher too).") console.print() - console.print(" [yellow]mute[/yellow] Per-branch suppress. Only the muted branch stops receiving") - console.print(" dispatch. All other branches continue normally. Muted errors") - console.print(" logged to medic_suppressed.jsonl.") + console.print(" [yellow]mute[/yellow] Per-branch suppress for 24h (default).") + console.print(" --for 48h or --for 7d for custom duration.") + console.print(" --forever for permanent. Auto-expires — no need to unmute.") console.print() console.rule("EXAMPLES") console.print() @@ -231,20 +238,78 @@ def print_help() -> None: console.print() +def _parse_duration_args(args: list) -> tuple: + """Extract --for and --forever from args. + + Returns: + (duration_seconds_or_None, is_forever, remaining_args) + Default (no flags): duration=DEFAULT_MUTE_SECONDS, is_forever=False + """ + remaining = [] + duration = None + is_forever = False + i = 0 + while i < len(args): + if args[i] == "--forever": + is_forever = True + i += 1 + elif args[i] == "--for" and i + 1 < len(args): + parsed = parse_duration(args[i + 1]) + if parsed is not None: + duration = parsed + i += 2 + else: + remaining.append(args[i]) + i += 1 + if is_forever: + return None, True, remaining + if duration is not None: + return duration, False, remaining + return float(DEFAULT_MUTE_SECONDS), False, remaining + + +def _fmt_remaining(iso_expiry: str) -> str: + """Format time remaining from an ISO expiry timestamp.""" + from datetime import datetime + + try: + expires = datetime.fromisoformat(iso_expiry) + remaining = expires - datetime.now() + total_secs = max(0, int(remaining.total_seconds())) + if total_secs <= 0: + return "expired" + hours, rem = divmod(total_secs, 3600) + minutes = rem // 60 + if hours >= 24: + days = hours // 24 + hours = hours % 24 + return f"{days}d {hours}h" + return f"{hours}h {minutes}m" + except Exception as exc: + logger.warning("[MEDIC] _fmt_remaining parse failed: %s", exc) + return "unknown" + + def _handle_mute(console, args: list) -> None: - """Handle 'medic mute @branch'.""" + """Handle 'medic mute @branch [--for ] [--forever]'.""" from aipass.cli.apps.modules import error - if not args: - error("Missing branch name", suggestion="Usage: medic mute @branch") + duration_secs, is_forever, rest = _parse_duration_args(args) + + if not rest: + error("Missing branch name", suggestion="Usage: medic mute @branch [--for 48h] [--forever]") return - branch_name = _extract_branch_name(args[0]) + branch_name = _extract_branch_name(rest[0]) if not branch_name: - error("Missing branch name", suggestion="Usage: medic mute @branch") + error("Missing branch name", suggestion="Usage: medic mute @branch [--for 48h] [--forever]") return - if mute_branch(branch_name): + if mute_branch(branch_name, duration_seconds=duration_secs): logger.info(f"[MEDIC] Muted branch: {branch_name}") - console.print(f" [yellow]Muted[/yellow] @{branch_name} — errors logged but not dispatched") + if is_forever or duration_secs is None: + console.print(f" [yellow]Muted[/yellow] @{branch_name} — permanent (use unmute to restore)") + else: + hours = int(duration_secs) // 3600 + console.print(f" [yellow]Muted[/yellow] @{branch_name} — auto-expires in {hours}h") else: error(f"Failed to mute @{branch_name}", suggestion="Check trigger_config.json") @@ -274,17 +339,33 @@ def _handle_status(console) -> None: suppression = get_suppression_stats() rate_limits = get_rate_limit_stats() - muted = get_muted_branches() + muted_detail = get_muted_branches_detail() state_color = "green" if enabled else "yellow" state_text = "ENABLED" if enabled else "DISABLED" + + disabled_until = get_disabled_until() + if not enabled and disabled_until: + remaining = _fmt_remaining(disabled_until) + state_text = f"DISABLED (auto-resumes in {remaining})" + if watcher_active: watcher_text = "[green]running[/green] (systemd)" elif enabled: watcher_text = "[yellow]stopped[/yellow] — run [bold]medic on[/bold] to start" else: watcher_text = "stopped" - muted_text = ", ".join(f"@{b}" for b in muted) if muted else "none" + + if muted_detail: + muted_parts = [] + for m in muted_detail: + if m["expires_at"] is None: + muted_parts.append(f"@{m['name']} [dim](permanent)[/dim]") + else: + muted_parts.append(f"@{m['name']} [dim]({_fmt_remaining(m['expires_at'])} left)[/dim]") + muted_text = ", ".join(muted_parts) + else: + muted_text = "none" console.print("Medic Status") console.print(f" State: [{state_color}]{state_text}[/{state_color}]") @@ -330,30 +411,47 @@ def _handle_on(console) -> None: ) -def _handle_off(console) -> None: - """Handle 'medic off' — disable dispatch and stop watcher.""" +def _handle_off(console, args: list | None = None) -> None: + """Handle 'medic off [--forever]' — disable dispatch with 24h TTL or permanently.""" from rich.panel import Panel from aipass.cli.apps.modules import error - if not set_enabled(False): - error("Failed to disable Medic", suggestion="Check trigger_config.json") - return + args = args or [] + is_forever = "--forever" in args - logger.info("[MEDIC] Medic DISABLED - error dispatch suppressed") - if _is_service_active(): - _systemctl("stop") - logger.info("[MEDIC] Log watcher service stopped") - - console.print( - Panel( - "[bold yellow]Medic DISABLED[/bold yellow]\n\n" - "Error dispatch is [yellow]suppressed[/yellow]. Errors are still detected\n" - "and logged to [dim]medic_suppressed.jsonl[/dim] for review.\n" - "Log watcher: [yellow]stopped[/yellow]", - title="Medic", - border_style="yellow", + if is_forever: + if not set_enabled(False): + error("Failed to disable Medic", suggestion="Check trigger_config.json") + return + logger.info("[MEDIC] Medic DISABLED permanently") + if _is_service_active(): + _systemctl("stop") + logger.info("[MEDIC] Log watcher service stopped") + console.print( + Panel( + "[bold yellow]Medic DISABLED (permanent)[/bold yellow]\n\n" + "Error dispatch is [yellow]suppressed[/yellow]. Log watcher stopped.\n" + "Use [bold]medic on[/bold] to re-enable.", + title="Medic", + border_style="yellow", + ) + ) + else: + hours = DEFAULT_OFF_SECONDS // 3600 + if not set_enabled(False, duration_seconds=float(DEFAULT_OFF_SECONDS)): + error("Failed to disable Medic", suggestion="Check trigger_config.json") + return + logger.info("[MEDIC] Medic DISABLED for %dh", hours) + console.print( + Panel( + f"[bold yellow]Medic DISABLED ({hours}h)[/bold yellow]\n\n" + f"Error dispatch suppressed for {hours} hours, then auto-resumes.\n" + "Detection continues (log watcher stays running).\n" + "Use [bold]medic off --forever[/bold] for permanent disable.", + title="Medic", + border_style="yellow", + ) ) - ) def _route_medic_module(args: list) -> bool: @@ -401,7 +499,7 @@ def handle_command(command: str, args: list) -> bool: "mute": lambda: _handle_mute(console, args), "unmute": lambda: _handle_unmute(console, args), "on": lambda: _handle_on(console), - "off": lambda: _handle_off(console), + "off": lambda: _handle_off(console, args), "status": lambda: _handle_status(console), } handler = handlers.get(command) diff --git a/src/aipass/trigger/tests/test_error_detected.py b/src/aipass/trigger/tests/test_error_detected.py index 87619d05..93b4d92f 100644 --- a/src/aipass/trigger/tests/test_error_detected.py +++ b/src/aipass/trigger/tests/test_error_detected.py @@ -8,7 +8,9 @@ """Tests for error_detected event handler: set_send_email_callback, handle_error_detected, and fallback stubs.""" +import json import sys +from datetime import datetime, timedelta from pathlib import Path from unittest.mock import MagicMock @@ -399,3 +401,249 @@ class TestFallbackStubs: """Module reports registry dispatch as unavailable.""" mod = _import_module() assert mod._REGISTRY_DISPATCH_AVAILABLE is False + + +# --------------------------------------------------------------------------- +# TTL-aware medic enable/disable +# --------------------------------------------------------------------------- + + +class TestMedicEnabledTTL: + """Tests for _is_medic_enabled TTL expiry behavior.""" + + def test_medic_enabled_ttl_expired(self) -> None: + """medic_enabled=False with expired TTL -> treated as enabled, dispatch proceeds.""" + mod = _import_module() + real_is_medic_enabled = mod._is_medic_enabled + send = _setup_happy_path(mod) + mod._is_medic_enabled = real_is_medic_enabled # type: ignore[attr-defined] + + config_file = mod.TRIGGER_CONFIG_FILE + config_file.parent.mkdir(parents=True, exist_ok=True) + past = (datetime.now() - timedelta(hours=1)).isoformat() + config_file.write_text( + json.dumps( + { + "config": { + "medic_enabled": False, + "medic_disabled_until": past, + } + } + ), + encoding="utf-8", + ) + + mod.handle_error_detected( + branch="flow", + module="cfg", + message="err", + error_hash="h1", + count=2, + fingerprint="fp_ttl_exp", + ) + + send.assert_called_once() + + def test_medic_enabled_ttl_active(self) -> None: + """medic_enabled=False with future TTL -> medic still disabled, dispatch suppressed.""" + mod = _import_module() + real_is_medic_enabled = mod._is_medic_enabled + send = _setup_happy_path(mod) + mod._is_medic_enabled = real_is_medic_enabled # type: ignore[attr-defined] + + config_file = mod.TRIGGER_CONFIG_FILE + config_file.parent.mkdir(parents=True, exist_ok=True) + future = (datetime.now() + timedelta(hours=1)).isoformat() + config_file.write_text( + json.dumps( + { + "config": { + "medic_enabled": False, + "medic_disabled_until": future, + } + } + ), + encoding="utf-8", + ) + + mod.handle_error_detected( + branch="flow", + module="cfg", + message="err", + error_hash="h1", + count=2, + fingerprint="fp_ttl_act", + ) + + send.assert_not_called() + + +# --------------------------------------------------------------------------- +# Branch mute dict/string format support +# --------------------------------------------------------------------------- + + +class TestBranchMutedFormats: + """Tests for _is_branch_muted dict and string format support.""" + + def test_branch_muted_dict_format_active(self) -> None: + """Dict entry with future expires_at -> branch IS muted, dispatch suppressed.""" + mod = _import_module() + real_is_branch_muted = mod._is_branch_muted + send = _setup_happy_path(mod) + mod._is_branch_muted = real_is_branch_muted # type: ignore[attr-defined] + mod._get_registered_emails = MagicMock(return_value={"@api", "@flow"}) # type: ignore[attr-defined] + + config_file = mod.TRIGGER_CONFIG_FILE + config_file.parent.mkdir(parents=True, exist_ok=True) + future = (datetime.now() + timedelta(hours=1)).isoformat() + config_file.write_text( + json.dumps( + { + "config": { + "medic_enabled": True, + "muted_branches": [{"name": "api", "expires_at": future}], + } + } + ), + encoding="utf-8", + ) + + mod.handle_error_detected( + branch="api", + module="cfg", + message="err", + error_hash="h1", + count=2, + fingerprint="fp_mute_act", + ) + + send.assert_not_called() + + def test_branch_muted_dict_format_expired(self) -> None: + """Dict entry with past expires_at -> branch NOT muted, dispatch proceeds.""" + mod = _import_module() + real_is_branch_muted = mod._is_branch_muted + send = _setup_happy_path(mod) + mod._is_branch_muted = real_is_branch_muted # type: ignore[attr-defined] + mod._get_registered_emails = MagicMock(return_value={"@api", "@flow"}) # type: ignore[attr-defined] + + config_file = mod.TRIGGER_CONFIG_FILE + config_file.parent.mkdir(parents=True, exist_ok=True) + past = (datetime.now() - timedelta(hours=1)).isoformat() + config_file.write_text( + json.dumps( + { + "config": { + "medic_enabled": True, + "muted_branches": [{"name": "api", "expires_at": past}], + } + } + ), + encoding="utf-8", + ) + + mod.handle_error_detected( + branch="api", + module="cfg", + message="err", + error_hash="h1", + count=2, + fingerprint="fp_mute_exp", + ) + + send.assert_called_once() + + def test_branch_muted_plain_string_backcompat(self) -> None: + """Plain string entry in muted_branches -> branch IS muted (permanent).""" + mod = _import_module() + real_is_branch_muted = mod._is_branch_muted + send = _setup_happy_path(mod) + mod._is_branch_muted = real_is_branch_muted # type: ignore[attr-defined] + mod._get_registered_emails = MagicMock(return_value={"@api", "@flow"}) # type: ignore[attr-defined] + + config_file = mod.TRIGGER_CONFIG_FILE + config_file.parent.mkdir(parents=True, exist_ok=True) + config_file.write_text( + json.dumps( + { + "config": { + "medic_enabled": True, + "muted_branches": ["api"], + } + } + ), + encoding="utf-8", + ) + + mod.handle_error_detected( + branch="api", + module="cfg", + message="err", + error_hash="h1", + count=2, + fingerprint="fp_str_perm", + ) + + send.assert_not_called() + + def test_branch_muted_dict_permanent(self) -> None: + """Dict entry with expires_at=null -> branch IS muted (permanent).""" + mod = _import_module() + real_is_branch_muted = mod._is_branch_muted + send = _setup_happy_path(mod) + mod._is_branch_muted = real_is_branch_muted # type: ignore[attr-defined] + mod._get_registered_emails = MagicMock(return_value={"@api", "@flow"}) # type: ignore[attr-defined] + + config_file = mod.TRIGGER_CONFIG_FILE + config_file.parent.mkdir(parents=True, exist_ok=True) + config_file.write_text( + json.dumps( + { + "config": { + "medic_enabled": True, + "muted_branches": [{"name": "api", "expires_at": None}], + } + } + ), + encoding="utf-8", + ) + + mod.handle_error_detected( + branch="api", + module="cfg", + message="err", + error_hash="h1", + count=2, + fingerprint="fp_dict_perm", + ) + + send.assert_not_called() + + +# --------------------------------------------------------------------------- +# circuit_breaker_probe_succeeded after dispatch +# --------------------------------------------------------------------------- + + +class TestProbeSucceeded: + """Tests for circuit_breaker_probe_succeeded called after dispatch.""" + + def test_probe_succeeded_called_after_dispatch(self) -> None: + """circuit_breaker_probe_succeeded is called after successful dispatch with fingerprint.""" + mod = _import_module() + send = _setup_happy_path(mod) + mod.circuit_breaker_probe_succeeded = MagicMock() # type: ignore[attr-defined] + + mod.handle_error_detected( + branch="flow", + module="cfg", + message="err", + error_hash="h1", + count=2, + fingerprint="fp_probe", + ) + + send.assert_called_once() + mod.registry_record_dispatch.assert_called_once_with("fp_probe") # type: ignore[attr-defined] + mod.circuit_breaker_probe_succeeded.assert_called_once() # type: ignore[attr-defined] diff --git a/src/aipass/trigger/tests/test_error_registry.py b/src/aipass/trigger/tests/test_error_registry.py index ea3579a0..b0183284 100644 --- a/src/aipass/trigger/tests/test_error_registry.py +++ b/src/aipass/trigger/tests/test_error_registry.py @@ -1012,3 +1012,165 @@ def test_purge_stale_custom_days(tmp_path: Path) -> None: # 7-day cutoff should remove it assert er.purge_stale(days=7) == 1 + + +# =========================================================================== +# 17. Circuit breaker self-heal (_evaluate_state, probe_succeeded, status) +# =========================================================================== + + +def test_evaluate_state_transitions_open_to_half_open_after_cooldown(tmp_path: Path) -> None: + """_evaluate_state transitions open -> half_open when cooldown has expired.""" + _seed_registry(tmp_path) + er = _import_registry() + er.circuit_breaker_reset() + + er.circuit_breaker_trip(reason="test") + # Backdate opened_at so cooldown is expired + er._circuit_breaker.opened_at = time.time() - er._circuit_breaker.cooldown_seconds - 10 + + status = er.get_circuit_breaker_status() + assert status["state"] == "half_open" + + +def test_evaluate_state_no_transition_before_cooldown(tmp_path: Path) -> None: + """_evaluate_state keeps state open when cooldown has not yet expired.""" + _seed_registry(tmp_path) + er = _import_registry() + er.circuit_breaker_reset() + + er.circuit_breaker_trip(reason="test") + # opened_at is now (cooldown is 300s), so it should stay open + assert er._circuit_breaker.state == "open" + + status = er.get_circuit_breaker_status() + assert status["state"] == "open" + + +def test_evaluate_state_no_op_when_closed(tmp_path: Path) -> None: + """_evaluate_state is a no-op when breaker is already closed.""" + _seed_registry(tmp_path) + er = _import_registry() + er.circuit_breaker_reset() + + assert er._circuit_breaker.state == "closed" + + status = er.get_circuit_breaker_status() + assert status["state"] == "closed" + + +def test_probe_succeeded_closes_breaker(tmp_path: Path) -> None: + """circuit_breaker_probe_succeeded transitions half_open -> closed and resets cooldown.""" + _seed_registry(tmp_path) + er = _import_registry() + er.circuit_breaker_reset() + + # Trip the breaker and expire cooldown to get to half_open + er.circuit_breaker_trip(reason="test") + er._circuit_breaker.opened_at = time.time() - er._circuit_breaker.cooldown_seconds - 1 + er.circuit_breaker_allows() # Transitions to half_open + assert er._circuit_breaker.state == "half_open" + + er.circuit_breaker_probe_succeeded() + + assert er._circuit_breaker.state == "closed" + assert er._circuit_breaker.cooldown_seconds == er._circuit_breaker.base_cooldown + assert er._circuit_breaker.opened_at == 0.0 + assert er._circuit_breaker.recent_errors == [] + + +def test_probe_succeeded_noop_when_closed(tmp_path: Path) -> None: + """circuit_breaker_probe_succeeded is a no-op when breaker is closed.""" + _seed_registry(tmp_path) + er = _import_registry() + er.circuit_breaker_reset() + + assert er._circuit_breaker.state == "closed" + + er.circuit_breaker_probe_succeeded() + + assert er._circuit_breaker.state == "closed" + + +def test_probe_succeeded_noop_when_open(tmp_path: Path) -> None: + """circuit_breaker_probe_succeeded is a no-op when breaker is open.""" + _seed_registry(tmp_path) + er = _import_registry() + er.circuit_breaker_reset() + + er.circuit_breaker_trip(reason="test") + assert er._circuit_breaker.state == "open" + + er.circuit_breaker_probe_succeeded() + + assert er._circuit_breaker.state == "open" + + +def test_status_returns_remaining_seconds(tmp_path: Path) -> None: + """get_circuit_breaker_status returns approximately correct remaining_seconds.""" + _seed_registry(tmp_path) + er = _import_registry() + er.circuit_breaker_reset() + + er.circuit_breaker_trip(reason="test") + cooldown = er._circuit_breaker.cooldown_seconds + # Backdate opened_at by 100 seconds so remaining ~ cooldown - 100 + er._circuit_breaker.opened_at = time.time() - 100 + + status = er.get_circuit_breaker_status() + expected_remaining = cooldown - 100 + # Allow 2-second tolerance for timing + assert abs(status["remaining_seconds"] - expected_remaining) <= 2 + + +def test_status_remaining_zero_when_closed(tmp_path: Path) -> None: + """remaining_seconds is 0 when breaker is closed.""" + _seed_registry(tmp_path) + er = _import_registry() + er.circuit_breaker_reset() + + status = er.get_circuit_breaker_status() + assert status["remaining_seconds"] == 0 + + +def test_breaker_half_open_on_read_then_allows_probe(tmp_path: Path) -> None: + """Expired open breaker transitions to half_open on allows() and consumes probe slot.""" + _seed_registry(tmp_path) + er = _import_registry() + er.circuit_breaker_reset() + + er.circuit_breaker_trip(reason="test") + # Expire the cooldown + er._circuit_breaker.opened_at = time.time() - er._circuit_breaker.cooldown_seconds - 1 + + # First call: transitions open -> half_open, returns True (probe allowed) + result = er.circuit_breaker_allows() + assert result is True + assert er._circuit_breaker.state == "half_open" + assert er._circuit_breaker.half_open_allow is False + + +def test_breaker_closes_after_successful_probe_dispatch(tmp_path: Path) -> None: + """Full self-heal cycle: open -> half_open -> probe allowed -> probe_succeeded -> closed.""" + _seed_registry(tmp_path) + er = _import_registry() + er.circuit_breaker_reset() + + base_cooldown = er._circuit_breaker.base_cooldown + + # Trip the breaker + er.circuit_breaker_trip(reason="test") + assert er._circuit_breaker.state == "open" + + # Expire the cooldown + er._circuit_breaker.opened_at = time.time() - er._circuit_breaker.cooldown_seconds - 1 + + # Probe dispatch: transitions open -> half_open and allows + assert er.circuit_breaker_allows() is True + assert er._circuit_breaker.state == "half_open" + + # Probe succeeded: transitions half_open -> closed + er.circuit_breaker_probe_succeeded() + assert er._circuit_breaker.state == "closed" + assert er._circuit_breaker.cooldown_seconds == base_cooldown + assert er._circuit_breaker.opened_at == 0.0 diff --git a/src/aipass/trigger/tests/test_medic.py b/src/aipass/trigger/tests/test_medic.py index 2fda1a28..21274dc2 100644 --- a/src/aipass/trigger/tests/test_medic.py +++ b/src/aipass/trigger/tests/test_medic.py @@ -47,6 +47,8 @@ def _mock_infrastructure(monkeypatch): medic_state_mod.is_enabled = MagicMock(return_value=True) medic_state_mod.set_enabled = MagicMock(return_value=True) medic_state_mod.get_muted_branches = MagicMock(return_value=[]) + medic_state_mod.get_muted_branches_detail = MagicMock(return_value=[]) + medic_state_mod.get_disabled_until = MagicMock(return_value=None) medic_state_mod.mute_branch = MagicMock(return_value=True) medic_state_mod.unmute_branch = MagicMock(return_value=True) medic_state_mod.get_suppression_stats = MagicMock( @@ -61,6 +63,9 @@ def _mock_infrastructure(monkeypatch): "last_rate_limited": "never", } ) + medic_state_mod.parse_duration = MagicMock(return_value=None) + medic_state_mod.DEFAULT_MUTE_SECONDS = 86400 + medic_state_mod.DEFAULT_OFF_SECONDS = 86400 monkeypatch.setitem(sys.modules, "aipass.trigger.apps.handlers.medic_state", medic_state_mod) # -- CLI console (lazy import inside handle_command) -------------------- @@ -189,7 +194,7 @@ def test_handle_command_on_failure_prints_error(): def test_handle_command_off_disables_medic(): - """handle_command('off', []) calls set_enabled(False), prints Panel, returns True.""" + """handle_command('off', []) calls set_enabled with 24h TTL, prints Panel, returns True.""" medic = _import_medic() with patch.object(medic, "_systemctl", return_value=True): @@ -198,21 +203,34 @@ def test_handle_command_off_disables_medic(): assert result is True state = _get_medic_state() - state.set_enabled.assert_called_with(False) - # Verify console.print was called (Panel is a mock object for success output) + state.set_enabled.assert_called_with(False, duration_seconds=86400.0) console = _get_console() assert console.print.call_count >= 1, "console.print should be called with success Panel" -def test_handle_command_off_stops_active_service(): - """handle_command('off', []) stops the service when it is active.""" +def test_handle_command_off_forever_stops_service(): + """handle_command('off', ['--forever']) stops the service and disables permanently.""" + medic = _import_medic() + + with patch.object(medic, "_systemctl", return_value=True) as mock_ctl: + with patch.object(medic, "_is_service_active", return_value=True): + result = medic.handle_command("off", ["--forever"]) + + assert result is True + state = _get_medic_state() + state.set_enabled.assert_called_with(False) + mock_ctl.assert_called_with("stop") + + +def test_handle_command_off_ttl_keeps_watcher(): + """handle_command('off', []) with default TTL does NOT stop the log watcher.""" medic = _import_medic() with patch.object(medic, "_systemctl", return_value=True) as mock_ctl: with patch.object(medic, "_is_service_active", return_value=True): medic.handle_command("off", []) - mock_ctl.assert_called_with("stop") + mock_ctl.assert_not_called() def test_handle_command_off_failure_prints_error(): @@ -249,7 +267,7 @@ def test_handle_command_status_returns_current_state(): assert result is True state = _get_medic_state() state.is_enabled.assert_called_once() - state.get_muted_branches.assert_called_once() + state.get_muted_branches_detail.assert_called_once() state.get_suppression_stats.assert_called_once() state.get_rate_limit_stats.assert_called_once() @@ -283,19 +301,22 @@ def test_handle_command_status_shows_disabled(): def test_handle_command_status_shows_muted_branches(): - """When branches are muted, status lists them in the muted branches line.""" + """When branches are muted, status lists them with expiry info.""" medic = _import_medic() state = _get_medic_state() - state.get_muted_branches.return_value = ["speakeasy", "api"] + state.get_muted_branches_detail.return_value = [ + {"name": "speakeasy", "expires_at": None}, + {"name": "api", "expires_at": None}, + ] with patch.object(medic, "_is_service_active", return_value=True): medic.handle_command("status", []) console = _get_console() printed = _get_print_str_args(console) - # Source builds: " Muted branches: @speakeasy, @api" - muted_line = " Muted branches: @speakeasy, @api" - assert muted_line in printed, f"Expected muted line '{muted_line}' in printed args: {printed}" + muted_lines = [p for p in printed if "Muted branches:" in p] + assert muted_lines, f"Expected muted branches line in printed args: {printed}" + assert "@speakeasy" in muted_lines[0] and "@api" in muted_lines[0] def test_handle_command_status_suppression_hint_when_disabled(): @@ -319,13 +340,13 @@ def test_handle_command_status_suppression_hint_when_disabled(): def test_handle_command_mute_branch(): - """handle_command('mute', ['@speakeasy']) mutes the branch.""" + """handle_command('mute', ['@speakeasy']) mutes the branch with 24h default TTL.""" medic = _import_medic() result = medic.handle_command("mute", ["@speakeasy"]) assert result is True state = _get_medic_state() - state.mute_branch.assert_called_once_with("speakeasy") + state.mute_branch.assert_called_once_with("speakeasy", duration_seconds=86400.0) def test_handle_command_mute_branch_without_at(): @@ -334,18 +355,18 @@ def test_handle_command_mute_branch_without_at(): medic.handle_command("mute", ["speakeasy"]) state = _get_medic_state() - state.mute_branch.assert_called_once_with("speakeasy") + state.mute_branch.assert_called_once_with("speakeasy", duration_seconds=86400.0) def test_handle_command_mute_prints_confirmation(): - """Successful mute prints the exact confirmation message with the branch name.""" + """Successful mute prints confirmation with TTL info.""" medic = _import_medic() medic.handle_command("mute", ["@api"]) console = _get_console() printed = _get_print_str_args(console) - expected = " [yellow]Muted[/yellow] @api — errors logged but not dispatched" - assert expected in printed, f"Expected mute confirmation '{expected}' in printed args: {printed}" + mute_lines = [p for p in printed if "Muted" in p and "@api" in p] + assert mute_lines, f"Expected mute confirmation for @api in printed args: {printed}" def test_handle_command_mute_failure_prints_error(): @@ -522,7 +543,7 @@ def test_handle_command_medic_routes_mute_with_args(): assert result is True state = _get_medic_state() - state.mute_branch.assert_called_once_with("speakeasy") + state.mute_branch.assert_called_once_with("speakeasy", duration_seconds=86400.0) # --------------------------------------------------------------------------- diff --git a/src/aipass/trigger/tests/test_medic_state.py b/src/aipass/trigger/tests/test_medic_state.py index 33486750..bc9e79d9 100644 --- a/src/aipass/trigger/tests/test_medic_state.py +++ b/src/aipass/trigger/tests/test_medic_state.py @@ -502,3 +502,441 @@ class TestGetRateLimitStats: assert result["rate_limited_count"] == 1 assert result["last_rate_limited"] == "unknown" + + +# --------------------------------------------------------------------------- +# Tests -- parse_duration +# --------------------------------------------------------------------------- + + +class TestParseDuration: + """Tests for parse_duration.""" + + def test_parse_duration_hours(self, state_mod): + """parse_duration converts '24h' to 86400.0 seconds.""" + result = state_mod.parse_duration("24h") + + assert result == 86400.0 + + def test_parse_duration_days(self, state_mod): + """parse_duration converts '7d' to 604800.0 seconds.""" + result = state_mod.parse_duration("7d") + + assert result == 604800.0 + + def test_parse_duration_invalid(self, state_mod): + """parse_duration returns None for invalid input.""" + result = state_mod.parse_duration("abc") + + assert result is None + + +# --------------------------------------------------------------------------- +# Tests -- mute_branch TTL +# --------------------------------------------------------------------------- + + +class TestMuteBranchTTL: + """Tests for mute_branch with TTL support.""" + + def test_mute_branch_default_permanent(self, state_mod): + """mute_branch with no duration stores dict with expires_at null.""" + state_mod.mute_branch("api") + + config_file = state_mod.TRIGGER_CONFIG_FILE + data = json.loads(config_file.read_text(encoding="utf-8")) + muted = data["config"]["muted_branches"] + assert len(muted) == 1 + assert muted[0] == {"name": "api", "expires_at": None} + + def test_mute_branch_with_ttl(self, state_mod): + """mute_branch with duration stores expires_at roughly 1h from now.""" + from datetime import datetime, timedelta + + before = datetime.now() + state_mod.mute_branch("api", duration_seconds=3600) + after = datetime.now() + + config_file = state_mod.TRIGGER_CONFIG_FILE + data = json.loads(config_file.read_text(encoding="utf-8")) + muted = data["config"]["muted_branches"] + assert len(muted) == 1 + assert muted[0]["name"] == "api" + expires = datetime.fromisoformat(muted[0]["expires_at"]) + assert expires >= before + timedelta(seconds=3600) + assert expires <= after + timedelta(seconds=3600) + + def test_mute_forever_null_expires(self, state_mod): + """mute_branch with duration_seconds=None stores expires_at as null.""" + state_mod.mute_branch("api", duration_seconds=None) + + config_file = state_mod.TRIGGER_CONFIG_FILE + data = json.loads(config_file.read_text(encoding="utf-8")) + muted = data["config"]["muted_branches"] + assert len(muted) == 1 + assert muted[0]["expires_at"] is None + + +# --------------------------------------------------------------------------- +# Tests -- get_muted_branches TTL filtering +# --------------------------------------------------------------------------- + + +class TestGetMutedBranchesTTL: + """Tests for get_muted_branches with TTL-aware filtering.""" + + def test_get_muted_branches_filters_expired(self, state_mod): + """get_muted_branches excludes dict entries whose expires_at is in the past.""" + from datetime import datetime, timedelta + + expired_ts = (datetime.now() - timedelta(hours=1)).isoformat() + config_file = state_mod.TRIGGER_CONFIG_FILE + config_file.parent.mkdir(parents=True, exist_ok=True) + config_file.write_text( + json.dumps( + { + "config": { + "muted_branches": [ + {"name": "api", "expires_at": expired_ts}, + ] + } + } + ), + encoding="utf-8", + ) + + result = state_mod.get_muted_branches() + + assert result == [] + + def test_get_muted_branches_keeps_active(self, state_mod): + """get_muted_branches includes dict entries whose expires_at is in the future.""" + from datetime import datetime, timedelta + + future_ts = (datetime.now() + timedelta(hours=1)).isoformat() + config_file = state_mod.TRIGGER_CONFIG_FILE + config_file.parent.mkdir(parents=True, exist_ok=True) + config_file.write_text( + json.dumps( + { + "config": { + "muted_branches": [ + {"name": "api", "expires_at": future_ts}, + ] + } + } + ), + encoding="utf-8", + ) + + result = state_mod.get_muted_branches() + + assert result == ["api"] + + def test_get_muted_branches_plain_string_backcompat(self, state_mod): + """get_muted_branches returns plain string entries as permanent mutes.""" + config_file = state_mod.TRIGGER_CONFIG_FILE + config_file.parent.mkdir(parents=True, exist_ok=True) + config_file.write_text( + json.dumps({"config": {"muted_branches": ["speakeasy"]}}), + encoding="utf-8", + ) + + result = state_mod.get_muted_branches() + + assert result == ["speakeasy"] + + +# --------------------------------------------------------------------------- +# Tests -- get_muted_branches_detail +# --------------------------------------------------------------------------- + + +class TestGetMutedBranchesDetail: + """Tests for get_muted_branches_detail.""" + + def test_get_muted_branches_detail_returns_expiry(self, state_mod): + """get_muted_branches_detail returns dicts with name and expires_at.""" + from datetime import datetime, timedelta + + future_ts = (datetime.now() + timedelta(hours=2)).isoformat() + config_file = state_mod.TRIGGER_CONFIG_FILE + config_file.parent.mkdir(parents=True, exist_ok=True) + config_file.write_text( + json.dumps( + { + "config": { + "muted_branches": [ + {"name": "api", "expires_at": future_ts}, + "speakeasy", + ] + } + } + ), + encoding="utf-8", + ) + + result = state_mod.get_muted_branches_detail() + + assert len(result) == 2 + assert result[0] == {"name": "api", "expires_at": future_ts} + assert result[1] == {"name": "speakeasy", "expires_at": None} + + +# --------------------------------------------------------------------------- +# Tests -- is_enabled TTL +# --------------------------------------------------------------------------- + + +class TestIsEnabledTTL: + """Tests for is_enabled with TTL-based disable.""" + + def test_is_enabled_ttl_expired_returns_true(self, state_mod): + """is_enabled returns True when disabled but medic_disabled_until is in the past.""" + from datetime import datetime, timedelta + + past_ts = (datetime.now() - timedelta(hours=1)).isoformat() + config_file = state_mod.TRIGGER_CONFIG_FILE + config_file.parent.mkdir(parents=True, exist_ok=True) + config_file.write_text( + json.dumps( + { + "config": { + "medic_enabled": False, + "medic_disabled_until": past_ts, + } + } + ), + encoding="utf-8", + ) + + result = state_mod.is_enabled() + + assert result is True + + def test_is_enabled_ttl_active_returns_false(self, state_mod): + """is_enabled returns False when disabled and medic_disabled_until is in the future.""" + from datetime import datetime, timedelta + + future_ts = (datetime.now() + timedelta(hours=1)).isoformat() + config_file = state_mod.TRIGGER_CONFIG_FILE + config_file.parent.mkdir(parents=True, exist_ok=True) + config_file.write_text( + json.dumps( + { + "config": { + "medic_enabled": False, + "medic_disabled_until": future_ts, + } + } + ), + encoding="utf-8", + ) + + result = state_mod.is_enabled() + + assert result is False + + def test_is_enabled_permanent_off(self, state_mod): + """is_enabled returns False when permanently disabled (no medic_disabled_until).""" + config_file = state_mod.TRIGGER_CONFIG_FILE + config_file.parent.mkdir(parents=True, exist_ok=True) + config_file.write_text( + json.dumps({"config": {"medic_enabled": False}}), + encoding="utf-8", + ) + + result = state_mod.is_enabled() + + assert result is False + + +# --------------------------------------------------------------------------- +# Tests -- set_enabled with duration +# --------------------------------------------------------------------------- + + +class TestSetEnabledDuration: + """Tests for set_enabled with duration_seconds parameter.""" + + def test_set_enabled_off_with_duration(self, state_mod): + """set_enabled(False, duration) stores medic_disabled_until timestamp.""" + from datetime import datetime, timedelta + + before = datetime.now() + state_mod.set_enabled(False, duration_seconds=86400) + after = datetime.now() + + config_file = state_mod.TRIGGER_CONFIG_FILE + data = json.loads(config_file.read_text(encoding="utf-8")) + assert data["config"]["medic_enabled"] is False + disabled_until = data["config"]["medic_disabled_until"] + ts = datetime.fromisoformat(disabled_until) + assert ts >= before + timedelta(seconds=86400) + assert ts <= after + timedelta(seconds=86400) + + def test_set_enabled_on_clears_disabled_until(self, state_mod): + """set_enabled(True) clears any existing medic_disabled_until.""" + # First disable with TTL + state_mod.set_enabled(False, duration_seconds=3600) + config_file = state_mod.TRIGGER_CONFIG_FILE + data = json.loads(config_file.read_text(encoding="utf-8")) + assert "medic_disabled_until" in data["config"] + + # Then re-enable + state_mod.set_enabled(True) + data = json.loads(config_file.read_text(encoding="utf-8")) + assert data["config"]["medic_enabled"] is True + assert "medic_disabled_until" not in data["config"] + + +# --------------------------------------------------------------------------- +# Tests -- get_disabled_until +# --------------------------------------------------------------------------- + + +class TestGetDisabledUntil: + """Tests for get_disabled_until.""" + + def test_get_disabled_until_returns_timestamp(self, state_mod): + """get_disabled_until returns the ISO timestamp when set.""" + from datetime import datetime + + state_mod.set_enabled(False, duration_seconds=86400) + + result = state_mod.get_disabled_until() + + assert result is not None + ts = datetime.fromisoformat(result) + assert ts > datetime.now() + + def test_get_disabled_until_returns_none(self, state_mod): + """get_disabled_until returns None when no TTL is set.""" + state_mod.set_enabled(False) + + result = state_mod.get_disabled_until() + + assert result is None + + +# --------------------------------------------------------------------------- +# Tests -- unmute_branch with dict entries +# --------------------------------------------------------------------------- + + +class TestUnmuteBranchDict: + """Tests for unmute_branch handling dict-format entries.""" + + def test_unmute_handles_dict_entries(self, state_mod): + """unmute_branch removes a dict-format mute entry.""" + state_mod.mute_branch("api", duration_seconds=3600) + assert "api" in state_mod.get_muted_branches() + + result = state_mod.unmute_branch("api") + + assert result is True + assert "api" not in state_mod.get_muted_branches() + + def test_unmute_handles_mixed_entries(self, state_mod): + """unmute_branch removes target from list with both string and dict entries.""" + from datetime import datetime, timedelta + + future_ts = (datetime.now() + timedelta(hours=2)).isoformat() + config_file = state_mod.TRIGGER_CONFIG_FILE + config_file.parent.mkdir(parents=True, exist_ok=True) + config_file.write_text( + json.dumps( + { + "config": { + "muted_branches": [ + "speakeasy", + {"name": "api", "expires_at": future_ts}, + {"name": "drone", "expires_at": None}, + ] + } + } + ), + encoding="utf-8", + ) + + state_mod.unmute_branch("api") + + muted = state_mod.get_muted_branches() + assert "speakeasy" in muted + assert "api" not in muted + assert "drone" in muted + + +# --------------------------------------------------------------------------- +# Tests -- _clean_expired_mutes +# --------------------------------------------------------------------------- + + +class TestCleanExpiredMutes: + """Tests for _clean_expired_mutes.""" + + def test_clean_expired_mutes_removes_old(self, state_mod): + """_clean_expired_mutes removes expired dict entries, keeps strings and active dicts.""" + from datetime import datetime, timedelta + + expired_ts = (datetime.now() - timedelta(hours=1)).isoformat() + future_ts = (datetime.now() + timedelta(hours=1)).isoformat() + + data = { + "config": { + "muted_branches": [ + "speakeasy", + {"name": "api", "expires_at": expired_ts}, + {"name": "drone", "expires_at": future_ts}, + {"name": "flow", "expires_at": None}, + ] + } + } + + state_mod._clean_expired_mutes(data) + + remaining = data["config"]["muted_branches"] + names = [] + for entry in remaining: + if isinstance(entry, str): + names.append(entry) + else: + names.append(entry["name"]) + assert "speakeasy" in names + assert "api" not in names + assert "drone" in names + assert "flow" in names + assert len(remaining) == 3 + + +# --------------------------------------------------------------------------- +# Tests -- write_config cleans expired mutes +# --------------------------------------------------------------------------- + + +class TestWriteConfigCleansMutes: + """Tests for write_config calling _clean_expired_mutes.""" + + def test_write_config_cleans_expired_mutes(self, state_mod): + """write_config removes expired mute entries before persisting.""" + from datetime import datetime, timedelta + + expired_ts = (datetime.now() - timedelta(hours=1)).isoformat() + future_ts = (datetime.now() + timedelta(hours=1)).isoformat() + + data = { + "config": { + "muted_branches": [ + {"name": "api", "expires_at": expired_ts}, + {"name": "drone", "expires_at": future_ts}, + ] + } + } + + state_mod.write_config(data) + + config_file = state_mod.TRIGGER_CONFIG_FILE + written = json.loads(config_file.read_text(encoding="utf-8")) + muted = written["config"]["muted_branches"] + assert len(muted) == 1 + assert muted[0]["name"] == "drone"