diff --git a/council-automation/.gitignore b/council-automation/.gitignore new file mode 100644 index 0000000..fb74ddf --- /dev/null +++ b/council-automation/.gitignore @@ -0,0 +1 @@ +tests/__pycache__/ diff --git a/council-automation/PPLX-BROWSER-CLOSED-RESOLUTION-2026-08-07.md b/council-automation/PPLX-BROWSER-CLOSED-RESOLUTION-2026-08-07.md new file mode 100644 index 0000000..7e8faeb --- /dev/null +++ b/council-automation/PPLX-BROWSER-CLOSED-RESOLUTION-2026-08-07.md @@ -0,0 +1,118 @@ +# RESOLVED — PPLX runner "Target page, context or browser has been closed" + +**Resolved:** 2026-08-07 14:2x PT. Companion to `PPLX-BROWSER-CLOSED-EVIDENCE-2026-08-07.md`. +**Status:** FIXED and verified with real successful `research_query` calls. + +## Root cause + +The runner attached to **the wrong Chrome**. + +`PerplexityCouncil._start_via_cdp()` decided the session keeper was alive using a single +signal: *does `http://127.0.0.1:9222/json/version` respond?* That is not proof of identity — +any process can serve plausible JSON on a port. Three independent facts lined up: + +1. **The keeper was not running.** The `PerplexitySessionKeeper` scheduled task has been + **Disabled since 2026-07-30** (`LastRunTime 7/30 12:31`, 1451 missed runs). Research still + worked because CDP attach quietly failed and the runner fell back to its local-launch path. +2. **A stale endpoint file pointed at 9222.** `~/.claude/config/session_keeper.cdp` was last + written **2026-08-02 06:34** and records `pid 175804` — a PID that has not existed since the + reboot. Nothing ever invalidated it. +3. **Another app took port 9222.** `~/.claude/browser-relay/relay.mjs` (the `/takeover` phone + relay) launches Chrome with `--remote-debugging-port=9222` on + `--user-data-dir=C:\Temp\igx-cdp-profile`. It started **12:57:20 PT**, seven minutes after + the ~12:50 reboot. + +From 12:57 onward the stale `.cdp` file's health check *passed* — against the relay's Chrome. +Every run then attached to `contexts[0]` of a throwaway profile with **zero Perplexity +cookies**, navigated to a logged-out wall, and was torn down ~128s later: +`Page.wait_for_timeout: Target page, context or browser has been closed`. + +**The timeline is exact:** last success **11:49** (pre-reboot, launch path). Relay claims 9222 +at **12:57**. First failure **13:19**. Four consecutive failures, all ~128s. + +This also explains the two misleading secondary symptoms: +- **`BROWSER_BUSY` with `active: null`** — the semaphore is released early for CDP-attached + runs, so slots leak from runs that die inside the doomed attach. +- **"a dead browser that killed two other sessions' queries"** — on attach the runner closes + every `/search/` page in the shared context, so each new doomed run tore down the pages of + the other doomed runs sharing the relay's Chrome. + +The reboot was a trigger, not the cause. The cause is that a *reachable* port was treated as a +*trusted* port. + +## Fix + +### 1. CDP identity gate (`council_browser.py`) — the actual fix + +`_start_via_cdp()` now proves the endpoint is the keeper's Chrome before attaching. Layered so +an indeterminate probe degrades to the next check rather than blocking a healthy keeper: + +- **Gate A — recorded PID liveness.** A dead `pid` in `session_keeper.cdp` means the file is a + leftover from a previous boot. The stale file is deleted so later runs re-probe cleanly. +- **Gate B — `DevToolsActivePort` match.** Chrome writes `\n` into its own + `--user-data-dir`. We compare that GUID against the endpoint's reported + `webSocketDebuggerUrl`. This is the canonical launch-time↔discovery-time match; portable and + decisive. (Independently confirmed by Perplexity research during this fix — skipping it is a + known SSRF-style attack vector against CDP clients, not just an ops bug.) +- **Gate C — port-owner command line.** If no `DevToolsActivePort` exists, the process + listening on the port must have `session_keeper_profile` in its command line. +- **Gate D — post-attach cookie check.** The attached context must carry a Perplexity auth + cookie (`__Secure-next-auth.session-token` / `pplx.session-id`). Backstop when process + introspection is unavailable, and it *also* catches a live keeper whose login has expired + (hypothesis 2 in the evidence file) — previously indistinguishable from this failure. + +The "synthesize a `.cdp` file because 9222 answers" fast path — the specific line that turned a +squatter into a false positive — is now gated behind the same ownership check. + +Any refusal falls back to the local-launch path, which is what was working before 12:57. + +### 2. Orphaned-result recovery (`council_query.py`) + +`invocation_id` is a fresh UUID per call, so when a caller dies the completed result on disk +becomes unreachable — the work succeeded, only delivery was lost +(`council_fae5c9e0.json`, `intellegix-business-projections`). + +Results now carry a `query_fingerprint` (sha256 of mode+context+query). `run_browser_query` +checks the cache for a matching, usable, recent result before spending minutes of browser +automation. Details: + +- Age is computed from the result's **recorded `timestamp`**, not file mtime — recovering an + orphan re-saves it under a new invocation id, which would otherwise reset mtime and let one + stale result be re-served forever. +- Window: 2h (`ORPHAN_RESULT_MAX_AGE_S`). Scan bounded to 200 newest files. +- Error stubs and empty syntheses are never served (`_result_is_usable`). +- Pre-fingerprint results already on disk are matched on stored query text, so existing + orphans are recoverable. +- Opt out with `COUNCIL_NO_ORPHAN_RECOVERY=1`. + +## Verification + +Queue log, `~/.claude/council-logs/perplexity-queue.json` — patch landed 14:07:55 PT: + +| seq | time | duration | result | +|---|---|---|---| +| 905 | 13:50 | 128.2s | ❌ browser closed | +| 906 | 14:01 | 128.1s | ❌ browser closed | +| **908** | **14:08** | **159.5s** | ✅ **completed** (first post-patch run) | + +- Gate unit-tested against live machine state: correctly **refuses** the relay's Chrome by both + stale-PID and foreign-owner paths; correctly **refuses** wrong-GUID and wrong-port; correctly + **accepts** a simulated healthy keeper (so the CDP path is not broken for when the keeper + returns). +- Orphan recovery tested against the real `council_fae5c9e0.json`: recovers the identical + synthesis, respects the 2h window, is immune to mtime reset, and returns `None` for an unseen + query. +- End-to-end through the MCP tool: a repeated query returned **instantly** from cache; new + queries run and complete normally. + +## Follow-up worth owning (NOT changed here) + +1. **The keeper task is still Disabled** (since 07-30). Research works without it via the + launch path. Re-enable deliberately, not as part of this fix. +2. **Port 9222 is contended.** `session_keeper.py` and `browser-relay/relay.mjs` both want it, + and the relay's header documents 9222 as its takeover contract. If the keeper is ever + re-enabled while the relay is up, the keeper will fail to bind. Per the port-registry rules + this pair needs an explicit assignment — it is a real collision, currently latent. +3. **`BROWSER_BUSY` vs `BROWSER_DEAD`** are still not distinguished in the caller-facing error. + The gate removes the cause that made it misleading today, but the semaphore can still leak a + slot on a crashed run and report contention that does not exist. diff --git a/council-automation/council_browser.py b/council-automation/council_browser.py index b30d0d7..26ec4a1 100644 --- a/council-automation/council_browser.py +++ b/council-automation/council_browser.py @@ -21,7 +21,9 @@ from pathlib import Path import shutil +import subprocess import tempfile +import urllib.parse from council_config import ( BROWSER_HEADLESS, @@ -91,6 +93,20 @@ class BrowserBusyError(Exception): pass +class SessionStaleError(Exception): + """Raised when critical Perplexity cookies are EXPIRED and no refresh path + could renew them, so submitting the query would be knowingly doomed. + + Replaces the old `keeper-timeout-stale-proceed` behaviour. Proceeding on + expired cookies converted a clean, diagnosable "the session is dead" into a + 5-8 minute mid-run death that also consumed a slot on a serialized queue + (2026-08-08 outage: two consecutive runs burned ~9 minutes each this way). + Failing here costs seconds and names the fix. Set COUNCIL_ALLOW_STALE=1 to + restore the old proceed-anyway behaviour. + """ + pass + + class SessionSemaphore: """File-based named-slot semaphore for concurrent browser sessions. @@ -293,6 +309,31 @@ def __exit__(self, *args): # ~/.claude/mcp-servers/browser-bridge/server.js. Don't grep-and-hunt. PERPLEXITY_COMMIT_KEY = "Space" +# Basename of the profile dir session_keeper.py launches Chrome with (see +# session_keeper.py: keeper_profile_dir). Used to prove a CDP endpoint really +# belongs to the keeper and not to another app squatting the same port. +KEEPER_PROFILE_DIRNAME = "session_keeper_profile" +KEEPER_PROFILE_DIR = Path.home() / ".claude" / "config" / KEEPER_PROFILE_DIRNAME + +# CDP port the keeper owns. Historically 9222 -- which is ALSO the port used by +# the Chrome on the C:\Temp\igx-cdp-profile profile (launched by the Dogfood +# Supervisor, start-dogfood.cmd, at logon; the profile name is browser-relay's, +# which is why it was long misattributed to relay.mjs / /takeover). It auto- +# starts and holds the port, so the keeper simply lost. That produced the 08-07 +# outage (runner attached to the relay's cookie-less Chrome) and again the +# 2026-08-08 outage (relay held 9222, so the freshness guard below believed +# "the keeper is alive" and fired a refresh that could never land). The keeper +# now gets a DEDICATED port so the two can never contend. Override with +# COUNCIL_KEEPER_CDP_PORT if 9223 ever conflicts. See memory port-registry. +KEEPER_CDP_PORT = int(os.environ.get("COUNCIL_KEEPER_CDP_PORT", "9223")) + +# Cookies that only exist in a browser profile actually logged in to Perplexity. +# Their absence proves the attached CDP context is not a usable keeper browser. +PERPLEXITY_AUTH_COOKIES = { + "__Secure-next-auth.session-token", + "pplx.session-id", +} + def _log(msg: str) -> None: """Log to stderr (stdout reserved for JSON result). @@ -832,6 +873,246 @@ async def start(self) -> None: else: await self._start_non_persistent() + @staticmethod + def _cdp_port_owner_cmdline(port: int) -> str | None: + """Return the command line of the process LISTENING on `port`, or None. + + Windows-only (uses Get-NetTCPConnection + Win32_Process). Returns None + when the owner cannot be determined — callers must treat None as + "unknown", never as "not the keeper", so a failed probe degrades to the + post-attach cookie gate rather than blocking a healthy keeper. + """ + if sys.platform != "win32": + return None + ps = ( + f"$c = Get-NetTCPConnection -LocalPort {port} -State Listen " + f"-ErrorAction SilentlyContinue | Select-Object -First 1; " + f"if ($c) {{ (Get-CimInstance Win32_Process -Filter " + f"\"ProcessId=$($c.OwningProcess)\").CommandLine }}" + ) + try: + out = subprocess.run( + ["powershell", "-NoProfile", "-Command", ps], + capture_output=True, text=True, timeout=15, + ) + except Exception as e: + _log(f"CDP owner probe failed ({type(e).__name__}: {e}) — owner unknown") + return None + cmdline = (out.stdout or "").strip() + return cmdline or None + + @classmethod + def _cdp_endpoint_is_keeper(cls, endpoint: str, recorded_pid: int | None) -> tuple[bool, str]: + """Verify a live CDP endpoint actually belongs to the session keeper. + + A reachable CDP port is NOT proof the keeper is alive. Port 9222 is also + used by `~/.claude/browser-relay/relay.mjs` (the /takeover phone relay), + which launches Chrome on a throwaway `C:\\Temp\\igx-cdp-profile` with no + Perplexity cookies. On 2026-08-07 the keeper had been disabled since + 07-30 while a *stale* session_keeper.cdp from 08-02 still pointed at + 9222; when the relay claimed that port at 12:57 PT, every research run + attached to the relay's cookie-less Chrome and died ~128s later with + "Target page, context or browser has been closed". Four consecutive + failures, machine-wide research outage. + + Returns (is_keeper, reason). `is_keeper=False` means "definitely not the + keeper — do not attach". An indeterminate probe returns True with a + reason noting the uncertainty; the post-attach cookie gate is the + backstop for that case. + """ + # Gate ordering matters, and getting it wrong is not a safe failure. + # + # The recorded-PID check used to run FIRST and veto everything else. But + # the keeper is one-shot per invocation: it launches Chrome DETACHED and + # exits within seconds, so the pid it published was ALWAYS dead by the + # time a reader looked (it published its own pid, not Chrome's — fixed + # in session_keeper.py). Every healthy keeper was therefore judged a + # stale .cdp file, the file was deleted, and every run fell back to a + # local launch. That is how the 08-08 outage reached the local-launch + # freshness guard at all: an over-strict identity gate silently disabled + # the CDP path it was written to protect. + # + # So: run the DECISIVE gates first (DevToolsActivePort GUID, then port + # ownership). The recorded pid is the weakest signal — it describes the + # launcher, not the server — and now only breaks ties when neither + # decisive gate can reach a verdict. + port = KEEPER_CDP_PORT + try: + port = int(urllib.parse.urlparse(endpoint).port or KEEPER_CDP_PORT) + except Exception: + pass + + # Gate B — DevToolsActivePort match. Chrome writes "\n" + # into its own --user-data-dir at startup. Comparing that GUID against the + # webSocketDebuggerUrl the endpoint reports is the canonical way to prove + # "this endpoint is the Chrome launched with THAT profile" — any process + # can serve plausible JSON on a port, so the response alone proves nothing. + # Portable (no process introspection) and decisive when the file exists. + active_port_file = KEEPER_PROFILE_DIR / "DevToolsActivePort" + try: + lines = active_port_file.read_text(encoding="utf-8").splitlines() + except OSError: + lines = [] + if len(lines) >= 2 and lines[0].strip().isdigit(): + keeper_port = int(lines[0].strip()) + keeper_guid = lines[1].strip() + if keeper_port != port: + return False, ( + f"keeper's DevToolsActivePort says port {keeper_port}, " + f"but endpoint points at {port}" + ) + try: + import urllib.request as _ur + with _ur.urlopen(f"{endpoint}/json/version", timeout=3) as _r: + ws = json.loads(_r.read().decode()).get("webSocketDebuggerUrl", "") + except Exception as e: + return True, f"DevToolsActivePort read but /json/version probe failed ({type(e).__name__})" + if keeper_guid and keeper_guid in ws: + return True, f"DevToolsActivePort GUID matches endpoint on port {port}" + return False, ( + f"endpoint on port {port} reports a different DevTools GUID than the " + f"keeper's profile — a foreign Chrome is serving this port" + ) + + # Gate C — no DevToolsActivePort (keeper not running, or its profile was + # cleaned). Fall back to checking that the process listening on the port + # is running the keeper's profile dir. + cmdline = cls._cdp_port_owner_cmdline(port) + if cmdline is not None: + if KEEPER_PROFILE_DIRNAME in cmdline: + return True, f"port {port} owned by keeper profile ({KEEPER_PROFILE_DIRNAME})" + # Label the squatter accurately -- this string is what a human + # reads when deciding what to go kill, and getting it wrong sends + # them after the wrong process. The C:\Temp\igx-cdp-profile Chrome + # is LAUNCHED BY the Dogfood Supervisor (start-dogfood.cmd, auto- + # starts at logon), verified 2026-08-08 by walking the parent chain: + # chrome pid -> cmd.exe -> start-dogfood.cmd. The profile name comes + # from browser-relay, which is why it was long misattributed to + # relay.mjs / /takeover. Either way it is LOAD-BEARING and must not + # be killed: the supervisor relaunches a byte-identical Chrome + # within ~3s, and the phone-takeover bridge on :7070 depends on it. + foreign = ( + "Dogfood Supervisor (start-dogfood.cmd) -- load-bearing, do NOT kill" + if "igx-cdp-profile" in cmdline + else "unidentified app" + ) + return False, ( + f"port {port} is owned by a FOREIGN Chrome ({foreign}) — " + f"no '{KEEPER_PROFILE_DIRNAME}' in its command line" + ) + + # Gate D — last resort, and the ONLY place the recorded pid is consulted. + # Both decisive gates were indeterminate (no DevToolsActivePort file and + # the owner probe is unavailable, e.g. non-Windows), so fall back to + # "was the process that wrote this file still around". A dead pid here + # is genuine evidence of a leftover .cdp file from a previous boot. + if recorded_pid and recorded_pid > 0 and not cls._pid_alive(recorded_pid): + return False, ( + f"no identity proof available and recorded pid {recorded_pid} is " + f"dead — treating .cdp file as stale" + ) + return True, "port owner unknown (probe unavailable) — deferring to cookie gate" + + @classmethod + def _keeper_cdp_alive(cls) -> tuple[bool, str]: + """True only if the keeper's OWN Chrome is serving CDP on KEEPER_CDP_PORT. + + This is the identity-gated twin of the check `_start_via_cdp` performs. + It exists because the 2026-08-08 outage came from the *unguarded* copy of + this probe in `_ensure_fresh_session`: that one asked only "does the port + answer /json/version?", concluded the keeper was alive when the /takeover + relay held 9222, and so routed every refresh down the keeper branch -- + which then could not possibly land. Reachability is not identity; prove + ownership before believing a port belongs to us. + """ + endpoint = f"http://127.0.0.1:{KEEPER_CDP_PORT}" + try: + import urllib.request as _ur + with _ur.urlopen(f"{endpoint}/json/version", timeout=2) as _r: + _ = _r.read() + except Exception as e: + return False, f"no CDP on port {KEEPER_CDP_PORT} ({type(e).__name__})" + return cls._cdp_endpoint_is_keeper(endpoint, None) + + @staticmethod + def _keeper_task_state() -> str: + """State of the PerplexitySessionKeeper scheduled task. + + Returns one of "Ready" | "Running" | "Disabled" | "Missing" | "Unknown". + + Firing `Start-ScheduledTask` at a **Disabled** task is a silent no-op: + it neither errors usefully nor runs anything. The task had been Disabled + since 2026-07-30, so every `_ensure_fresh_session` keeper refresh since + then burned the full SESSION_KEEPER_WAIT_S and then proceeded stale -- + the 120s of dead wall-clock at the head of every doomed run on 08-08. + Check the state first and skip the branch when it cannot possibly work. + """ + if sys.platform != "win32": + return "Unknown" + try: + out = subprocess.run( + ["powershell", "-NoProfile", "-Command", + "$t = Get-ScheduledTask -TaskName 'PerplexitySessionKeeper' " + "-ErrorAction SilentlyContinue; " + "if ($t) { $t.State } else { 'Missing' }"], + capture_output=True, text=True, timeout=15, + ) + except Exception as e: + _log(f"Keeper task-state probe failed ({type(e).__name__}: {e})") + return "Unknown" + state = (out.stdout or "").strip() + return state or "Unknown" + + @staticmethod + def _pid_alive(pid: int) -> bool: + """True if `pid` names a live process. Windows-safe (see _is_session_keeper_running).""" + try: + os.kill(pid, 0) + return True + except PermissionError: + return True # exists but inaccessible + except (ProcessLookupError, OSError, SystemError): + if sys.platform == "win32": + try: + import ctypes + PROCESS_QUERY_LIMITED_INFORMATION = 0x1000 + h = ctypes.windll.kernel32.OpenProcess( + PROCESS_QUERY_LIMITED_INFORMATION, False, pid + ) + if h: + ctypes.windll.kernel32.CloseHandle(h) + return True + except Exception: + pass + return False + + async def _cdp_context_has_perplexity_session(self) -> bool: + """True if the attached CDP context carries a Perplexity login cookie. + + Backstop for `_cdp_endpoint_is_keeper` when the port-owner probe is + unavailable: a foreign/throwaway Chrome profile has zero perplexity.ai + cookies, so this separates "the keeper" from "somebody else's browser" + without depending on Windows process introspection. It also catches a + live keeper whose login has actually expired. + """ + try: + cookies = await self.context.cookies() + except Exception as e: + _log(f"CDP cookie gate: could not read cookies ({type(e).__name__}) — allowing attach") + return True # don't block on an unexpected Playwright error + names = { + c.get("name") + for c in cookies + if "perplexity.ai" in (c.get("domain") or "") + } + if names & PERPLEXITY_AUTH_COOKIES: + return True + _log( + f"CDP cookie gate: attached context has {len(names)} perplexity.ai cookie(s), " + f"none of {sorted(PERPLEXITY_AUTH_COOKIES)} — not a logged-in keeper browser" + ) + return False + async def _start_via_cdp(self) -> bool: """Attach to a running session_keeper.py via Chrome DevTools Protocol. @@ -858,21 +1139,36 @@ async def _start_via_cdp(self) -> bool: # removed (observed 2026-05-25 — user closed Chrome window, the file # got cleaned up by some path, but Chrome's child processes kept # serving the port). + # NOTE: "port 9222 answers" is NOT the same as "the keeper is up" — the + # /takeover browser-relay serves CDP on the same port. Only synthesize + # the endpoint file once the listening process is confirmed to be running + # the keeper's own profile, otherwise we manufacture a valid-looking + # endpoint pointing at somebody else's Chrome (the 2026-08-07 outage). if not cdp_file.exists(): try: import urllib.request as _ur - with _ur.urlopen("http://127.0.0.1:9222/json/version", timeout=2) as _r: + with _ur.urlopen(f"http://127.0.0.1:{KEEPER_CDP_PORT}/json/version", timeout=2) as _r: _ = _r.read() + owner = self._cdp_port_owner_cmdline(KEEPER_CDP_PORT) + if owner is not None and KEEPER_PROFILE_DIRNAME not in owner: + _log( + f"CDP-attach: port {KEEPER_CDP_PORT} is serving CDP but is NOT the keeper " + f"(no '{KEEPER_PROFILE_DIRNAME}' in owner cmdline) — refusing to " + "synthesize an endpoint file for a foreign browser" + ) + raise RuntimeError("foreign CDP owner") # Port is up — write a fresh CDP file so the rest of the path # can use it. The keeper would have written this normally. cdp_file.parent.mkdir(parents=True, exist_ok=True) cdp_file.write_text( - json.dumps({"port": 9222, "endpoint": "http://127.0.0.1:9222"}), + json.dumps({"port": KEEPER_CDP_PORT, + "endpoint": f"http://127.0.0.1:{KEEPER_CDP_PORT}"}), encoding="utf-8", ) - _log("CDP-attach: Chrome alive on 9222 but .cdp file missing — synthesized it") + _log(f"CDP-attach: Chrome alive on {KEEPER_CDP_PORT} but .cdp file " + f"missing — synthesized it") except Exception: - pass # port not reachable; fall through to keeper auto-start + pass # port not reachable / not the keeper; fall through # Auto-start the keeper task if CDP still isn't reachable. Avoids the # headful local-launch fallback that creates focus-stealing popups. @@ -915,6 +1211,23 @@ async def _start_via_cdp(self) -> bool: except Exception as e: _log(f"CDP-attach skipped: endpoint {endpoint_probe} unreachable ({type(e).__name__}); falling back to launch") return False + # Reachable is not enough — prove it is the keeper's Chrome. + # A stale .cdp file plus a foreign app on the same port is the + # exact combination that produced the 2026-08-07 outage. + recorded_pid = data.get("pid") + ok, why = self._cdp_endpoint_is_keeper( + endpoint_probe, + recorded_pid if isinstance(recorded_pid, int) else None, + ) + if not ok: + _log(f"CDP-attach REFUSED: {why} — falling back to local launch") + try: + cdp_file.unlink() + _log("Removed stale session_keeper.cdp so later runs re-probe cleanly") + except Exception: + pass + return False + _log(f"CDP identity OK: {why}") except Exception: pass # let the connect_over_cdp below try and produce its own error try: @@ -939,6 +1252,19 @@ async def _start_via_cdp(self) -> bool: self._browser = None return False self.context = contexts[0] + + # Backstop identity gate: a browser that is not logged in to + # Perplexity is useless to us, and a foreign/throwaway profile has + # no perplexity.ai cookies at all. Catches the wrong-browser case + # even when the Windows port-owner probe was unavailable, and also + # catches a genuinely expired keeper login — both of which used to + # present as a ~128s hang ending in "browser has been closed". + if not await self._cdp_context_has_perplexity_session(): + _log("CDP attach REFUSED by cookie gate — falling back to local launch") + self.context = None + self._browser = None # never .close() a CDP browser: kills remote Chrome + return False + self._cdp_attached = True existing_count = len(self.context.pages) _log(f"CDP attach OK: {len(contexts)} context(s); using contexts[0] with {existing_count} existing page(s)") @@ -1075,11 +1401,17 @@ async def _ensure_fresh_session(self, reason: str = "") -> None: Fires the keeper task (idempotent — Task Scheduler refuses a 2nd instance of the one-shot, so no lock is needed) and waits up to SESSION_KEEPER_WAIT_S - for the cookies-file mtime to bump. On timeout/failure it logs and PROCEEDS - with the current session — it never aborts, because an abort cascades into - the runner's SKIPPED-NETWORK-STREAK and a stale session still usually - succeeds. Because we CDP-attach to the same keeper Chrome that the keeper - refreshes in-place, the live attached session picks up the new cookies. + for the cookies-file mtime to bump, then falls back to refresh_session.py. + Because we CDP-attach to the same keeper Chrome that the keeper refreshes + in-place, the live attached session picks up the new cookies. + + Raises SessionStaleError if a critical cookie is still EXPIRED after both + refresh paths. This reverses the pre-2026-08-08 policy of always + proceeding: the old rationale ("a stale session still usually succeeds") + holds for cookies that are merely *expiring soon*, which is why only + hard-expired cookies abort here. It does not hold for expired ones — + those produced a Cloudflare wall and a guaranteed mid-run death 5-8 + minutes later, on a serialized queue, with no error ever logged. """ freshness = self._check_session_freshness(self.session_path) if hasattr(self, "_query_inst"): @@ -1106,19 +1438,39 @@ async def _ensure_fresh_session(self, reason: str = "") -> None: self._query_inst["auto_refresh_path"] = "no_refresh" return - # Prefer refreshing THROUGH the keeper task (no popup) when its Chrome is - # serving CDP; else fall back to refresh_session.py (headful popup). - chrome_alive = False - try: - import urllib.request as _ur - with _ur.urlopen("http://127.0.0.1:9222/json/version", timeout=2) as _r: - _ = _r.read() - chrome_alive = True - except Exception: - chrome_alive = False - - if chrome_alive: - _log("Keeper Chrome alive on CDP; firing PerplexitySessionKeeper to refresh in-place...") + # Prefer refreshing THROUGH the keeper task (no popup) when the KEEPER'S + # OWN Chrome is serving CDP; else fall back to refresh_session.py. + # + # Two gates, both added 2026-08-08 after this block caused a machine-wide + # research outage: + # + # 1. IDENTITY, not reachability. This used to be a bare urlopen against + # port 9222 -- the exact "a reachable port is a trusted port" mistake + # that `_start_via_cdp` was hardened against on 08-07. The /takeover + # relay held 9222, so `chrome_alive` came back True, the keeper branch + # was taken, and the refresh could never land. Note the two paths + # disagreed: `_start_via_cdp` correctly REFUSED the same port seconds + # earlier and fell back to a local launch. Fixing one call site and + # not its twin is what turned a closed bug back into an open one. + # + # 2. TASK RUNNABILITY. `Start-ScheduledTask` against a Disabled task is a + # silent no-op. PerplexitySessionKeeper had been Disabled since + # 2026-07-30, so the wait below could only ever time out. Every run + # paid SESSION_KEEPER_WAIT_S for nothing before proceeding doomed. + keeper_ok, keeper_why = self._keeper_cdp_alive() + task_state = self._keeper_task_state() if keeper_ok else "n/a" + keeper_usable = keeper_ok and task_state in ("Ready", "Running", "Unknown") + refreshed_via_keeper = False + + if keeper_ok and not keeper_usable: + _log(f"Keeper Chrome present ({keeper_why}) but its scheduled task is " + f"{task_state} — firing it would be a no-op; using refresh_session.py") + elif not keeper_ok: + _log(f"Keeper CDP not usable: {keeper_why} — using refresh_session.py") + + if keeper_usable: + _log(f"Keeper verified on CDP ({keeper_why}); task={task_state}; " + f"firing PerplexitySessionKeeper to refresh in-place...") if hasattr(self, "_query_inst"): self._query_inst["auto_refresh_path"] = "keeper_task" try: @@ -1138,21 +1490,55 @@ async def _ensure_fresh_session(self, reason: str = "") -> None: mtime_now = 0.0 if mtime_now > mtime_before: _log("Keeper refreshed cookies (mtime bumped); session fresh") - return + refreshed_via_keeper = True + break _time.sleep(1) - # Most likely the keeper's own ~20-min auto-cycle is mid-run, so - # Task Scheduler refused our 2nd instance and no bump occurred. - _log(f"WARN: keeper-timeout-stale-proceed (no mtime bump in " - f"{SESSION_KEEPER_WAIT_S}s) — proceeding with current session") + if not refreshed_via_keeper: + # Most likely the keeper's own ~20-min auto-cycle is mid-run, + # so Task Scheduler refused our 2nd instance. Do NOT proceed + # on stale cookies -- fall through to the direct refresher. + _log(f"WARN: keeper did not bump cookie mtime in " + f"{SESSION_KEEPER_WAIT_S}s — falling back to refresh_session.py") except Exception as e: - _log(f"Keeper-task refresh failed: {type(e).__name__}: {e} — proceeding") - else: - _log("No keeper Chrome on CDP; auto-refresh ON — invoking refresh_session.py ...") + _log(f"Keeper-task refresh failed: {type(e).__name__}: {e} — " + f"falling back to refresh_session.py") + + if not refreshed_via_keeper: if hasattr(self, "_query_inst"): - self._query_inst["auto_refresh_path"] = "refresh_session_fallback" + self._query_inst["auto_refresh_path"] = ( + "keeper_then_refresh_session" if keeper_usable else "refresh_session_fallback" + ) + _log("Auto-refresh ON — invoking refresh_session.py ...") refreshed = await self._auto_refresh_session() - _log("Auto-refresh succeeded" if refreshed - else "WARNING: auto-refresh failed — proceeding with stale session") + _log("Auto-refresh succeeded" if refreshed else "WARNING: auto-refresh failed") + + # Fail loud rather than submitting a doomed run. Only HARD-EXPIRED + # critical cookies abort: `expiring_soon` is a soft signal (the guard + # fires at SESSION_FRESHNESS_THRESHOLD_S = 8 min of remaining TTL, and a + # session with 7 minutes left still answers fine), whereas an expired + # pplx.session-id / __cf_bm means Cloudflare will serve a bot challenge + # Playwright cannot pass and the run WILL die -- just 5-8 minutes later, + # after it has consumed a slot on the serialized queue. + post = self._check_session_freshness(self.session_path) + still_expired = post.get("stale_critical") or [] + if hasattr(self, "_query_inst"): + self._query_inst["cookies_stale_critical"] = still_expired + if not still_expired: + return + if os.environ.get("COUNCIL_ALLOW_STALE", "").lower() in ("1", "true", "yes", "on"): + _log("COUNCIL_ALLOW_STALE=1 — proceeding on expired cookies anyway") + return + detail_expired = ", ".join(f"{n}({age}m ago)" for n, age in still_expired) + _log(f"MONITOR-SIGNAL session_stale_abort {detail_expired}") + raise SessionStaleError( + f"Perplexity session cookies are EXPIRED and could not be refreshed " + f"[{detail_expired}]. Refresh path tried: " + f"keeper={'usable' if keeper_usable else f'unusable ({keeper_why}, task={task_state})'}, " + f"refresh_session.py=failed. Aborting before submit instead of burning a " + f"queue slot on a run that cannot succeed. Fix: run /cache-perplexity-session, " + f"or check that PerplexitySessionKeeper is Enabled and owns CDP port " + f"{KEEPER_CDP_PORT}." + ) async def _load_session(self) -> None: """Load session from playwright-session.json + playwright-localstorage.json. @@ -1517,6 +1903,24 @@ async def _verify_council_activation(self, page) -> bool: Tier 2: text-scan for 'Model council' (tolerates DOM drift). Both miss → SELECTOR_DRIFT_DETECTED, return False (was: optimistic True). """ + # Tier 0 (2026-07-22): aria-label + aria-pressed on the icon-only mode + # button (same Perplexity redesign that broke the research verifier). + try: + aria_found = await page.evaluate("""() => { + const els = document.querySelectorAll('[aria-label]'); + for (const el of els) { + const label = (el.getAttribute('aria-label') || '').trim().toLowerCase(); + if ((label === 'model council' || label === 'council') + && el.getAttribute('aria-pressed') === 'true') return true; + } + return false; + }""") + if aria_found: + _log("activate_mode verify=OK mode=council indicator=tier0_aria_pressed") + return True + except Exception as _e0: + _log(f"activate_mode verify=tier0_ERROR mode=council exception={_e0!r}") + # Tier 1: stable aria-label selector try: three_models = self.selectors.get("threeModelsDropdown", "button[aria-label='3 models']") @@ -1545,12 +1949,35 @@ async def _verify_council_activation(self, page) -> bool: async def _verify_research_activation(self, page) -> bool: """Verify Research mode activated via 2-tier selector cascade. + Tier 0 (2026-07-22): aria-label + aria-pressed on the toolbar mode + button. Perplexity moved the indicator from a TEXT pill to an + ICON-ONLY button (aria-label="Deep research", aria-pressed="true") + with EMPTY textContent, so the tier1/tier2 text scans below can no + longer see it and falsely report SELECTOR_DRIFT. Verified via live DOM + probe. aria-pressed is the reliable active-state discriminator. Tier 1: exact-text match for the activated mode pill ("Deep research" or "Research" exactly). Catches the canonical activated state. Tier 2: case-insensitive contains scan for 'deep research' or exact 'research'. Tolerates minor Perplexity UI tweaks. - Both miss → SELECTOR_DRIFT_DETECTED, return False (was: optimistic True). + All miss → SELECTOR_DRIFT_DETECTED, return False (was: optimistic True). """ + # Tier 0: aria-label + aria-pressed on the icon-only mode button + try: + aria_found = await page.evaluate("""() => { + const els = document.querySelectorAll('[aria-label]'); + for (const el of els) { + const label = (el.getAttribute('aria-label') || '').trim().toLowerCase(); + if ((label === 'deep research' || label === 'research') + && el.getAttribute('aria-pressed') === 'true') return true; + } + return false; + }""") + if aria_found: + _log("activate_mode verify=OK mode=research indicator=tier0_aria_pressed") + return True + except Exception as e: + _log(f"activate_mode verify=tier0_ERROR mode=research exception={e!r}") + # Tier 1: exact-text match on the toolbar mode pill try: primary_found = await page.evaluate("""() => { @@ -1591,10 +2018,28 @@ async def _verify_research_activation(self, page) -> bool: async def _verify_labs_activation(self, page) -> bool: """Verify Labs mode activated via 2-tier selector cascade. + Tier 0 (2026-07-22): aria-label + aria-pressed on the icon-only mode + button (same Perplexity redesign that broke the research verifier). Tier 1: exact-text 'Labs' on a toolbar pill. Tier 2: case-insensitive contains 'labs' (looser fallback). - Both miss → SELECTOR_DRIFT_DETECTED, return False (was: optimistic True). + All miss → SELECTOR_DRIFT_DETECTED, return False (was: optimistic True). """ + # Tier 0: aria-label + aria-pressed on the icon-only mode button + try: + aria_found = await page.evaluate("""() => { + const els = document.querySelectorAll('[aria-label]'); + for (const el of els) { + const label = (el.getAttribute('aria-label') || '').trim().toLowerCase(); + if (label === 'labs' && el.getAttribute('aria-pressed') === 'true') return true; + } + return false; + }""") + if aria_found: + _log("activate_mode verify=OK mode=labs indicator=tier0_aria_pressed") + return True + except Exception as e: + _log(f"activate_mode verify=tier0_ERROR mode=labs exception={e!r}") + # Tier 1: exact-text match on the toolbar mode pill try: primary_found = await page.evaluate("""() => { @@ -3068,7 +3513,24 @@ async def _run_impl(self, query: str) -> dict: try: _log("Starting Playwright browser...") - await self.start() + try: + await self.start() + except SessionStaleError as e: + # Abort cleanly instead of submitting a run that cannot succeed. + # Surfaced to the caller as a coded error (same shape as + # BROWSER_BUSY) so the MCP layer and the queue both see a real + # failure rather than a silent 5-8 minute death. + self._query_inst["chrome_path_used"] = "aborted_session_stale" + self._query_inst["exit_reason"] = "session_stale" + if not self._query_inst_emitted: + _emit_query_instrumentation(self._query_inst) + self._query_inst_emitted = True + _log(f"ABORT session_stale: {e}") + return { + "error": str(e), + "code": "SESSION_STALE", + "step": "session_freshness", + } # CDP-attached sessions share the keeper's Chrome — no per-session # browser was launched, so the semaphore slot isn't actually diff --git a/council-automation/council_query.py b/council-automation/council_query.py index 3d0941e..e8a2103 100644 --- a/council-automation/council_query.py +++ b/council-automation/council_query.py @@ -19,6 +19,7 @@ import argparse import asyncio +import hashlib import io import json import os @@ -504,6 +505,123 @@ def run_opus_synthesis( } +# How recently a completed-but-undelivered result may be reused instead of +# re-running the query. Long enough to cover a session dying and being restarted +# (the 2026-08-07 intellegix-business-projections case), short enough that +# genuinely stale research is never silently served. +ORPHAN_RESULT_MAX_AGE_S = 2 * 60 * 60 + +# Cap on how many cache files one lookup inspects (newest-first). +ORPHAN_SCAN_LIMIT = 200 + + +def query_fingerprint(query: str, context: str, mode: str) -> str: + """Stable content hash identifying one logical query. + + `invocation_id` is a fresh UUID per call, so a result whose caller died + before reading it becomes unreachable — the work succeeded but the only + pointer to it was lost. Fingerprinting by content lets a re-run of the same + query find that orphan instead of burning another few minutes reproducing it. + """ + payload = f"{mode}\x00{context}\x00{query}" + return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16] + + +def _result_is_usable(data: dict) -> bool: + """True if a cached result actually carries a synthesis (not an error stub).""" + if not isinstance(data, dict) or data.get("error"): + return False + synthesis = data.get("synthesis") or {} + if synthesis.get("error"): + return False + return bool(synthesis.get("response") or synthesis.get("parsed")) + + +def _result_age_s(data: dict, path: Path, now: float) -> float | None: + """Seconds since the result was PRODUCED, preferring its recorded timestamp.""" + ts = data.get("timestamp") + if isinstance(ts, str): + try: + produced = datetime.fromisoformat(ts.replace("Z", "+00:00")) + if produced.tzinfo is None: + produced = produced.replace(tzinfo=timezone.utc) + return now - produced.timestamp() + except ValueError: + pass + try: + return now - path.stat().st_mtime + except OSError: + return None + + +def find_orphaned_result( + query: str, + context: str, + mode: str, + max_age_s: float = ORPHAN_RESULT_MAX_AGE_S, +) -> dict | None: + """Return a recent completed result for this exact query, or None. + + Scans `council_*.json` in CACHE_DIR for a usable result whose fingerprint + matches. Results written before fingerprints existed are matched by + recomputing the hash from their stored `query` field, so orphans that + already exist on disk are recoverable too. + """ + try: + fp = query_fingerprint(query, context, mode) + except Exception: + return None + now = time.time() + best: tuple[float, dict] | None = None + try: + # Newest-first, bounded: the cache accumulates indefinitely and we only + # ever care about the last couple of hours. + candidates = sorted( + CACHE_DIR.glob("council_*.json"), + key=lambda p: p.stat().st_mtime, + reverse=True, + )[:ORPHAN_SCAN_LIMIT] + except OSError: + return None + for path in candidates: + if path.name == "council_latest.json": + continue + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + continue + # Age from the RESULT's own timestamp, not the file's mtime: recovering + # an orphan re-saves it under a new invocation id, which would otherwise + # reset mtime and let one stale result be re-served indefinitely. + age = _result_age_s(data, path, now) + if age is None or age > max_age_s: + continue + if not _result_is_usable(data): + continue + stored_fp = data.get("query_fingerprint") + if stored_fp is None: + # Pre-fingerprint result — recompute from what it recorded. `context` + # is not stored, so fall back to matching on the query text alone. + if data.get("query") != query: + continue + elif stored_fp != fp: + continue + if best is None or age < best[0]: + best = (age, data) + if best is None: + return None + age, data = best + print( + f"Orphaned-result recovery: reusing completed result for this exact query " + f"({int(age)}s old) instead of re-running.", + file=sys.stderr, + ) + data = dict(data) + data["recovered_from_cache"] = True + data["recovered_age_s"] = int(age) + return data + + def save_results(results: dict, invocation_id: str | None = None) -> Path: """Save results to cache and history. @@ -724,6 +842,16 @@ async def run_browser_query( fallback_log: list[dict] = [] full_query = f"{context}\n\n{query}" if context else query + # Before spending minutes of browser automation, check whether this exact + # query already completed and its result was simply never delivered (caller + # died / session restarted). Opt out with COUNCIL_NO_ORPHAN_RECOVERY=1. + if os.environ.get("COUNCIL_NO_ORPHAN_RECOVERY", "").strip() not in ("1", "true", "True"): + orphan = find_orphaned_result(query, context, perplexity_mode) + if orphan is not None: + # Re-save under this invocation so the caller can read it normally. + save_results(orphan, invocation_id=invocation_id) + return orphan + # Use config default (BROWSER_HEADLESS=False, i.e. headful) unless explicitly overridden kwargs = {"save_artifacts": opus_synthesis, "perplexity_mode": perplexity_mode} if headful is not None: @@ -890,6 +1018,9 @@ async def run_browser_query( error_fallbacks = [f for f in fallback_log if f.get("severity") != "info"] results = { "query": query, + # Content hash of (mode, context, query) so a result whose caller died + # before reading it can still be matched by a re-run (find_orphaned_result). + "query_fingerprint": query_fingerprint(query, context, perplexity_mode), "timestamp": datetime.now(timezone.utc).isoformat(), "mode": "browser", "models": models_dict, @@ -940,6 +1071,10 @@ def format_synthesis_output(results: dict) -> str: f"**Code:** {code}\n" f"**Step:** {step}\n\n" f"If BROWSER_BUSY: another session is using Playwright. Wait ~2 min.\n" + f"If SESSION_STALE: Perplexity cookies are expired and auto-refresh " + f"could not renew them. Run `/cache-perplexity-session`, then retry. " + f"This aborts BEFORE submitting, so no queue slot was wasted — do not " + f"treat it as a transient failure to retry blindly.\n" f"If session expired: run `python council_browser.py --save-session`\n" ) diff --git a/council-automation/queue_monitor.py b/council-automation/queue_monitor.py index 97fd935..c1368f9 100644 --- a/council-automation/queue_monitor.py +++ b/council-automation/queue_monitor.py @@ -181,7 +181,14 @@ def evaluate( seen_in_batch: Set[str] = set() for ev in recent_events or []: event = ev.get("event") - if event not in ("error", "timeout"): + # `dropped` (added 2026-08-08) is the reclaim record research_queue + # writes for a run whose process was killed without reporting an + # outcome -- the dominant real-world failure mode, since the MCP layer + # hard-kills the runner on timeout and Windows TerminateProcess runs no + # finally-blocks. It was invisible here for as long as it was invisible + # in errors_today, which is why two consecutive machine-wide outages + # were both found by hand instead of by this monitor. + if event not in research_queue.FAILURE_EVENTS: continue run_id = ev.get("run_id", "?") dedupe_key = f"{run_id}:{event}" @@ -190,7 +197,7 @@ def evaluate( seen_in_batch.add(dedupe_key) session = ev.get("session", "?") detail = ev.get("error") or ev.get("query_preview") or "" - severity = "critical" if event == "error" else "warning" + severity = "warning" if event == "timeout" else "critical" alerts.append( Alert( key=f"event:{dedupe_key}", diff --git a/council-automation/research_monitor.py b/council-automation/research_monitor.py new file mode 100644 index 0000000..eea2701 --- /dev/null +++ b/council-automation/research_monitor.py @@ -0,0 +1,654 @@ +"""research_monitor.py — active monitoring sidecar for Perplexity-browser jobs. + +Tails the instrumentation / ledger / log sinks of a DETACHED extended-research +run (or an autonomous ``loop_driver`` run) in real time, distills health into a +single ``monitor-status.json``, fires Pushover on state transitions, and +auto-triggers the session keeper when a critical cookie's TTL is too short to +outlast the run. + +Design contract: +- **Standalone + fail-open.** Any internal error is logged and swallowed; the + monitor NEVER raises into the parent and NEVER kills the runner. +- **Read-only on runner state.** It writes only its own ``monitor-status.json`` + (+ ``monitor.log``) and an ADVISORY ``abort_recommended`` field. It never + signals the runner to stop — the runner's own circuit breakers own that. +- **Thresholds sit one below the runner's** so the monitor WARNS before the + runner's breaker fires (no race, no masking). + +Usage: + python research_monitor.py --workdir --kind extended-research \ + [--pid ] [--expected-run-min N] [--interval S] \ + [--once] [--no-pushover] [--no-remediate] + +``--once`` performs a single evaluation pass over the current files and exits +(used by the unit tests); the default mode loops until the run is terminal. +""" +from __future__ import annotations + +import argparse +import json +import os +import subprocess +import sys +import time +import urllib.error +import urllib.parse +import urllib.request +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +# --------------------------------------------------------------------------- paths + +HOME = Path.home() +CACHE_DIR = HOME / ".claude" / "council-cache" +GLOBAL_QUERY_INST = CACHE_DIR / "instrumentation-query.jsonl" +SESSION_PATH = HOME / ".claude" / "config" / "playwright-session.json" +PUSHOVER_CONFIG = HOME / ".claude" / "config" / "pushover.json" + +# --------------------------------------------------------------------------- thresholds +# Mirror extended_research_runner.py constants, but the streak WARN sits one +# below the runner's ABORT so we warn before the breaker fires. + +HEARTBEAT_STALE_S = 300 # runner HEARTBEAT_STALE_THRESHOLD_S +PARSE_FAIL_STREAK_ABORT = 3 # runner PARSE_FAIL_STREAK_THRESHOLD +SKIPPED_NETWORK_STREAK_ABORT = 3 # runner SKIPPED_NETWORK_STREAK_THRESHOLD +STREAK_WARN = 2 # one below the abort threshold +EMPTY_RETURN_LOUD = 2 # consecutive empty returns → critical +REASONING_TRAIL_WINDOW = 10 # evaluate rate over last N queries (was 5 — too noisy) +REASONING_TRAIL_RATE = 0.4 # >40% reasoning-trail-only → loud +COOKIE_TTL_FLOOR_MIN = 10.0 # never let critical cookie TTL drop below this +KEEPER_COOLDOWN_S = 300.0 # don't re-fire the keeper within this window +KEEPER_THRASH_WINDOW_S = 900.0 # look-back window for keeper-thrash detection +KEEPER_THRASH_COUNT = 3 # keeper fired this many times in window + still short → upstream problem +ABORT_PERSIST_TICKS = 3 # advisory abort persists this many ticks → write handshake flag +CRITICAL_COOKIES = ("__cf_bm", "pplx.edge-sid", "pplx.session-id") + +LEDGER_TERMINAL_STATUSES = {"COMPLETED", "ABORTED", "ABORTED-NETWORK", "INTERRUPTED"} +DEGRADED_CDP_PATHS = {"cdp_busy", "local_persistent", "local_nonpersistent"} + +STATUS_SCHEMA = 1 + + +# --------------------------------------------------------------------------- log + +def _now_iso() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _log(workdir: Path, line: str) -> None: + """Append one timestamped line to ``{workdir}/monitor.log``. Never raises.""" + try: + path = workdir / "monitor.log" + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("a", encoding="utf-8") as f: + f.write(f"{_now_iso()} {line}\n") + except OSError: + pass + + +# --------------------------------------------------------------------------- io helpers + +def _read_jsonl_tail(path: Path, offset: int) -> tuple[list[dict[str, Any]], int]: + """Read JSONL records appended since byte ``offset``. + + Returns ``(records, new_offset)``. Drops a trailing partial line by only + advancing the offset past the last complete newline. Fail-open: returns + ``([], offset)`` on any error. + """ + try: + if not path.exists(): + return [], offset + size = path.stat().st_size + if size < offset: + # File was truncated (e.g. 10 MB tail-rotate) — restart from 0. + offset = 0 + if size == offset: + return [], offset + with path.open("rb") as f: + f.seek(offset) + data = f.read() + except OSError: + return [], offset + last_nl = data.rfind(b"\n") + if last_nl == -1: + return [], offset # no complete line yet + complete = data[: last_nl + 1] + new_offset = offset + len(complete) + records: list[dict[str, Any]] = [] + for raw in complete.decode("utf-8", errors="replace").splitlines(): + raw = raw.strip() + if not raw: + continue + try: + obj = json.loads(raw) + except json.JSONDecodeError: + continue + if isinstance(obj, dict): + records.append(obj) + return records, new_offset + + +def _read_json(path: Path) -> dict[str, Any] | None: + try: + obj = json.loads(path.read_text(encoding="utf-8")) + return obj if isinstance(obj, dict) else None + except (OSError, json.JSONDecodeError): + return None + + +def _parse_ts(value: Any) -> float | None: + """Parse an ISO-ish timestamp to epoch seconds. None on failure.""" + if not isinstance(value, str) or not value: + return None + txt = value.strip().replace("Z", "+00:00") + try: + return datetime.fromisoformat(txt).timestamp() + except ValueError: + return None + + +def _pid_alive(pid: int | None) -> bool: + if not pid: + return True # unknown → assume alive, rely on other terminal signals + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except PermissionError: + return True + except OSError: + return False + return True + + +# --------------------------------------------------------------------------- cookies + +def _critical_cookie_ttl() -> dict[str, Any]: + """Compute the minimum remaining TTL (minutes) across critical cookies. + + Returns ``{min_critical_minutes, stale_critical: [(name, minutes_left)]}``. + ``min_critical_minutes`` is None when no critical cookie carries an expiry. + Mirrors council_browser._check_session_freshness's cookie format. + """ + out: dict[str, Any] = {"min_critical_minutes": None, "stale_critical": []} + try: + data = json.loads(SESSION_PATH.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return out + if not isinstance(data, list): + return out + now = time.time() + min_left: float | None = None + for cookie in data: + if not isinstance(cookie, dict): + continue + if cookie.get("name") not in CRITICAL_COOKIES: + continue + exp = cookie.get("expires", -1) + if not isinstance(exp, (int, float)) or exp <= 0: + continue # session cookie / no expiry + minutes_left = (exp - now) / 60.0 + out["stale_critical"].append([cookie.get("name", ""), round(minutes_left, 1)]) + if min_left is None or minutes_left < min_left: + min_left = minutes_left + if min_left is not None: + out["min_critical_minutes"] = round(min_left, 1) + return out + + +def _trigger_keeper(workdir: Path) -> bool: + """Fire the PerplexitySessionKeeper scheduled task (idempotent). Never raises.""" + try: + subprocess.run( + ["powershell", "-NoProfile", "-Command", + "Start-ScheduledTask -TaskName 'PerplexitySessionKeeper'"], + capture_output=True, timeout=20, check=False, + ) + _log(workdir, "REMEDIATE triggered PerplexitySessionKeeper") + return True + except (OSError, subprocess.SubprocessError) as exc: + _log(workdir, f"REMEDIATE keeper trigger failed: {exc!r}") + return False + + +# --------------------------------------------------------------------------- pushover + +def _send_pushover(workdir: Path, title: str, body: str, priority: int) -> bool: + """Minimal Pushover sender reusing ~/.claude/config/pushover.json. Fail-open.""" + try: + cfg = json.loads(PUSHOVER_CONFIG.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return False + token, user = cfg.get("app_token"), cfg.get("user_key") + if not token or not user: + return False + form = { + "token": token, "user": user, + "title": title[:250], "message": body[:1000], + "priority": str(priority), + "sound": cfg.get("default_sound", "incoming"), + } + data = urllib.parse.urlencode(form).encode() + req = urllib.request.Request("https://api.pushover.net/1/messages.json", + data=data, method="POST") + try: + with urllib.request.urlopen(req, timeout=5) as resp: + ok = json.loads(resp.read()).get("status") == 1 + _log(workdir, f"PUSHOVER sent ok={ok} title={title!r}") + return ok + except (urllib.error.URLError, TimeoutError, OSError, ValueError) as exc: + _log(workdir, f"PUSHOVER failed: {exc!r}") + return False + + +# --------------------------------------------------------------------------- state + +@dataclass +class MonitorState: + """Accumulated, bounded view of a run used to compute health each tick.""" + + workdir: Path + kind: str + pid: int | None + expected_run_min: float + start_ts: float + + inst_offset: int = 0 + query_offset: int = 0 + trace_offset: int = 0 + + pass_statuses: list[str] = field(default_factory=list) # trailing window + pass_empty: list[bool] = field(default_factory=list) # parallel: was the pass empty? + query_window: list[dict[str, Any]] = field(default_factory=list) + counters: dict[str, int] = field(default_factory=lambda: { + "empty_returns": 0, "reasoning_trail": 0, "parse_failed": 0, + "skipped_network": 0, "cdp_busy": 0, "local_fallback": 0, + "result_suspect": 0, + }) + passes_completed: int = 0 + last_event_ts: float = 0.0 + + keeper_last_fired: float = 0.0 + keeper_fire_history: list[float] = field(default_factory=list) # epoch times keeper fired + cookie_auto_refreshed: bool = False + cookie_last_refresh_iso: str | None = None + + abort_persist_ticks: int = 0 # consecutive ticks abort_recommended has been true + notified_keys: set[str] = field(default_factory=set) + last_health: str = "OK" + + +def _trailing_streak(statuses: list[str], target: str) -> int: + """Count trailing consecutive entries equal to ``target``.""" + count = 0 + for status in reversed(statuses): + if status == target: + count += 1 + else: + break + return count + + +# --------------------------------------------------------------------------- evaluate + +def evaluate(state: MonitorState, *, remediate: bool) -> dict[str, Any]: + """Ingest new records, update counters, and return the status dict.""" + wd = state.workdir + + # 1. Ingest per-run pass instrumentation (extended-research only). + inst_path = wd / "instrumentation.jsonl" + recs, state.inst_offset = _read_jsonl_tail(inst_path, state.inst_offset) + for rec in recs: + status = str(rec.get("status") or "") + is_empty = rec.get("raw_response_len_chars", 1) == 0 or status == "SKIPPED-NETWORK" + state.pass_statuses.append(status) + state.pass_empty.append(is_empty) + ts = _parse_ts(rec.get("ts") or rec.get("timestamp")) + if ts: + state.last_event_ts = max(state.last_event_ts, ts) + if status == "COMPLETED": + state.passes_completed += 1 + if status == "PARSE-FAILED": + state.counters["parse_failed"] += 1 + if status == "SKIPPED-NETWORK": + state.counters["skipped_network"] += 1 + if is_empty: + state.counters["empty_returns"] += 1 + state.pass_statuses = state.pass_statuses[-20:] + state.pass_empty = state.pass_empty[-20:] + + # 2. Ingest global per-query instrumentation, filtered to this run's window. + qrecs, state.query_offset = _read_jsonl_tail(GLOBAL_QUERY_INST, state.query_offset) + for rec in qrecs: + ts = _parse_ts(rec.get("ts")) + if ts is not None and ts < state.start_ts - 5: + continue # belongs to a different/older run + if ts: + state.last_event_ts = max(state.last_event_ts, ts) + state.query_window.append(rec) + if rec.get("reasoning_trail_detection") == "flagged_blanked": + state.counters["reasoning_trail"] += 1 + chrome = rec.get("chrome_path_used") + if chrome == "cdp_busy": + state.counters["cdp_busy"] += 1 + elif isinstance(chrome, str) and chrome.startswith("local"): + state.counters["local_fallback"] += 1 + if rec.get("result_validation") == "suspect_truncated": + state.counters["result_suspect"] += 1 + state.query_window = state.query_window[-REASONING_TRAIL_WINDOW:] + + # 3. Ingest autonomous-loop trace (loop kind only) for heartbeat. Accept + # --workdir pointing at either the project root or its .workflow dir. + if state.kind == "autonomous-loop": + trace_path = wd / ".workflow" / "trace.jsonl" + if not trace_path.exists() and (wd / "trace.jsonl").exists(): + trace_path = wd / "trace.jsonl" + # Heartbeat only: loop-level liveness comes from trace timestamps; the + # Perplexity-call health (empty / reasoning-trail / cdp / cookie TTL) for + # the loop's research calls comes from the global query log ingested above. + trecs, state.trace_offset = _read_jsonl_tail(trace_path, state.trace_offset) + for rec in trecs: + ts = _parse_ts(rec.get("timestamp")) + if ts: + state.last_event_ts = max(state.last_event_ts, ts) + + # 4. Terminal detection. + ledger = _read_json(wd / "ledger.json") + done_sentinel = (wd / "runner.log.done").exists() + ledger_status = str(ledger.get("status")) if ledger else None + terminal = bool( + done_sentinel + or (ledger_status in LEDGER_TERMINAL_STATUSES) + or not _pid_alive(state.pid) + ) + + # 5. Cookie TTL + proactive remediation. + cookie = _critical_cookie_ttl() + elapsed_min = (time.time() - state.start_ts) / 60.0 + remaining_run_min = max(0.0, state.expected_run_min - elapsed_min) + ttl_floor = max(COOKIE_TTL_FLOOR_MIN, remaining_run_min) + min_ttl = cookie.get("min_critical_minutes") + cookie_short = isinstance(min_ttl, (int, float)) and min_ttl < ttl_floor + if cookie_short and remediate and not terminal: + if (time.time() - state.keeper_last_fired) > KEEPER_COOLDOWN_S: + if _trigger_keeper(wd): + now = time.time() + state.keeper_last_fired = now + state.keeper_fire_history.append(now) + state.cookie_auto_refreshed = True + state.cookie_last_refresh_iso = _now_iso() + # Keeper-thrash: fired repeatedly in the look-back window but cookies are + # STILL short → the problem has moved upstream (Perplexity policy / account) + # and automation can no longer remediate it. Surface as a distinct class so + # a successful-looking refresh doesn't mask a persistent failure. + recent_fires = [t for t in state.keeper_fire_history + if (time.time() - t) <= KEEPER_THRASH_WINDOW_S] + state.keeper_fire_history = recent_fires + keeper_thrashing = cookie_short and len(recent_fires) >= KEEPER_THRASH_COUNT + cookie_out = { + "min_critical_minutes": min_ttl, + "stale_critical": cookie.get("stale_critical", []), + "auto_refreshed": state.cookie_auto_refreshed, + "last_refresh_ts": state.cookie_last_refresh_iso, + } + + # 6. Build alerts. + alerts: list[dict[str, Any]] = [] + parse_streak = _trailing_streak(state.pass_statuses, "PARSE-FAILED") + net_streak = _trailing_streak(state.pass_statuses, "SKIPPED-NETWORK") + empty_streak = _trailing_streak(["EMPTY" if e else "X" for e in state.pass_empty], "EMPTY") + abort_recommended = False + abort_reason: str | None = None + + def add(signal: str, severity: str, detail: str) -> None: + alerts.append({"signal": signal, "severity": severity, + "detail": detail, "ts": _now_iso()}) + + if parse_streak >= PARSE_FAIL_STREAK_ABORT: + abort_recommended, abort_reason = True, f"{parse_streak} consecutive PARSE-FAILED" + add("parse_failed_streak", "critical", abort_reason + " (runner aborts)") + elif parse_streak >= STREAK_WARN: + add("parse_failed_streak", "warn", + f"{parse_streak} consecutive PARSE-FAILED ({PARSE_FAIL_STREAK_ABORT - parse_streak} from abort)") + + if net_streak >= SKIPPED_NETWORK_STREAK_ABORT: + abort_recommended, abort_reason = True, f"{net_streak} consecutive SKIPPED-NETWORK" + add("skipped_network_streak", "critical", abort_reason + " (runner aborts)") + elif net_streak >= STREAK_WARN: + add("skipped_network_streak", "warn", + f"{net_streak} consecutive SKIPPED-NETWORK ({SKIPPED_NETWORK_STREAK_ABORT - net_streak} from abort)") + + if empty_streak >= EMPTY_RETURN_LOUD: + add("empty_return", "critical", f"{empty_streak} consecutive empty returns") + elif empty_streak == 1: + add("empty_return", "warn", "1 empty return") + + if state.query_window: + rt = sum(1 for q in state.query_window + if q.get("reasoning_trail_detection") == "flagged_blanked") + if rt / len(state.query_window) > REASONING_TRAIL_RATE: + add("reasoning_trail", "critical", + f"{rt}/{len(state.query_window)} recent queries reasoning-trail-only") + # Empty-synthesis queries (disjoint from reasoning-trail). This is the + # primary empty-return signal for autonomous loops, which have no + # runner-side instrumentation.jsonl. Relies on council_browser now + # recording exit_reason="empty_synthesis" / 0 chars instead of "completed". + eq = sum(1 for q in state.query_window + if (q.get("exit_reason") == "empty_synthesis" + or q.get("extracted_synthesis_chars", 1) == 0) + and q.get("reasoning_trail_detection") != "flagged_blanked") + if eq / len(state.query_window) > REASONING_TRAIL_RATE: + add("empty_query", "critical", + f"{eq}/{len(state.query_window)} recent queries returned empty synthesis") + # Result-validation survivors: synthesis was stored but failed the + # post-extraction completeness check (truncation / parse / UI regression). + rs = sum(1 for q in state.query_window + if q.get("result_validation") == "suspect_truncated") + if rs: + sev = "critical" if rs / len(state.query_window) > REASONING_TRAIL_RATE else "warn" + add("result_suspect", sev, + f"{rs}/{len(state.query_window)} recent queries failed result-validation " + "(suspect truncated/incomplete)") + + if state.query_window and state.query_window[-1].get("chrome_path_used") in DEGRADED_CDP_PATHS: + add("cdp_degraded", "warn", + f"chrome_path_used={state.query_window[-1].get('chrome_path_used')} (keeper cookies not in use)") + + if keeper_thrashing: + add("keeper_thrashing", "critical", + f"keeper fired {len(recent_fires)}x in {int(KEEPER_THRASH_WINDOW_S / 60)}min but " + f"critical cookie TTL still {min_ttl}min — problem is upstream (Perplexity policy / " + f"account); automation cannot remediate. Needs manual session re-login.") + elif cookie_short: + sev = "critical" if (state.cookie_auto_refreshed and isinstance(min_ttl, (int, float)) + and min_ttl < COOKIE_TTL_FLOOR_MIN) else "warn" + add("cookie_short_ttl", sev, + f"critical cookie TTL {min_ttl}min < floor {round(ttl_floor, 1)}min" + + (" (keeper fired)" if state.cookie_auto_refreshed else "")) + + seconds_since = (time.time() - state.last_event_ts) if state.last_event_ts else 0.0 + stalled = (not terminal and state.last_event_ts > 0 + and seconds_since > HEARTBEAT_STALE_S) + if stalled: + add("stalled", "critical", f"no new event in {int(seconds_since)}s") + + empty_result_run = bool( + terminal and state.passes_completed > 0 + and state.counters["parse_failed"] + state.counters["skipped_network"] == 0 + and sum(1 for s in state.pass_statuses if s == "COMPLETED") > 0 + and ledger is not None + and not (ledger.get("pass_log") and any( + (p or {}).get("net_new_findings", 0) for p in ledger.get("pass_log", []))) + ) + if empty_result_run: + add("empty_result_run", "warn", "run completed with zero net findings") + + # 6b. Zombie handshake: if the advisory abort persists across several ticks + # while the runner has NOT self-aborted (its own breaker never fired — + # e.g. events stopped arriving so the streak counter is frozen), write a + # sentinel that Claude's orchestrator explicitly checks and acts on. The + # monitor still never kills the runner — this is a two-step handshake. + if abort_recommended and not terminal: + state.abort_persist_ticks += 1 + else: + state.abort_persist_ticks = 0 + if state.abort_persist_ticks >= ABORT_PERSIST_TICKS: + try: + (wd / "monitor-abort-requested.flag").write_text( + json.dumps({"ts": _now_iso(), "reason": abort_reason, + "persisted_ticks": state.abort_persist_ticks}), + encoding="utf-8") + except OSError as exc: + _log(wd, f"abort-flag write failed: {exc!r}") + + # 7. Health rollup. + if terminal: + health = "DONE" + elif stalled: + health = "STALLED" + elif any(a["severity"] == "critical" for a in alerts): + health = "CRITICAL" + elif alerts: + health = "DEGRADED" + else: + health = "OK" + + status = { + "schema": STATUS_SCHEMA, + "run_kind": state.kind, + "workdir": str(state.workdir), + "pid": state.pid, + "updated_ts": _now_iso(), + "health": health, + "last_event_ts": (datetime.fromtimestamp(state.last_event_ts, timezone.utc) + .strftime("%Y-%m-%dT%H:%M:%SZ") if state.last_event_ts else None), + "seconds_since_last_event": int(seconds_since), + "passes_completed": state.passes_completed, + "counters": dict(state.counters), + "cookie_ttl": cookie_out, + "active_alerts": alerts, + "abort_recommended": abort_recommended, + "abort_reason": abort_reason, + "abort_persist_ticks": state.abort_persist_ticks, + "ledger_status": ledger_status, + "terminal": terminal, + } + return status + + +# --------------------------------------------------------------------------- status io + +def write_status(workdir: Path, status: dict[str, Any]) -> None: + """Atomically write monitor-status.json. Never raises.""" + try: + path = workdir / "monitor-status.json" + tmp = workdir / "monitor-status.json.tmp" + path.parent.mkdir(parents=True, exist_ok=True) + tmp.write_text(json.dumps(status, indent=2), encoding="utf-8") + os.replace(tmp, path) + except OSError as exc: + _log(workdir, f"write_status failed: {exc!r}") + + +def maybe_notify(state: MonitorState, status: dict[str, Any], *, enabled: bool) -> None: + """Fire Pushover on health worsening or a new alert key (dedupe via state).""" + if not enabled: + return + severity_rank = {"OK": 0, "DONE": 0, "DEGRADED": 1, "STALLED": 2, "CRITICAL": 2} + health = status["health"] + worsened = severity_rank.get(health, 0) > severity_rank.get(state.last_health, 0) + current_keys = {f'{a["signal"]}:{a["severity"]}' for a in status["active_alerts"]} + new_keys = [a for a in status["active_alerts"] + if f'{a["signal"]}:{a["severity"]}' not in state.notified_keys] + if worsened or new_keys: + lines = [f'- {a["signal"]} [{a["severity"]}]: {a["detail"]}' + for a in (new_keys or status["active_alerts"])] + priority = 1 if health in ("CRITICAL", "STALLED") else 0 + _send_pushover( + state.workdir, + f"[research-monitor] {health}: {state.kind}", + "\n".join(lines) or f"health={health}", + priority, + ) + # Reset to exactly the currently-active keys: resolved alerts drop out (so a + # recurrence re-fires), still-active alerts stay suppressed (no spam). + state.notified_keys = current_keys + state.last_health = health + + +# --------------------------------------------------------------------------- calibration + +def _finalize_calibration(workdir: Path) -> None: + """Run the existing calibration aggregator at terminal. Best-effort.""" + try: + sys.path.insert(0, str(Path(__file__).parent)) + import calibration_log # type: ignore + summary = calibration_log.aggregate_run(Path(workdir)) + if summary and hasattr(calibration_log, "append_to_global_summary"): + calibration_log.append_to_global_summary(summary) + _log(workdir, "calibration aggregate_run done") + except Exception as exc: # noqa: BLE001 — best-effort, never block exit + _log(workdir, f"calibration finalize skipped: {exc!r}") + + +# --------------------------------------------------------------------------- main + +def _run_once(state: MonitorState, *, remediate: bool, notify: bool) -> dict[str, Any]: + status = evaluate(state, remediate=remediate) + write_status(state.workdir, status) + maybe_notify(state, status, enabled=notify) + return status + + +def main(argv: list[str]) -> int: + parser = argparse.ArgumentParser(description="Active monitor for Perplexity-browser runs.") + parser.add_argument("--workdir", required=True) + parser.add_argument("--kind", default="extended-research", + choices=["extended-research", "autonomous-loop"]) + parser.add_argument("--pid", type=int, default=None) + parser.add_argument("--expected-run-min", type=float, default=40.0) + parser.add_argument("--interval", type=float, default=12.0) + parser.add_argument("--once", action="store_true") + parser.add_argument("--no-pushover", action="store_true") + parser.add_argument("--no-remediate", action="store_true") + args = parser.parse_args(argv) + + workdir = Path(args.workdir).expanduser() + state = MonitorState( + workdir=workdir, kind=args.kind, pid=args.pid, + expected_run_min=args.expected_run_min, start_ts=time.time(), + ) + remediate = not args.no_remediate + notify = not args.no_pushover + _log(workdir, f"START kind={args.kind} pid={args.pid} once={args.once}") + + if args.once: + _run_once(state, remediate=remediate, notify=notify) + return 0 + + while True: + try: + status = _run_once(state, remediate=remediate, notify=notify) + except Exception as exc: # noqa: BLE001 — fail-open monitor loop + _log(workdir, f"tick error (continuing): {exc!r}") + time.sleep(args.interval) + continue + if status.get("terminal"): + _log(workdir, f"TERMINAL health={status['health']} " + f"ledger={status.get('ledger_status')}") + _finalize_calibration(workdir) + return 0 + time.sleep(args.interval) + + +if __name__ == "__main__": + try: + sys.exit(main(sys.argv[1:])) + except Exception as exc: # noqa: BLE001 — must never crash into the parent + try: + _log(Path.cwd(), f"top_level_exception {type(exc).__name__}: {exc}") + except Exception: + pass + sys.exit(0) diff --git a/council-automation/research_queue.py b/council-automation/research_queue.py index c5a9a90..36c89e2 100644 --- a/council-automation/research_queue.py +++ b/council-automation/research_queue.py @@ -63,6 +63,15 @@ MAX_WAIT_S = int(os.environ.get("RESEARCH_QUEUE_MAX_WAIT", "1200")) ACTIVITY_LOG_LOCK_TIMEOUT_S = 10 # exposed for tests (monkeypatchable) SNAPSHOT_LOCK_TIMEOUT_S = 5 # exposed for tests (monkeypatchable) +# os.replace() onto a destination with a concurrent open handle raises +# PermissionError (WinError 5 / ERROR_ACCESS_DENIED) on Windows. The snapshot +# FileLock serializes writers but cannot exclude readers (monitor, queue +# daemon, MCP status tool, other runners checking their position), so the +# swap is retried a few times with a short backoff to ride out that transient +# sharing violation. 12 * 25ms = 300ms worst case; readers hold the file for +# microseconds so the swap almost always succeeds on the first or second try. +_ATOMIC_REPLACE_RETRIES = 12 # exposed for tests (monkeypatchable) +_ATOMIC_REPLACE_BACKOFF_S = 0.025 # exposed for tests (monkeypatchable) STATS_LOOKBACK_LINES = 5000 # tail bound for today's-stats scan (see _compute_today_stats) # Windows OpenProcess() access right sufficient only to query existence -- @@ -130,7 +139,24 @@ def _atomic_write(path: Path, text: str) -> None: try: with os.fdopen(fd, "w", encoding="utf-8") as fh: fh.write(text) - os.replace(tmp_name, path) + # os.replace is atomic, but on Windows it raises PermissionError + # (WinError 5) when the destination has a concurrent open handle -- a + # reader (monitor / queue daemon / MCP status tool / another runner + # checking its position) holding the file open at the instant of the + # swap. That sharing violation is transient (readers hold it for + # microseconds), so retry the swap a few times with a short backoff + # before surfacing the error. Without this, a lost race silently drops + # a run's completion/instrumentation write, or aborts the caller's run + # outright (see publish_snapshot: WinError 5 is not the Timeout it + # catches, so it would otherwise propagate). + for _attempt in range(_ATOMIC_REPLACE_RETRIES): + try: + os.replace(tmp_name, path) + break + except PermissionError: + if _attempt == _ATOMIC_REPLACE_RETRIES - 1: + raise + time.sleep(_ATOMIC_REPLACE_BACKOFF_S) except Exception: try: os.unlink(tmp_name) @@ -316,6 +342,46 @@ def gc_dead_tickets() -> None: f.unlink() except OSError as exc: logger.debug("gc_dead_tickets: could not unlink %s (%s)", f, exc) + continue + # A reclaimed ticket is a run that DIED WITHOUT SAYING SO -- the only + # trace it ever leaves. `acquire_slot`'s `except BaseException` logs + # an "error" event for anything that raises inside the with-block, + # but it cannot run at all when the process is killed outright, and + # that is the normal end for a research run: the MCP layer calls + # execFileAsync(..., {timeout: 540_000}) and Node's timeout kill on + # Windows is TerminateProcess -- no signal, no unwinding, no + # finally-blocks. So the activity log recorded `started` and nothing + # else, `errors_today` stayed 0, and two machine-wide outages in 24h + # (2026-08-07, 2026-08-08) both had to be discovered by a session + # tripping over them instead of by the monitor. Logging the reclaim + # is what closes that hole: it is the one moment another process can + # observe that the run is gone. See also the silent-drop analysis in + # memory `reading-queue-feeds-correctly`. + # + # Emitted only by the process that WON the unlink race (the loser + # gets OSError and `continue`s above), so exactly one event per run. + try: + log_event( + "dropped", + d.get("run_id", "?"), + d.get("session", "?"), + int(d.get("seq", -1)), + d.get("query_preview", ""), + d.get("mode", "research"), + error=( + "runner process died without logging an outcome " + f"(pid={d.get('pid')} {'alive-but-heartbeat-stale' if alive else 'dead'})" + ), + ticket_state=d.get("state", "unknown"), + pid_alive=bool(alive), + heartbeat_age_s=round(now - d.get("heartbeat_at", now), 1), + duration_s=( + round(now - d["active_since"], 1) + if isinstance(d.get("active_since"), (int, float)) else None + ), + ) + except Exception as exc: # never let bookkeeping break the GC + logger.debug("gc_dead_tickets: could not log dropped event (%s)", exc) def live_tickets() -> List[Tuple[Path, Dict[str, Any]]]: @@ -360,9 +426,12 @@ def log_event( ) -> None: """Append one JSON line describing a queue transition to ACTIVITY_LOG. - Events: enqueued | started | completed | error | timeout. Extra + Events: enqueued | started | completed | error | timeout | dropped. Extra keyword args (wait_s, duration_s, status, error, ...) are merged directly into the record. + + `dropped` is emitted by `gc_dead_tickets()`, not by the run itself -- it is + the outcome for a run whose process was killed before it could report one. """ ACTIVITY_LOG.parent.mkdir(parents=True, exist_ok=True) record: Dict[str, Any] = { @@ -428,20 +497,32 @@ def _is_same_local_day(ts: float, now: float) -> bool: return (a.tm_year, a.tm_yday) == (b.tm_year, b.tm_yday) -def _compute_today_stats(now: float) -> Tuple[int, int]: - """Compute (total_today, errors_today) from the activity log. +#: Activity-log events that mean "this run did not deliver a result". `dropped` +#: is here because a killed runner is a failed run: excluding it is precisely +#: what let `errors_today: 0` coexist with a total research outage. +FAILURE_EVENTS = ("error", "timeout", "dropped") + + +def _compute_today_stats(now: float) -> Tuple[int, int, Optional[Dict[str, Any]]]: + """Compute (total_today, errors_today, last_failure) from the activity log. `total_today` counts DISTINCT RUNS (not raw events) that started today, - identified by a `started` event's run_id. `errors_today` counts `error` - and `timeout` events logged today. Bounded to the last - STATS_LOOKBACK_LINES lines of the activity log for performance (this is - read on every publish_snapshot() call, which can fire every ~1.5s per - waiting session) -- more than sufficient to cover a single day's events - under normal usage. + identified by a `started` event's run_id. `errors_today` counts + FAILURE_EVENTS logged today. `last_failure` is a compact summary of the + most recent one (or None), so a caller reading only the snapshot can see + WHAT broke without opening the activity log -- the MCP + `research_queue_status` tool previously exposed a bare count, which reads + as "healthy" the moment the count is wrong. + + Bounded to the last STATS_LOOKBACK_LINES lines of the activity log for + performance (this is read on every publish_snapshot() call, which can fire + every ~1.5s per waiting session) -- more than sufficient to cover a single + day's events under normal usage. """ events = _read_activity_events(limit=STATS_LOOKBACK_LINES) started_run_ids_today: set[str] = set() errors_today = 0 + last_failure: Optional[Dict[str, Any]] = None for e in events: ts = e.get("ts") if not isinstance(ts, (int, float)) or not _is_same_local_day(ts, now): @@ -451,9 +532,17 @@ def _compute_today_stats(now: float) -> Tuple[int, int]: run_id = e.get("run_id") if run_id: started_run_ids_today.add(run_id) - elif event in ("error", "timeout"): + elif event in FAILURE_EVENTS: errors_today += 1 - return len(started_run_ids_today), errors_today + last_failure = { + "event": event, + "run_id": e.get("run_id"), + "session": e.get("session"), + "ts": ts, + "age_s": round(now - ts, 1), + "error": (str(e.get("error") or ""))[:300], + } + return len(started_run_ids_today), errors_today, last_failure def publish_snapshot(my_seq: Optional[int] = None) -> None: @@ -507,7 +596,7 @@ def publish_snapshot(my_seq: Optional[int] = None) -> None: item["position"] = i recent = _read_recent_activity(limit=20) - total_today, errors_today = _compute_today_stats(now) + total_today, errors_today, last_failure = _compute_today_stats(now) snapshot = { "version": time.time_ns(), @@ -519,6 +608,7 @@ def publish_snapshot(my_seq: Optional[int] = None) -> None: "depth": len(queued) + (1 if active else 0), "total_today": total_today, "errors_today": errors_today, + "last_failure": last_failure, }, } text = json.dumps(snapshot) diff --git a/council-automation/session_keeper.py b/council-automation/session_keeper.py index a42952e..fd5f8dd 100644 --- a/council-automation/session_keeper.py +++ b/council-automation/session_keeper.py @@ -129,7 +129,16 @@ def _harden_for_windows_daemon() -> None: KEEPER_HEARTBEAT_FILE = Path.home() / ".claude" / "config" / "session_keeper.heartbeat" KEEPER_CDP_FILE = Path.home() / ".claude" / "config" / "session_keeper.cdp" DEFAULT_INTERVAL_S = 1200 # 20 min (CF __cf_bm has ~30min TTL; refresh well before expiry) -DEFAULT_CDP_PORT = 9222 # Chrome DevTools Protocol port for council_browser CDP attach +# Chrome DevTools Protocol port the keeper owns, for council_browser's CDP +# attach. Moved off 9222 on 2026-08-08: ~/.claude/browser-relay/relay.mjs (the +# /takeover phone relay) binds 9222 too, whichever process boots first wins, +# and that collision caused BOTH the 08-07 and 08-08 machine-wide research +# outages. The keeper and the relay now have disjoint ports and can coexist, +# which is also what unblocks re-enabling the PerplexitySessionKeeper task +# (Disabled since 2026-07-30 because re-enabling it would have fought the relay +# for 9222). Must match council_browser.KEEPER_CDP_PORT -- both read +# COUNCIL_KEEPER_CDP_PORT so a single env var moves them together. +DEFAULT_CDP_PORT = int(os.environ.get("COUNCIL_KEEPER_CDP_PORT", "9223")) def _log(msg: str) -> None: @@ -337,6 +346,55 @@ async def _navigate_and_warm(page) -> bool: return False +def _cdp_serving_pid(port: int) -> int | None: + """PID of the process LISTENING on `port`, or None if it cannot be read. + + This is the pid published in session_keeper.cdp -- readers need the process + that actually answers CDP, which outlives this one-shot keeper invocation. + """ + if sys.platform != "win32": + return None + try: + import subprocess + out = subprocess.run( + ["powershell", "-NoProfile", "-Command", + f"(Get-NetTCPConnection -LocalPort {port} -State Listen " + f"-ErrorAction SilentlyContinue | Select-Object -First 1).OwningProcess"], + capture_output=True, text=True, timeout=15, + ) + return int((out.stdout or "").strip()) + except Exception as exc: + _log(f"CDP serving-pid probe failed ({type(exc).__name__}: {exc})") + return None + + +def _cdp_port_owner_cmdline(port: int) -> str | None: + """Command line of the process LISTENING on `port`, or None if unknown. + + Windows-only. None means "could not determine" and must be treated as + unknown, never as "foreign" -- a failed probe should not stop the keeper. + Mirrors council_browser.PerplexityCouncil._cdp_port_owner_cmdline. + """ + if sys.platform != "win32": + return None + ps = ( + f"$c = Get-NetTCPConnection -LocalPort {port} -State Listen " + f"-ErrorAction SilentlyContinue | Select-Object -First 1; " + f"if ($c) {{ (Get-CimInstance Win32_Process -Filter " + f"\"ProcessId=$($c.OwningProcess)\").CommandLine }}" + ) + try: + import subprocess + out = subprocess.run( + ["powershell", "-NoProfile", "-Command", ps], + capture_output=True, text=True, timeout=15, + ) + except Exception as exc: + _log(f"CDP owner probe failed ({type(exc).__name__}: {exc}) - owner unknown") + return None + return (out.stdout or "").strip() or None + + async def _wait_for_cdp(port: int, timeout: float = 30.0) -> dict | None: """Poll Chrome's CDP /json/version endpoint until it responds, or timeout.""" import urllib.request as _ur @@ -438,9 +496,22 @@ async def main_loop(interval_s: int, cdp_port: int = DEFAULT_CDP_PORT) -> None: with _ur.urlopen(f"http://127.0.0.1:{cdp_port}/json/version", timeout=2) as _r: _ = _r.read() chrome_already_up = True - _log(f"Chrome already serving CDP on port {cdp_port}; reusing it (no relaunch)") except Exception: chrome_already_up = False + if chrome_already_up: + # "Something answers CDP here" is NOT "our Chrome is here". Reusing a + # foreign browser hands the keeper a cookie-less profile it then + # publishes as the live Perplexity session -- the 2026-08-07 failure, + # from the other side. Prove the listener is running OUR profile dir + # before adopting it; otherwise refuse the port loudly rather than + # silently keeping a session nobody is logged in to. + owner = _cdp_port_owner_cmdline(cdp_port) + if owner is not None and keeper_profile_dir.name not in owner: + _log(f"REFUSING port {cdp_port}: it is served by a FOREIGN process " + f"(no '{keeper_profile_dir.name}' in its command line). Set " + f"COUNCIL_KEEPER_CDP_PORT to a free port. Owner: {owner[:160]}") + return + _log(f"Chrome already serving CDP on port {cdp_port}; reusing it (no relaunch)") if not chrome_already_up: # Only sweep orphans if NOTHING is serving CDP — that means whatever # Chrome was tied to our profile dir is in a half-dead state and @@ -456,16 +527,63 @@ async def main_loop(interval_s: int, cdp_port: int = DEFAULT_CDP_PORT) -> None: if not chrome_already_up: DETACHED_PROCESS = 0x00000008 if sys.platform == "win32" else 0 CREATE_NEW_PROCESS_GROUP = 0x00000200 if sys.platform == "win32" else 0 - creationflags = (DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP) if sys.platform == "win32" else 0 + # CREATE_BREAKAWAY_FROM_JOB is what actually makes the one-shot design + # work, and its absence is why the keeper has effectively never stayed + # up. DETACHED_PROCESS only detaches the console; it does NOT escape a + # Windows job object, and every realistic launcher puts us in one -- + # Task Scheduler assigns each task instance a job, and so does the + # agent harness that runs ad-hoc commands. When the launching task + # instance ended, the job was torn down and took Chrome with it within + # a minute or two, cleanly (Chrome even removed its DevToolsActivePort + # file, which is what made this look like a normal shutdown rather than + # a kill). Every subsequent run then found no CDP, fell back to a local + # launch, and ran the local-launch freshness guard -- the code path that + # produced the 2026-08-08 outage. + # + # MEASURED CAVEAT — the flag helps, but NOT under Task Scheduler. + # Breakaway only succeeds if the CONTAINING job grants + # JOB_OBJECT_LIMIT_BREAKAWAY_OK, and Task Scheduler's job does not: + # launched from the task, this raises [WinError 5] Access is denied and + # falls back below (observed 2026-08-08 07:27). Launched from an + # interactive shell it works and Chrome persists. This is the same + # constraint that killed the Santee demo app — already proved with + # IsProcessInJob, see memory port-registry, do not re-litigate it. + # + # The proper fix there was to make the long-lived process the task's OWN + # process rather than a child of it. That is NOT applied here: it would + # need either a new scheduled task (schtasks /Create is denied without + # elevation on this box) or reverting the keeper to a long-lived loop, + # which was abandoned in May because pythonw under Task Scheduler died + # silently inside asyncio.sleep (see memory session-keeper-history). + # + # Consequence, stated plainly: under Task Scheduler the keeper's Chrome + # lives on the order of minutes and is re-established by the task's own + # repetition. Nothing DEPENDS on it — the local-launch path has a + # correct freshness guard and a working refresher — so this is a + # latency optimization, not a availability requirement. Left as a known + # open item rather than papered over. import subprocess - chrome_proc = subprocess.Popen( - chrome_args, - creationflags=creationflags, + CREATE_BREAKAWAY_FROM_JOB = 0x01000000 if sys.platform == "win32" else 0 + base_flags = (DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP) if sys.platform == "win32" else 0 + popen_kwargs = dict( stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, close_fds=True, ) + try: + chrome_proc = subprocess.Popen( + chrome_args, + creationflags=base_flags | CREATE_BREAKAWAY_FROM_JOB, + **popen_kwargs, + ) + _log("Chrome launched with CREATE_BREAKAWAY_FROM_JOB (survives launcher teardown)") + except OSError as exc: + _log(f"Job breakaway not permitted ({exc}); launching without it — " + f"Chrome may not outlive this launcher") + chrome_proc = subprocess.Popen( + chrome_args, creationflags=base_flags, **popen_kwargs, + ) _log(f"Chrome subprocess pid={chrome_proc.pid}; waiting for CDP port {cdp_port}...") # Wait for CDP to be responsive — whether from a fresh launch or a # pre-existing Chrome. @@ -517,8 +635,23 @@ async def main_loop(interval_s: int, cdp_port: int = DEFAULT_CDP_PORT) -> None: # Windows, but Chrome's --remote-debugging-port listens on IPv4 only, # causing Playwright's connect_over_cdp to ECONNREFUSED on IPv6. cdp_endpoint = f"http://127.0.0.1:{cdp_port}" + # `pid` MUST be the pid of the process actually SERVING CDP, i.e. Chrome + # -- not this keeper python. The keeper is one-shot per invocation and + # exits seconds from now while its DETACHED_PROCESS Chrome lives on, so + # recording os.getpid() here published a pid that was always dead by the + # time any reader looked. council_browser's liveness gate then read the + # healthy keeper as a stale .cdp file, deleted it, and fell back to a + # local launch on EVERY run -- which is how the 08-08 outage reached the + # local-launch freshness guard at all. Keep the launcher pid separately + # for diagnostics; it is not an identity signal. + chrome_pid = _cdp_serving_pid(cdp_port) KEEPER_CDP_FILE.write_text( - json.dumps({"port": cdp_port, "endpoint": cdp_endpoint, "pid": os.getpid()}), + json.dumps({ + "port": cdp_port, + "endpoint": cdp_endpoint, + "pid": chrome_pid, + "keeper_python_pid": os.getpid(), + }), encoding="utf-8", ) _log(f"Refresh complete: {n} cookies written. CDP endpoint: {cdp_endpoint}") diff --git a/council-automation/tests/test_pplx_failure_visibility.py b/council-automation/tests/test_pplx_failure_visibility.py new file mode 100644 index 0000000..ad8d442 --- /dev/null +++ b/council-automation/tests/test_pplx_failure_visibility.py @@ -0,0 +1,338 @@ +"""Regression tests for the 2026-08-08 Perplexity research outage. + +Two independent defects, both covered here: + +1. **The outage.** `_ensure_fresh_session` probed CDP port 9222 with a bare + `urlopen` and concluded "the keeper is alive" whenever *anything* answered. + The /takeover browser relay binds the same port, so the guard fired a keeper + refresh that could never land, waited SESSION_KEEPER_WAIT_S, and then + submitted the query on knowingly-expired cookies. `_start_via_cdp` had been + hardened against exactly this on 08-07; its twin had not. + +2. **The monitoring trap** (arguably worse). The queue logged `started` and + never `error`, so `errors_today` read 0 through a total outage. Cause: the + MCP layer hard-kills the runner on timeout, and on Windows that is + TerminateProcess -- no signal, no finally-blocks -- so `acquire_slot`'s + error path cannot run. The reclaim in `gc_dead_tickets` is the only place + another process can observe the death, so that is where the record is now + written. + +ISOLATION: as in test_research_queue.py, every queue path global is redirected +under tmp_path. Nothing here touches live queue state. +""" +from __future__ import annotations + +import asyncio +import json +import os +import sys +import time +from pathlib import Path +from typing import Any + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import council_browser as cb # noqa: E402 +import queue_monitor as qm # noqa: E402 +import research_queue as rq # noqa: E402 + + +@pytest.fixture(autouse=True) +def isolated_paths(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + """Redirect every research_queue path global under tmp_path (autouse, so + no test can forget and clobber the live queue a real run may be using). + """ + qdir = tmp_path / "research-queue" + monkeypatch.setattr(rq, "QUEUE_DIR", qdir) + monkeypatch.setattr(rq, "ACTIVITY_LOG", tmp_path / "council-logs" / "perplexity-activity.jsonl") + monkeypatch.setattr(rq, "SNAPSHOT", tmp_path / "council-logs" / "perplexity-queue.json") + return qdir + + +def _events() -> list[dict[str, Any]]: + return rq._read_activity_events() + + +# -------------------------------------------------------------------------- +# Defect 2 -- a killed runner must leave a failure record +# -------------------------------------------------------------------------- +def test_reclaimed_active_ticket_logs_dropped_event() -> None: + """The exact shape of the outage: a ticket promoted to active whose owner + then vanishes. Before this fix the reclaim was silent. + """ + rq.ensure_dirs() + ticket = rq.write_ticket( + seq=1, pid=999_999_999, run_id="run-killed", session="sess-a", + state="active", query_preview="benchmark query", mode="research", + ) + # Stamp it the way acquire_slot does at promotion, so duration_s is real. + data = json.loads(ticket.read_text(encoding="utf-8")) + data["active_since"] = time.time() - 137.0 + ticket.write_text(json.dumps(data), encoding="utf-8") + rq.log_event("started", "run-killed", "sess-a", 1, "benchmark query", "research") + + rq.gc_dead_tickets() + + assert not ticket.exists(), "dead ticket should still be reclaimed" + dropped = [e for e in _events() if e.get("event") == "dropped"] + assert len(dropped) == 1, f"expected exactly one dropped event, got {_events()}" + rec = dropped[0] + assert rec["run_id"] == "run-killed" + assert rec["session"] == "sess-a" + assert rec["ticket_state"] == "active" + assert rec["pid_alive"] is False + assert rec["duration_s"] == pytest.approx(137.0, abs=2.0) + assert "died without logging an outcome" in rec["error"] + + +def test_dropped_event_emitted_once_not_per_gc_pass() -> None: + """gc runs on every live_tickets() call from every process. A duplicate + event per pass would make errors_today meaningless in the other direction. + """ + rq.ensure_dirs() + rq.write_ticket(seq=1, pid=999_999_999, run_id="run-x", session="s", + state="active", query_preview="q", mode="research") + for _ in range(4): + rq.gc_dead_tickets() + assert len([e for e in _events() if e.get("event") == "dropped"]) == 1 + + +def test_live_holder_is_never_reported_dropped() -> None: + """False positives here would page on healthy runs. Our own PID is alive + and its heartbeat is fresh, so nothing may be reclaimed or logged. + """ + rq.ensure_dirs() + ticket = rq.write_ticket(seq=1, pid=os.getpid(), run_id="run-live", session="s", + state="active", query_preview="q", mode="research") + rq.gc_dead_tickets() + assert ticket.exists() + assert [e for e in _events() if e.get("event") == "dropped"] == [] + + +def test_errors_today_counts_dropped_runs() -> None: + """The headline symptom: `errors_today: 0` while runs were dying.""" + now = time.time() + rq.ensure_dirs() + rq.log_event("started", "r1", "s", 1, "q", "research") + rq.log_event("dropped", "r1", "s", 1, "q", "research", error="runner process died") + + total, errors, last_failure = rq._compute_today_stats(now) + + assert total == 1 + assert errors == 1, "a killed run is a failed run" + assert last_failure is not None + assert last_failure["event"] == "dropped" + assert last_failure["run_id"] == "r1" + + +def test_snapshot_exposes_last_failure() -> None: + """A bare count reads as healthy the moment it is wrong; callers need to + see WHAT broke straight off research_queue_status. + """ + rq.ensure_dirs() + rq.log_event("started", "r1", "s", 1, "q", "research") + rq.log_event("dropped", "r1", "s", 1, "q", "research", error="runner process died") + + rq.publish_snapshot() + snap = json.loads(rq.SNAPSHOT.read_text(encoding="utf-8")) + + assert snap["stats"]["errors_today"] == 1 + assert snap["stats"]["last_failure"]["event"] == "dropped" + assert "runner process died" in snap["stats"]["last_failure"]["error"] + + +def test_queue_monitor_alerts_on_dropped() -> None: + """The monitor was blind to this event class even once it was logged.""" + alerts = qm.evaluate( + snapshot=None, + recent_events=[{ + "event": "dropped", "run_id": "r1", "session": "s", + "error": "runner process died without logging an outcome", + }], + ) + assert len(alerts) == 1 + assert alerts[0].severity == "critical" + assert "dropped" in alerts[0].message + assert alerts[0].key == "event:r1:dropped" + + +# -------------------------------------------------------------------------- +# Defect 1 -- identity, not reachability +# -------------------------------------------------------------------------- +def _council() -> cb.PerplexityCouncil: + return cb.PerplexityCouncil.__new__(cb.PerplexityCouncil) + + +def test_keeper_cdp_alive_rejects_foreign_owner(monkeypatch: pytest.MonkeyPatch) -> None: + """A port that answers CDP but is owned by the /takeover relay must NOT + count as a live keeper. This is the assertion the outage violated. + """ + monkeypatch.setattr(cb.PerplexityCouncil, "_cdp_endpoint_is_keeper", + classmethod(lambda cls, ep, pid: ( + False, "port is owned by a FOREIGN Chrome (browser-relay /takeover)"))) + monkeypatch.setattr(cb.PerplexityCouncil, "_cdp_port_owner_cmdline", + staticmethod(lambda port: r"chrome.exe --user-data-dir=C:\Temp\igx-cdp-profile")) + + class _Resp: + def read(self) -> bytes: return b"{}" + def __enter__(self): return self + def __exit__(self, *a): return False + monkeypatch.setattr("urllib.request.urlopen", lambda *a, **k: _Resp()) + + ok, why = cb.PerplexityCouncil._keeper_cdp_alive() + assert ok is False + assert "FOREIGN" in why + + +def test_keeper_and_relay_ports_are_disjoint() -> None: + """The structural fix: the keeper no longer shares a port with the relay, + so neither can lock the other out by booting first. + """ + assert cb.KEEPER_CDP_PORT != 9222 + + +def _run_freshness_guard(council: cb.PerplexityCouncil) -> None: + asyncio.run(council._ensure_fresh_session("test")) + + +def test_expired_cookies_abort_instead_of_proceeding(monkeypatch: pytest.MonkeyPatch) -> None: + """Replaces `keeper-timeout-stale-proceed`. Submitting on expired cookies + bought a guaranteed death 5-8 minutes later, on a serialized queue, with + no error ever logged. Failing here costs seconds and names the fix. + """ + council = _council() + council.session_path = Path("nonexistent-session.json") + monkeypatch.setattr(cb.PerplexityCouncil, "_check_session_freshness", + lambda self, p: {"stale_critical": [("pplx.session-id", 40)], + "expiring_soon": [], "min_critical_ttl_s": None}) + monkeypatch.setattr(cb.PerplexityCouncil, "_keeper_cdp_alive", + classmethod(lambda cls: (False, "no CDP on port 9223"))) + monkeypatch.setattr(cb.PerplexityCouncil, "_keeper_task_state", staticmethod(lambda: "Disabled")) + + async def _no_refresh(self) -> bool: + return False + monkeypatch.setattr(cb.PerplexityCouncil, "_auto_refresh_session", _no_refresh) + + with pytest.raises(cb.SessionStaleError) as exc: + _run_freshness_guard(council) + msg = str(exc.value) + assert "pplx.session-id(40m ago)" in msg + assert "/cache-perplexity-session" in msg + + +def test_expiring_soon_cookies_do_not_abort(monkeypatch: pytest.MonkeyPatch) -> None: + """Only HARD-EXPIRED cookies abort. A session with minutes left still + answers fine, and aborting on it would cascade into the runner's + SKIPPED-NETWORK-STREAK breaker -- the reason the old code never aborted. + """ + council = _council() + council.session_path = Path("nonexistent-session.json") + calls = {"n": 0} + + def _freshness(self: Any, p: Path) -> dict: + calls["n"] += 1 + # Pre-guard: expiring soon. Post-refresh: nothing hard-expired. + return {"stale_critical": [], "expiring_soon": [("pplx.session-id", 300)], + "min_critical_ttl_s": 300} + monkeypatch.setattr(cb.PerplexityCouncil, "_check_session_freshness", _freshness) + monkeypatch.setattr(cb.PerplexityCouncil, "_keeper_cdp_alive", + classmethod(lambda cls: (False, "no CDP"))) + monkeypatch.setattr(cb.PerplexityCouncil, "_keeper_task_state", staticmethod(lambda: "Disabled")) + + async def _no_refresh(self) -> bool: + return False + monkeypatch.setattr(cb.PerplexityCouncil, "_auto_refresh_session", _no_refresh) + + _run_freshness_guard(council) # must not raise + assert calls["n"] >= 2, "guard should re-verify freshness after refreshing" + + +def test_disabled_keeper_task_skips_the_120s_wait(monkeypatch: pytest.MonkeyPatch) -> None: + """`Start-ScheduledTask` at a Disabled task is a silent no-op. The task was + Disabled from 2026-07-30, so every refresh burned SESSION_KEEPER_WAIT_S for + nothing. The keeper branch must not be entered at all. + """ + council = _council() + council.session_path = Path("nonexistent-session.json") + monkeypatch.setattr(cb.PerplexityCouncil, "_check_session_freshness", + lambda self, p: {"stale_critical": [], "expiring_soon": [("__cf_bm", 120)], + "min_critical_ttl_s": 120}) + monkeypatch.setattr(cb.PerplexityCouncil, "_keeper_cdp_alive", + classmethod(lambda cls: (True, "DevToolsActivePort GUID matches"))) + monkeypatch.setattr(cb.PerplexityCouncil, "_keeper_task_state", staticmethod(lambda: "Disabled")) + + fired = {"keeper": False, "direct": False} + + def _fail_if_called(*a: Any, **k: Any) -> None: + fired["keeper"] = True + raise AssertionError("must not fire a Disabled scheduled task") + monkeypatch.setattr("subprocess.run", _fail_if_called) + + async def _refresh(self) -> bool: + fired["direct"] = True + return True + monkeypatch.setattr(cb.PerplexityCouncil, "_auto_refresh_session", _refresh) + + t0 = time.time() + _run_freshness_guard(council) + assert fired["keeper"] is False + assert fired["direct"] is True, "should route straight to the direct refresher" + assert time.time() - t0 < cb.SESSION_KEEPER_WAIT_S / 2, "must not wait on a no-op task" + + +# -------------------------------------------------------------------------- +# Defect 3 -- an over-strict identity gate that disabled the path it guarded +# -------------------------------------------------------------------------- +def test_live_keeper_not_rejected_for_dead_launcher_pid(monkeypatch: pytest.MonkeyPatch) -> None: + """The keeper is one-shot: it launches Chrome DETACHED and exits seconds + later. When the recorded-pid check ran first and vetoed, EVERY healthy + keeper was judged a stale .cdp file and every run fell back to a local + launch -- which is how the outage reached the local-launch guard at all. + Positive proof of ownership must win over a dead launcher pid. + """ + monkeypatch.setattr(cb, "KEEPER_PROFILE_DIR", Path("no-such-profile-dir")) + monkeypatch.setattr(cb.PerplexityCouncil, "_cdp_port_owner_cmdline", + staticmethod(lambda port: r"chrome.exe --user-data-dir=C:\x\session_keeper_profile")) + monkeypatch.setattr(cb.PerplexityCouncil, "_pid_alive", staticmethod(lambda pid: False)) + + ok, why = cb.PerplexityCouncil._cdp_endpoint_is_keeper( + f"http://127.0.0.1:{cb.KEEPER_CDP_PORT}", recorded_pid=9684) + + assert ok is True, f"live keeper wrongly rejected: {why}" + assert "keeper profile" in why + + +def test_foreign_owner_still_rejected_even_with_live_pid(monkeypatch: pytest.MonkeyPatch) -> None: + """Loosening gate A must not loosen the actual protection.""" + monkeypatch.setattr(cb, "KEEPER_PROFILE_DIR", Path("no-such-profile-dir")) + monkeypatch.setattr(cb.PerplexityCouncil, "_cdp_port_owner_cmdline", + staticmethod(lambda port: r"chrome.exe --user-data-dir=C:\Temp\igx-cdp-profile")) + monkeypatch.setattr(cb.PerplexityCouncil, "_pid_alive", staticmethod(lambda pid: True)) + + ok, why = cb.PerplexityCouncil._cdp_endpoint_is_keeper( + f"http://127.0.0.1:{cb.KEEPER_CDP_PORT}", recorded_pid=1234) + + assert ok is False + # The label must name the ACTUAL squatter and warn it is load-bearing -- + # a human reads this string to decide what to go kill. The igx-cdp-profile + # Chrome is launched by the Dogfood Supervisor (verified via the parent + # chain 2026-08-08), not by relay.mjs as long assumed. + assert "Dogfood Supervisor" in why + assert "do NOT kill" in why + + +def test_dead_pid_still_rejected_when_no_proof_available(monkeypatch: pytest.MonkeyPatch) -> None: + """With both decisive gates indeterminate, the recorded pid is all there + is, and a dead one is genuine evidence of a leftover file. + """ + monkeypatch.setattr(cb, "KEEPER_PROFILE_DIR", Path("no-such-profile-dir")) + monkeypatch.setattr(cb.PerplexityCouncil, "_cdp_port_owner_cmdline", staticmethod(lambda port: None)) + monkeypatch.setattr(cb.PerplexityCouncil, "_pid_alive", staticmethod(lambda pid: False)) + + ok, why = cb.PerplexityCouncil._cdp_endpoint_is_keeper( + f"http://127.0.0.1:{cb.KEEPER_CDP_PORT}", recorded_pid=9684) + + assert ok is False + assert "stale" in why diff --git a/mcp-servers/browser-bridge/server.js b/mcp-servers/browser-bridge/server.js index 50cfe70..c76ab29 100644 --- a/mcp-servers/browser-bridge/server.js +++ b/mcp-servers/browser-bridge/server.js @@ -1022,6 +1022,27 @@ class BrowserBridgeServer { return { synthesis: result }; } catch (queryErr) { log.error('query_error', { invocationId, queryType, error: queryErr.message, stderr: (queryErr.stderr || '').substring(0, 300), elapsedMs: Date.now() - startMs }); + // A timeout kill is NOT an ordinary error and must not be reported as + // a bare "Task failed". execFile's timeout kills the Python runner + // with TerminateProcess on Windows -- no signal, no unwinding -- so + // the runner cannot log its own failure and the queue only ever saw + // `started`. That is why errors_today read 0 through two machine-wide + // outages. research_queue.gc_dead_tickets now logs a `dropped` event + // for the orphaned ticket; this branch makes the CALLER's side of the + // same event legible instead of opaque. + if (queryErr.killed || queryErr.signal) { + return { + error: `Perplexity runner was killed after ${Math.round(timeout / 1000)}s ` + + `(MCP hard timeout). The run did not fail cleanly — it was terminated ` + + `mid-flight, so check the queue for a "dropped" event rather than assuming ` + + `a transient error. Common causes: expired session cookies (run ` + + `/cache-perplexity-session), or a long queue wait consuming the budget ` + + `before the run even started (check research_queue_status depth).`, + code: 'RUNNER_KILLED', + elapsed_ms: Date.now() - startMs, + stderr_tail: (queryErr.stderr || '').substring(0, 800), + }; + } throw queryErr; } finally { cleanupCtx();