diff --git a/condor/code_runner.py b/condor/code_runner.py index 35a86ce05..d7ac08c50 100644 --- a/condor/code_runner.py +++ b/condor/code_runner.py @@ -29,6 +29,12 @@ ``condor/code_runs.py`` and the action-log row (:func:`condor.runtime.danger.is_code_execution_call`) are the whole containment; unattended dry-run sessions refuse it outright in ``condor/agents/risk.py``. + +Containment against *accidents* is separate and lives in +:mod:`condor.runtime.code_guard`: the run's timeout can interrupt synchronous +code (a snippet that loops or walks the filesystem on the main thread used to +freeze all of Condor), and ``exit()``, ``input()`` and a few other ways to stop +or stall the process fail on their own line instead. """ from __future__ import annotations @@ -46,6 +52,7 @@ import condor.reports as reports from condor.code_runs import CodeRun, get_code_run_store, new_run_id from condor.memory.paths import CHAT_SLUG +from condor.runtime import code_guard logger = logging.getLogger(__name__) @@ -86,6 +93,14 @@ def flush(self): if buffer is None: self._stream.flush() + def isatty(self): + # A captured run is never a terminal. Forwarding this to a real tty is + # how ``help()`` decided to start an interactive pager on the main + # thread and wait for a keypress that would never come. + if _CAPTURE.get() is not None: + return False + return self._stream.isatty() + def __getattr__(self, name): return getattr(self._stream, name) @@ -142,6 +157,7 @@ async def _run_in_task( source: str, out: dict, owner_id: int = 0, + scope: code_guard.GuardScope | None = None, ): """The snippet itself, in its own task context. @@ -153,6 +169,8 @@ async def _run_in_task( reads/deletes against it (SEC-196). ``bind_context`` publishes the same context object the snippet sees, so a ``call_routine`` inside it inherits this run's server and user instead of an ambient default (FEAT-052). + ``scope`` is the run's guard (deadline and policy); it is entered last and + left first, so the harness's own bookkeeping never runs under it. """ from condor.primitives import bind_context @@ -164,9 +182,13 @@ async def _run_in_task( bind_context(namespace.get("context")), ): try: - outcome = eval(compiled, namespace) # noqa: S307 - this IS the feature - if inspect.isawaitable(outcome): - await outcome + with ( + code_guard.guarded(scope or code_guard.GuardScope(source), compiled), + code_guard.contain_exit(), + ): + outcome = eval(compiled, namespace) # noqa: S307 - this IS the feature + if inspect.isawaitable(outcome): + await outcome finally: out["report_id"] = reports.get_last_report_id() or "" @@ -217,6 +239,14 @@ async def execute_code( get_code_run_store().save(run) return run.to_dict() + refusal = code_guard.check_source(ast.parse(code, CODE_FILENAME)) + if refusal: + run.status = "error" + run.error = f"Refused: {refusal}" + run.duration_ms = int((time.monotonic() - started) * 1000) + get_code_run_store().save(run) + return run.to_dict() + # The exact context a routine gets, so the two paths cannot diverge — the # active server travels as a preference, which is how get_client resolves it. context = WebRoutineContext( @@ -238,6 +268,11 @@ async def execute_code( _install_stdout_proxy() out: dict = {"report_id": ""} + # Started here, not with ``started``: resolving the client above is the + # harness's time, not the snippet's. + scope = code_guard.GuardScope( + label=run_id, deadline=time.monotonic() + timeout, budget=timeout + ) task = asyncio.create_task( _run_in_task( compiled, @@ -247,6 +282,7 @@ async def execute_code( label or run_id, out, owner_id=chat_id, + scope=scope, ) ) try: @@ -255,6 +291,15 @@ async def execute_code( except (asyncio.TimeoutError, TimeoutError): run.status = "timeout" run.error = f"Timed out after {timeout:g}s" + except code_guard.CodeRunDeadline as e: + # The same outcome as above, reached by the guard instead of wait_for: + # the snippet was busy on the main thread and never yielded to be + # cancelled. The traceback is the useful part — it points at the line. + run.status = "timeout" + run.error = f"Timed out after {timeout:g}s ({e})" + run.traceback = _trim_traceback( + "".join(traceback.format_exception(type(e), e, e.__traceback__)) + ) except asyncio.CancelledError: run.status = "timeout" run.error = "Cancelled" @@ -266,6 +311,10 @@ async def execute_code( "".join(traceback.format_exception(type(e), e, e.__traceback__)) ) finally: + # A task the snippet spawned and never awaited still carries the scope; + # expiring it cuts that straggler at its next line instead of letting it + # run on, unowned, after its run was reported finished. + scope.expire() run.duration_ms = int((time.monotonic() - started) * 1000) run.stdout = buffer.getvalue() run.result = _render_result(namespace) diff --git a/condor/fetchers/bot_performance.py b/condor/fetchers/bot_performance.py index c03fc0b97..3eeda8541 100644 --- a/condor/fetchers/bot_performance.py +++ b/condor/fetchers/bot_performance.py @@ -975,11 +975,43 @@ def _warn_truncated() -> None: float(perf.get("cum_fees_quote", 0) or 0), ) return { - cid: (quotes.get(cid, ""), sorted(samples.items())) + cid: (quotes.get(cid, ""), _drop_dead_readings(sorted(samples.items()))) for cid, samples in by_controller.items() } +_DEAD_READING = (0.0, 0.0, 0.0, 0.0) + + +def _drop_dead_readings( + samples: list[tuple[float, tuple[float, float, float, float]]], +) -> list[tuple[float, tuple[float, float, float, float]]]: + """Drop all-zero readings that follow a real one in a controller's series. + + A controller whose strategy died while its bot stayed listed keeps + reporting, and reports 0/0: its cumulative appears to fall back to zero. + Every session figure is a difference of cumulatives (:func:`slice_history`), + so that fall read as negative volume and invented PnL for a bot adopted + mid-life, and erased a bot born and killed inside one session (0 − 0). + These figures are cumulative within an instance — a restart is a new + instance name — so a real reading followed by zeros is never the truth. + Dropping the zeros lets the forward-carry in + :func:`merge_controller_series` hold the last real reading, the same rule + the hackathon race monitor's ledger applies. Zeros *before* the first real + reading are a controller that has not traded yet, and are kept. + """ + out: list[tuple[float, tuple[float, float, float, float]]] = [] + traded = False + for epoch, values in samples: + if values == _DEAD_READING: + if traded: + continue + else: + traded = True + out.append((epoch, values)) + return out + + def merge_controller_series( by_controller: dict[str, ControllerSeries], usd_rates: Mapping[str, float] | None = None, diff --git a/condor/notifications.py b/condor/notifications.py index 012f65b38..d9928ce4b 100644 --- a/condor/notifications.py +++ b/condor/notifications.py @@ -222,6 +222,27 @@ def mark_read(user_id: int, ids: list[str] | None = None) -> int: return sum(1 for raw in items if isinstance(raw, dict) and not raw.get("read")) +async def clear(user_id: int) -> None: + """Delete every notification in ``user_id``'s bucket. + + Takes the write lock so a notice recorded mid-clear is not resurrected (or + lost) by the two writes racing. Scoped like :func:`mark_read`: other users' + buckets are left untouched. + + A failed write raises rather than being swallowed: unlike a read flag, a + clear the user was told succeeded would quietly come back on the next poll. + """ + if not user_id: + return + key = str(int(user_id)) + async with _write_lock: + data = _read_all() + if key not in data: + return + del data[key] + _write_all(data) + + def user_for_chat(chat_id: Any) -> int | None: """The dashboard user behind a Telegram chat id, or None if there is none. diff --git a/condor/routine_store.py b/condor/routine_store.py index f70e8ff86..bb58cb856 100644 --- a/condor/routine_store.py +++ b/condor/routine_store.py @@ -43,6 +43,7 @@ iter_agent_slugs, shared_routines_roots, ) +from condor.runtime import code_guard from condor.telemetry import taps as telemetry_taps from routines.base import ( RoutineResult, @@ -764,12 +765,24 @@ async def _execute_and_record( source_name=base_name, ), primitives.bind_context(ctx), + # Routines are agent-written code on the main loop too: the + # same policy as run_code, and exit() is an error, not a + # shutdown (condor/runtime/code_guard.py). + code_guard.guarded(code_guard.routine_scope(instance_id)), + code_guard.contain_exit(), ): cfg = routine.config_class(**config) raw = await routine.run_fn(cfg, ctx) result = normalize_result(raw) except asyncio.CancelledError: result = RoutineResult(text="Stopped by user") + except code_guard.CodeRunDeadline as e: + # A routine called under a snippet's deadline ran past it. Recorded + # like any failure: a BaseException passing through here would + # leave the instance "running" with no result, forever. + logger.warning(f"Routine {routine.name}[{instance_id}] interrupted: {e}") + result = RoutineResult(text=f"Error: CodeRunDeadline: {e}") + failed = True except Exception as e: tb = traceback.format_exc() logger.error( @@ -879,10 +892,13 @@ async def execute( on_complete=on_complete, ) + # Detached: a background run started from a snippet must not inherit + # that snippet's deadline and die with it (code_guard). task = asyncio.create_task( self._run_oneshot( instance_id, routine, config, server_name, user_id, agent=agent - ) + ), + context=code_guard.detached_context(), ) self._tasks[instance_id] = task return instance_id @@ -939,10 +955,13 @@ async def start_continuous( session_key=session_key, ) + # Detached: a background run started from a snippet must not inherit + # that snippet's deadline and die with it (code_guard). task = asyncio.create_task( self._run_continuous( instance_id, routine, config, server_name, user_id, agent=agent - ) + ), + context=code_guard.detached_context(), ) self._tasks[instance_id] = task return instance_id @@ -995,10 +1014,13 @@ async def schedule( schedule={"type": "interval", "interval_sec": interval_sec}, ) + # Detached: a background run started from a snippet must not inherit + # that snippet's deadline and die with it (code_guard). task = asyncio.create_task( self._run_scheduled( instance_id, routine, config, server_name, interval_sec, user_id - ) + ), + context=code_guard.detached_context(), ) self._tasks[instance_id] = task return instance_id diff --git a/condor/runtime/code_guard.py b/condor/runtime/code_guard.py new file mode 100644 index 000000000..d788aa8dd --- /dev/null +++ b/condor/runtime/code_guard.py @@ -0,0 +1,511 @@ +"""Keep agent-written code from taking the main process down with it. + +``run_code`` snippets and routines run on Condor's own event loop — that is the +feature (see ``condor/code_runner.py``): a snippet reaches exactly what a +routine reaches. The cost is that synchronous code in them runs *on* the loop, +so ``asyncio.wait_for`` cannot stop it, and the 2026-10-07 freeze was exactly +that: a snippet's ``glob.glob("/**/x.py", recursive=True)`` walked ``/sys`` +symlink cycles on the main thread for hours, and the dashboard, every agent and +Condor's own SIGTERM handler waited behind it. + +This is not a sandbox and does not try to be: the callers are models making +mistakes, not adversaries, and anything here can be stepped around on purpose. +What it does is turn the mistakes we have actually seen into an error on the +offending line instead of a dead process: + +* **A deadline that can interrupt synchronous code.** Two hooks raise + :class:`CodeRunDeadline` once a guarded run is past its budget: + ``sys.monitoring`` on the snippet's own code objects (a ``while True`` loop, + however tight) and an audit hook on everything the snippet calls (any + syscall: file I/O, ``scandir``, sockets). It is a ``BaseException`` so + ``except Exception`` in the snippet cannot swallow it. Code that is handling + an exception — a ``finally``, an ``except``, a ``with`` exit — may run for up + to :data:`INTERRUPT_GRACE` seconds, so cleanup can release what it holds; + everything else is interrupted again, so a loop that swallows it still ends. +* **A short policy list**, enforced through the same audit hook: no listing of + ``/``, ``/proc``, ``/sys`` or ``/dev``; no ``input()`` (there is no stdin); + no signalling Condor's own process; no ``time.sleep`` longer than + :data:`MAX_BLOCKING_SLEEP` (via a wrapper: ``time.sleep`` has no audit event + before Python 3.13). +* **A static check** (:func:`check_source`) for the exits with no audit event + at all — ``os._exit``, ``os.abort``, ``signal.raise_signal``. +* **``SystemExit`` / ``KeyboardInterrupt`` containment** (:func:`contain_exit`). + asyncio re-raises both out of the event loop, so a bare ``exit()`` in a + snippet used to shut Condor down with exit code 0. + +A run Condor starts on a snippet's behalf to outlive it — a background routine +from ``start_routine`` — must be spawned in :func:`detached_context`; otherwise +its task inherits the snippet's scope and is cut when the snippet's run ends. + +Everything is scoped by a :class:`ContextVar`, so only code running in a guarded +run's context — its own task and any task it spawned — is affected; the rest of +Condor never pays more than one ``ContextVar.get()`` per audit event. + +What it cannot do: interrupt a single long call into C (a huge ``sorted``, a +``subprocess.run`` waiting on its child) or cap memory. Those need the snippet +in another process; until then :mod:`condor.runtime.loop_watchdog` at least +logs where the loop is stuck. +""" + +from __future__ import annotations + +import ast +import contextlib +import contextvars +import dis +import functools +import logging +import os +import sys +import time +from dataclasses import dataclass, field +from types import CodeType +from typing import Iterator + +logger = logging.getLogger(__name__) + +# How long code handling a deadline interruption (a ``finally``, an ``except``, +# a ``with`` exit) may run before it is interrupted too. Long enough to release +# a lock or close a file; short enough that a cleanup that hangs is still cut. +INTERRUPT_GRACE = 2.0 + +# The longest ``time.sleep`` a guarded run may make. Every second of it is a +# second the whole event loop — dashboard, agents, Telegram — stands still. +MAX_BLOCKING_SLEEP = 1.0 + +# Directories no agent has a reason to list, and the root, which is where every +# "find this file" search that ran for hours started. ``/proc`` and ``/sys`` are +# the ones with cycles; ``/dev`` has entries that block on open. +_FORBIDDEN_TREES = ("/proc", "/sys", "/dev") + +# ``sys.monitoring`` tool ids 0-2 and 5 are claimed by convention (debugger, +# coverage, profiler, optimizer). Take a free one of the rest. +_TOOL_IDS = (4, 3) +_TOOL_NAME = "condor-code-guard" + + +class CodeRunDeadline(BaseException): + """A guarded run went past its budget. A ``BaseException`` on purpose.""" + + +class GuardViolation(Exception): + """A guarded run tried something that would take the main process down.""" + + +@dataclass +class GuardScope: + """One guarded run: who it is and when (if ever) it must stop. + + ``deadline`` is a ``time.monotonic()`` value, or ``None`` for code with no + budget (a continuous routine). It is mutable so :meth:`expire` can end a + finished run's stragglers: a task the snippet spawned and never awaited + still carries this scope in its context, and is cut at its next line. + """ + + label: str + deadline: float | None = None + budget: float | None = field(default=None, repr=False) + _interrupted_at: float | None = field(default=None, repr=False) + + def expire(self) -> None: + self.deadline = time.monotonic() + + def should_interrupt(self) -> bool: + """True when the run is past its deadline and is not cleaning up. + + Raising on every line after the deadline meant a ``finally`` never got + to run its first line, and a lock it held stayed held for everyone else. + So code that is handling the interruption is spared — for a while. The + test is the exception being handled, not a timer: the first raise can + land in C code that swallows it, and a timer started then would spend + the cleanup window on the very loop it was meant to stop. + """ + if not self.interrupt_due(): + return False + if self._interrupted_at is None: + self._interrupted_at = time.monotonic() + return True + + def interrupt_due(self) -> bool: + """:meth:`should_interrupt` without recording that it happened.""" + if self.deadline is None: + return False + now = time.monotonic() + if now <= self.deadline: + return False + # Handling *any* exception counts: the interruption itself, the + # CancelledError of a snippet that timed out while awaiting (wait_for + # cut it, not this guard), and the GeneratorExit a suspended + # generator's ``finally`` runs under. + if sys.exception() is None: + return True + # The grace runs from this guard's own interruption, or from the + # deadline when something else (wait_for) did the interrupting. + since = self._interrupted_at or self.deadline + return now - since >= INTERRUPT_GRACE + + +_SCOPE: contextvars.ContextVar[GuardScope | None] = contextvars.ContextVar( + "condor_code_guard_scope", default=None +) + +_installed = False +_tool_id: int | None = None + + +def current_scope() -> GuardScope | None: + return _SCOPE.get() + + +# ── Runtime hooks ── + + +def _deadline(scope: GuardScope, where: str) -> CodeRunDeadline: + budget = f" of {scope.budget:g}s" if scope.budget else "" + return CodeRunDeadline(f"interrupted {where} after its budget{budget} ran out") + + +def _as_path(value) -> str | None: + """An audit-event path argument as an absolute, normalized ``str``. + + Pure string work only: anything that could itself raise an audit event + (``os.getcwd`` inside ``abspath``) would re-enter the hook. Relative paths, + file descriptors and ``None`` (the cwd) are never the root or a system tree, + so they come back ``None`` and are let through. + """ + if isinstance(value, bytes): + value = value.decode(errors="replace") + elif not isinstance(value, str): + fspath = getattr(value, "__fspath__", None) + if fspath is None: + return None + try: + return _as_path(fspath()) + except Exception: + return None + if not value.startswith("/"): + return None + return os.path.normpath(value) + + +def _check_listing(args) -> None: + path = _as_path(args[0]) if args else None + if path is None: + return + if path == "/" or any( + path == tree or path.startswith(tree + "/") for tree in _FORBIDDEN_TREES + ): + raise GuardViolation( + f"listing {path!r} is not allowed here: a search from the filesystem " + "root (or through /proc, /sys, /dev) can run for hours and freezes " + "Condor while it does. Open the exact path you need instead." + ) + + +def _check_kill(args) -> None: + pid = args[0] if args else None + if pid in (0, -1, os.getpid(), -os.getpgrp()): + raise GuardViolation("signalling Condor's own process is not allowed here") + + +def _check_killpg(args) -> None: + if args and args[0] in (0, os.getpgrp()): + raise GuardViolation( + "signalling Condor's own process group is not allowed here" + ) + + +def _check_sleep(secs) -> None: + if isinstance(secs, (int, float)) and secs > MAX_BLOCKING_SLEEP: + raise GuardViolation( + f"time.sleep({secs:g}) would freeze all of Condor for {secs:g}s; " + f"use `await asyncio.sleep({secs:g})` instead" + ) + + +def _deny(message: str): + def check(_args) -> None: + raise GuardViolation(message) + + return check + + +_POLICY = { + "os.scandir": _check_listing, + "os.listdir": _check_listing, + "os.kill": _check_kill, + "os.killpg": _check_killpg, + "signal.pthread_kill": _deny("signalling Condor's threads is not allowed here"), + "builtins.input": _deny( + "input() is not available here: there is no one at stdin, and waiting " + "for it freezes Condor" + ), +} + + +def _audit(event: str, args) -> None: + scope = _SCOPE.get() + if scope is None: + return + if scope.should_interrupt(): + raise _deadline(scope, f"at {event}") + check = _POLICY.get(event) + if check is not None: + check(args) + + +_original_sleep = time.sleep + + +@functools.wraps(_original_sleep) +def _guarded_sleep(secs): + scope = _SCOPE.get() + if scope is not None: + if scope.should_interrupt(): + raise _deadline(scope, "at time.sleep") + _check_sleep(secs) + return _original_sleep(secs) + + +_WATCHED = 0 # set by install(): LINE | JUMP + +# Where an exception enters a handler. The exception is not yet marked as being +# handled there, so an interruption at this instruction would look like one +# outside any cleanup and bypass the handler — a ``with`` block's ``__exit__`` +# never ran, and the lock it would have released stayed held. +_HANDLER_ENTRY = dis.opmap["PUSH_EXC_INFO"] + + +def _arm(code: CodeType) -> None: + sys.monitoring.set_local_events( + _tool_id, code, _WATCHED | sys.monitoring.events.INSTRUCTION + ) + + +def _on_line(code: CodeType, _line: int) -> None: + """Arm :func:`_on_instruction` for ``code`` when an interruption is due. + + Neither this nor :func:`_on_jump` raises. Snippet code is only ever + interrupted from :func:`_on_instruction`, the one place that knows exactly + which instruction is next — and on CPython 3.12 that matters: raised from a + JUMP callback, on a compiler-generated instruction or on a handler's first + instruction, the exception skips the ``finally``/``except``/``__exit__`` it + should have run. + """ + scope = _SCOPE.get() + if scope is not None and scope.interrupt_due(): + _arm(code) + + +def _on_jump(code: CodeType, _src: int, _dst: int) -> None: + """As :func:`_on_line`. Needed because LINE does not reliably fire inside a + one-line loop, and a backward jump always does.""" + scope = _SCOPE.get() + if scope is not None and scope.interrupt_due(): + _arm(code) + + +_unsafe: dict[CodeType, frozenset[int]] = {} + + +def _safe_to_raise(code: CodeType, offset: int) -> bool: + """Whether raising at ``offset`` lands inside the frame's exception table. + + Not at a compiler-generated instruction (no source line — a loop's + back-edge, the shuffling around a handler) and not at a handler's entry. + """ + unsafe = _unsafe.get(code) + if unsafe is None: + lineless = { + off + for start, end, line in code.co_lines() + if line is None + for off in range(start, end, 2) + } + raw = code.co_code + entries = {off for off in range(0, len(raw), 2) if raw[off] == _HANDLER_ENTRY} + unsafe = frozenset(lineless | entries) + if len(_unsafe) > 256: # snippets are throwaway; so is this cache + _unsafe.clear() + _unsafe[code] = unsafe + return offset not in unsafe + + +def _on_instruction(code: CodeType, offset: int) -> None: + """Interrupt at the first instruction where it is safe to, then disarm. + + Stays armed while the interruption is not due — the run is cleaning up — + so the moment cleanup ends, or its grace runs out, is not missed in a + tight loop. Disarmed before raising, so it cannot fire again on the + handler the exception is about to enter. + """ + scope = _SCOPE.get() + if scope is None or not scope.interrupt_due(): + if scope is None: + sys.monitoring.set_local_events(_tool_id, code, _WATCHED) + return + if not _safe_to_raise(code, offset): + return + sys.monitoring.set_local_events(_tool_id, code, _WATCHED) + scope.should_interrupt() # records the interruption + raise _deadline(scope, f"in {code.co_name}") + + +def install() -> None: + """Install the audit hook and claim a ``sys.monitoring`` tool. Idempotent. + + An audit hook can never be removed, which is why :func:`_audit` returns on + its first line for everything outside a guarded run. + """ + global _installed, _tool_id, _WATCHED + if _installed: + return + sys.addaudithook(_audit) + # Process-wide, but a pass-through for everything outside a guarded run — + # including every other thread, whose context never carries a scope. + time.sleep = _guarded_sleep + monitoring = getattr(sys, "monitoring", None) + if monitoring is not None: + for tool_id in _TOOL_IDS: + if monitoring.get_tool(tool_id) is None: + monitoring.use_tool_id(tool_id, _TOOL_NAME) + events = monitoring.events + monitoring.register_callback(tool_id, events.LINE, _on_line) + monitoring.register_callback(tool_id, events.JUMP, _on_jump) + monitoring.register_callback( + tool_id, events.INSTRUCTION, _on_instruction + ) + _WATCHED = events.LINE | events.JUMP + _tool_id = tool_id + break + else: + logger.warning( + "No free sys.monitoring tool id; run_code deadlines will only " + "interrupt snippets at their next syscall, not in pure-CPU loops" + ) + _installed = True + + +def _code_objects(code: CodeType) -> Iterator[CodeType]: + yield code + for const in code.co_consts: + if isinstance(const, CodeType): + yield from _code_objects(const) + + +def _watch_lines(code: CodeType) -> None: + """Watch every line and jump of ``code`` for a passed deadline. + + Only the snippet's own code objects — the functions, lambdas and + comprehensions it defines included — so library code it calls runs at full + speed. INSTRUCTION events are armed per code object only once an + interruption is due (see :func:`_on_jump`), never before. + """ + if _tool_id is None: + return + for obj in _code_objects(code): + sys.monitoring.set_local_events(_tool_id, obj, _WATCHED) + + +@contextlib.contextmanager +def guarded(scope: GuardScope, code: CodeType | None = None) -> Iterator[GuardScope]: + """Run the body (and every task it spawns) under ``scope``. + + ``code`` is the compiled snippet; passing it lets the deadline interrupt + pure-CPU loops in it. Without it the deadline still fires at the next + syscall. + """ + install() + if code is not None and scope.deadline is not None: + _watch_lines(code) + token = _SCOPE.set(scope) + try: + yield scope + finally: + _SCOPE.reset(token) + + +def routine_scope(label: str) -> GuardScope: + """The scope a routine runs under: the policy, and its caller's deadline. + + A routine has no budget of its own (continuous ones run until stopped), but + one called from a snippet via ``call_routine`` must not outlive the snippet + that is waiting on it. + """ + parent = _SCOPE.get() + if parent is not None: + return GuardScope(label=label, deadline=parent.deadline, budget=parent.budget) + return GuardScope(label=label) + + +def detached_context() -> contextvars.Context: + """A copy of the current context with no guard scope in it. + + For ``asyncio.create_task(..., context=detached_context())`` wherever Condor + starts a run meant to outlive its caller. Without it, a background routine + started from a snippet carried that snippet's deadline and was cut — at its + next I/O, past ``except Exception`` — once the snippet finished. + """ + context = contextvars.copy_context() + context.run(_SCOPE.set, None) + return context + + +@contextlib.contextmanager +def contain_exit() -> Iterator[None]: + """Turn ``SystemExit`` / ``KeyboardInterrupt`` into an ordinary error. + + asyncio's task step re-raises both straight out of the event loop, which + ends ``asyncio.run`` and with it Condor — so ``exit()`` in a snippet was a + clean shutdown with exit code 0 and nothing in the log. + """ + try: + yield + except (SystemExit, KeyboardInterrupt) as e: + raise GuardViolation( + f"{type(e).__name__} is not allowed here: it would stop Condor. " + "Set `result` and let the code finish instead." + ) from None + + +# ── Static check ── + +# Calls that end or crash the process and raise no audit event first, so the +# hook never sees them. Matched by (module, attribute) on ``module.attr(...)`` +# and on ``from module import attr``. +_PROCESS_ENDERS = { + ("os", "_exit"), + ("os", "abort"), + ("posix", "_exit"), + ("posix", "abort"), + ("signal", "raise_signal"), + ("faulthandler", "_sigsegv"), + ("faulthandler", "_sigabrt"), +} + + +def check_source(tree: ast.AST) -> str | None: + """A reason to refuse this code before it runs, or ``None``. + + Deliberately narrow: only calls that end the process outright. Everything + else is the runtime hooks' job, where the error lands on the real line. + """ + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom) and node.module: + for alias in node.names: + if (node.module, alias.name) in _PROCESS_ENDERS: + return _ender_message(node.module, alias.name, node.lineno) + elif ( + isinstance(node, ast.Attribute) + and isinstance(node.value, ast.Name) + and (node.value.id, node.attr) in _PROCESS_ENDERS + ): + return _ender_message(node.value.id, node.attr, node.lineno) + return None + + +def _ender_message(module: str, name: str, lineno: int) -> str: + return ( + f"line {lineno}: {module}.{name} is not allowed: it would stop Condor's " + "main process (the dashboard and every agent with it)" + ) diff --git a/condor/runtime/loop_watchdog.py b/condor/runtime/loop_watchdog.py new file mode 100644 index 000000000..47b3e96c2 --- /dev/null +++ b/condor/runtime/loop_watchdog.py @@ -0,0 +1,94 @@ +"""Say where the event loop is stuck, while it is stuck. + +Condor runs the dashboard, Telegram, every agent and every routine on one +asyncio loop, so anything that blocks it freezes all of them at once — and +because the loop is also where log lines are produced, a freeze used to look +like silence. Diagnosing the 2026-10-07 freeze took ``sudo py-spy dump``. + +This is that dump, done automatically: a heartbeat scheduled on the loop, and a +daemon thread that, once the heartbeat is :data:`STALL_AFTER` seconds late, logs +the loop thread's current stack — once per stall, and again when it recovers. + +It only observes. :mod:`condor.runtime.code_guard` is what interrupts agent +code; this covers everything it cannot (a long call into C, a blocking call in +Condor's own code) by at least naming the line. +""" + +from __future__ import annotations + +import asyncio +import logging +import sys +import threading +import time +import traceback + +logger = logging.getLogger(__name__) + +STALL_AFTER = 10.0 +_BEAT = 1.0 + + +class LoopWatchdog: + def __init__( + self, loop: asyncio.AbstractEventLoop, stall_after: float = STALL_AFTER + ): + self._loop = loop + self._stall_after = stall_after + # Fine enough to see a stall of ``stall_after``, never busier than 1s. + self._interval = min(_BEAT, stall_after / 4) + self._loop_thread = threading.get_ident() + self._last_beat = time.monotonic() + self._stop = threading.Event() + self._handle: asyncio.TimerHandle | None = None + self._thread = threading.Thread( + target=self._watch, name="condor-loop-watchdog", daemon=True + ) + + def start(self) -> None: + self._beat() + self._thread.start() + + def stop(self) -> None: + self._stop.set() + if self._handle is not None: + self._handle.cancel() + + def _beat(self) -> None: + self._last_beat = time.monotonic() + if not self._stop.is_set(): + self._handle = self._loop.call_later(self._interval, self._beat) + + def _stack(self) -> str: + frame = sys._current_frames().get(self._loop_thread) + if frame is None: + return " (loop thread not found)\n" + return "".join(traceback.format_stack(frame)) + + def _watch(self) -> None: + stalled_since: float | None = None + while not self._stop.wait(self._interval): + lag = time.monotonic() - self._last_beat + if lag >= self._stall_after and stalled_since is None: + stalled_since = self._last_beat + logger.error( + "Event loop blocked for %.0fs — dashboard, agents and Telegram " + "are frozen until it returns. Loop thread is at:\n%s", + lag, + self._stack(), + ) + elif lag < self._interval * 2 and stalled_since is not None: + logger.warning( + "Event loop unblocked after %.0fs", + time.monotonic() - stalled_since, + ) + stalled_since = None + + +def start( + loop: asyncio.AbstractEventLoop | None = None, stall_after: float = STALL_AFTER +) -> LoopWatchdog: + """Start watching ``loop`` (default: the running one). Call from its thread.""" + watchdog = LoopWatchdog(loop or asyncio.get_running_loop(), stall_after) + watchdog.start() + return watchdog diff --git a/condor/web/routes/notifications.py b/condor/web/routes/notifications.py index 61e5d7ee6..db7393cb3 100644 --- a/condor/web/routes/notifications.py +++ b/condor/web/routes/notifications.py @@ -11,13 +11,17 @@ from __future__ import annotations -from fastapi import APIRouter, Depends, Query +import logging + +from fastapi import APIRouter, Depends, HTTPException, Query from pydantic import BaseModel from condor import notifications from condor.web.auth import get_current_user from condor.web.models import WebUser +log = logging.getLogger(__name__) + router = APIRouter(prefix="/notifications", tags=["notifications"]) @@ -46,3 +50,14 @@ async def mark_notifications_read( ): """Mark some (or all) of this user's notifications read.""" return {"unread": notifications.mark_read(user.id, body.ids)} + + +@router.delete("") +async def clear_notifications(user: WebUser = Depends(get_current_user)): + """Delete all of this user's notifications.""" + try: + await notifications.clear(user.id) + except Exception: # noqa: BLE001 - the path belongs in the log, not the response + log.exception("Could not clear notifications for %s", user.id) + raise HTTPException(status_code=500, detail="Could not clear notifications") + return {"unread": 0} diff --git a/frontend/src/components/layout/NotificationBell.test.tsx b/frontend/src/components/layout/NotificationBell.test.tsx new file mode 100644 index 000000000..432338473 --- /dev/null +++ b/frontend/src/components/layout/NotificationBell.test.tsx @@ -0,0 +1,189 @@ +/** + * The bell's "Clear all" (PR #281). + * + * Clearing deletes the history for good, so what is pinned here is that it + * only ever happens on a deliberate second click, that a failure says so, and + * that the list shown afterwards is the store's and not a guess. + * + * Needs a DOM, so this file overrides vitest's default `node` environment. + * + * @vitest-environment jsdom + */ + +import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; +import { act, type ReactNode } from "react"; +import { createRoot, type Root } from "react-dom/client"; +import { MemoryRouter } from "react-router-dom"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import type { AppNotification, NotificationsResponse } from "@/lib/api"; + +const getNotifications = vi.fn(); +const markNotificationsRead = vi.fn(); +const clearNotifications = vi.fn(); + +vi.mock("@/lib/api", () => ({ + api: { + getNotifications: (...a: unknown[]) => getNotifications(...a), + markNotificationsRead: (...a: unknown[]) => markNotificationsRead(...a), + clearNotifications: (...a: unknown[]) => clearNotifications(...a), + }, +})); + +vi.mock("@/lib/auth", () => ({ useAuth: () => ({ token: "t" }) })); +vi.mock("@/lib/appBadge", () => ({ setAppBadge: () => {} })); +vi.mock("@/lib/push", () => ({ onPushNavigate: () => () => {} })); +// The socket hook is a whole subsystem; the bell only needs its cache key. +vi.mock("@/hooks/useChatSocket", () => ({ NOTIFICATIONS_KEY: ["notifications"] })); +// Positioning is AnchoredMenu's own concern (and its own tests'); here it only +// has to show its children while open. +vi.mock("@/components/ui/AnchoredMenu", () => ({ + AnchoredMenu: ({ open, children }: { open: boolean; children: ReactNode }) => + open ?
{children}
: null, +})); + +const { NotificationBell } = await import("./NotificationBell"); + +declare global { + var IS_REACT_ACT_ENVIRONMENT: boolean; +} + +function note(over: Partial = {}): AppNotification { + return { + id: "n1", + user_id: 1, + ts: Date.now() / 1000, + kind: "system", + text: "Condor is online and ready.", + title: null, + link: null, + read: true, + ...over, + } as AppNotification; +} + +function listing(items: AppNotification[]): NotificationsResponse { + return { items, unread: items.filter((n) => !n.read).length }; +} + +let container: HTMLDivElement; +let root: Root; + +async function flush() { + await act(async () => { + await new Promise((r) => setTimeout(r, 0)); + }); +} + +function byLabel(label: string): HTMLButtonElement | null { + return container.querySelector(`button[title="${label}"]`); +} + +function byText(text: string): HTMLButtonElement | null { + return ( + Array.from(container.querySelectorAll("button")).find( + (b) => b.textContent?.trim() === text, + ) ?? null + ); +} + +async function click(el: Element | null) { + expect(el).not.toBeNull(); + await act(async () => { + (el as HTMLElement).click(); + }); + await flush(); +} + +async function render() { + const client = new QueryClient({ + defaultOptions: { queries: { retry: false }, mutations: { retry: false } }, + }); + await act(async () => { + root.render( + + + + + , + ); + }); + await flush(); +} + +const openBell = () => click(container.querySelector('button[aria-haspopup="menu"]')); + +beforeEach(() => { + globalThis.IS_REACT_ACT_ENVIRONMENT = true; + container = document.createElement("div"); + document.body.appendChild(container); + root = createRoot(container); + getNotifications.mockReset(); + markNotificationsRead.mockReset().mockResolvedValue({ unread: 0 }); + clearNotifications.mockReset(); +}); + +afterEach(() => { + act(() => root.unmount()); + container.remove(); +}); + +describe("NotificationBell clear all", () => { + it("deletes only on the second click, then re-reads the store", async () => { + getNotifications.mockResolvedValue(listing([note()])); + clearNotifications.mockResolvedValue({ unread: 0 }); + await render(); + await openBell(); + + await click(byLabel("Clear all notifications")); + expect(clearNotifications).not.toHaveBeenCalled(); + + // The store has moved on by the time it is re-read: one arrived after the delete. + getNotifications.mockResolvedValue(listing([note({ id: "n2", text: "Arrived after" })])); + const readsBefore = getNotifications.mock.calls.length; + + await click(byText("Clear")); + + expect(clearNotifications).toHaveBeenCalledTimes(1); + expect(getNotifications.mock.calls.length).toBeGreaterThan(readsBefore); + expect(container.textContent).toContain("Arrived after"); + expect(container.textContent).not.toContain("Condor is online"); + }); + + it("says so when the clear fails, and keeps Clear available to retry", async () => { + getNotifications.mockResolvedValue(listing([note()])); + clearNotifications.mockRejectedValue(new Error("500")); + await render(); + await openBell(); + + await click(byLabel("Clear all notifications")); + await click(byText("Clear")); + + expect(container.querySelector('[role="alert"]')?.textContent).toContain( + "Couldn't clear notifications", + ); + expect(container.textContent).toContain("Condor is online"); + const retry = byText("Clear"); + expect(retry).not.toBeNull(); + expect(retry?.disabled).toBe(false); + }); + + it("disarms the confirm when a notification is opened", async () => { + getNotifications.mockResolvedValue(listing([note({ link: "/bots" })])); + await render(); + await openBell(); + + await click(byLabel("Clear all notifications")); + expect(byText("Clear")).not.toBeNull(); + + const item = Array.from(container.querySelectorAll("button")).find((b) => + b.textContent?.includes("Condor is online and ready."), + ); + await click(item ?? null); + await openBell(); + + expect(byText("Clear")).toBeNull(); + expect(byLabel("Clear all notifications")).not.toBeNull(); + expect(clearNotifications).not.toHaveBeenCalled(); + }); +}); diff --git a/frontend/src/components/layout/NotificationBell.tsx b/frontend/src/components/layout/NotificationBell.tsx index 13296a393..e5741efca 100644 --- a/frontend/src/components/layout/NotificationBell.tsx +++ b/frontend/src/components/layout/NotificationBell.tsx @@ -1,6 +1,6 @@ import { useCallback, useEffect, useState } from "react"; import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; -import { Bell, Bot, CheckCheck, Terminal, Workflow, Zap } from "lucide-react"; +import { Bell, Bot, Loader2, Terminal, Trash2, Workflow, Zap } from "lucide-react"; import { useNavigate } from "react-router-dom"; import { AnchoredMenu } from "@/components/ui/AnchoredMenu"; @@ -48,7 +48,8 @@ export function NotificationBell() { const { token } = useAuth(); const [open, setOpen] = useState(false); const [anchor, setAnchor] = useState(null); - const close = useCallback(() => setOpen(false), []); + // Clearing deletes the history for good, so it takes a second click. + const [confirmingClear, setConfirmingClear] = useState(false); const navigate = useNavigate(); const queryClient = useQueryClient(); @@ -72,7 +73,7 @@ export function NotificationBell() { // The dock icon carries the same number as the bell (FEAT-082). Not a second // source of truth: it is a second render of the count this component already // holds, so it cannot drift from what the dropdown shows, and it follows the - // bell through the poll, a live socket write and marking everything read. + // bell through the poll, a live socket write, marking read and clearing. useEffect(() => setAppBadge(unread), [unread]); // A click on an OS notification lands here too (FEAT-083). `sw.js` focuses @@ -94,17 +95,51 @@ export function NotificationBell() { }, }); + const clearAll = useMutation({ + mutationFn: async () => { + // A poll already in flight was read before the delete; letting it land + // afterwards would put the cleared history straight back. + await queryClient.cancelQueries({ queryKey: NOTIFICATIONS_KEY }); + return api.clearNotifications(); + }, + onSuccess: () => { + queryClient.setQueryData(NOTIFICATIONS_KEY, { items: [], unread: 0 }); + setConfirmingClear(false); + // The empty list is a guess; the store is the answer. A notice recorded + // just after the delete may already have been pushed and then wiped by + // the line above, so re-read rather than leave the guess standing. + void queryClient.invalidateQueries({ queryKey: NOTIFICATIONS_KEY }); + }, + }); + + // Every way out of the dropdown disarms the confirm, so a reopened bell never + // shows a Clear that deletes on its first click. + const resetClear = clearAll.reset; + const close = useCallback(() => { + setOpen(false); + setConfirmingClear(false); + resetClear(); + }, [resetClear]); + + const cancelClear = () => { + setConfirmingClear(false); + clearAll.reset(); + }; + const toggle = () => { - const next = !open; - setOpen(next); + if (open) { + close(); + return; + } + setOpen(true); // Opening the list *is* reading it. The badge is there to pull the user in, // so leaving it lit after they looked would train them to ignore it. - if (next && unread > 0) markRead.mutate(undefined); + if (unread > 0) markRead.mutate(undefined); }; const openItem = (n: AppNotification) => { if (!n.link) return; - setOpen(false); + close(); navigate(n.link); }; @@ -138,18 +173,45 @@ export function NotificationBell() { Notifications - {items.length > 0 && ( - - )} + {items.length > 0 && + (confirmingClear ? ( + + Clear all? + + + + ) : ( + + ))} + {clearAll.isError && ( +
+ Couldn't clear notifications. Try again. +
+ )} + {items.length === 0 && (
Nothing yet. Finished tasks show up here. diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 5f534fc11..09a6f207d 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -4273,6 +4273,10 @@ export const api = { body: JSON.stringify({ ids: ids ?? null }), }), + /** Delete all of this user's notifications. */ + clearNotifications: () => + apiFetch<{ unread: number }>("/api/v1/notifications", { method: "DELETE" }), + // ── Web Push (FEAT-083) ── // // The bell with the window closed. Every one of these is scoped to the JWT diff --git a/main.py b/main.py index 0b2d3b1f8..f908f7f2e 100644 --- a/main.py +++ b/main.py @@ -1255,6 +1255,13 @@ def _signal_handler(): for sig in (signal.SIGINT, signal.SIGTERM): loop.add_signal_handler(sig, _signal_handler) + # Everything runs on this one loop, so a block anywhere is a freeze + # everywhere; the watchdog logs the stuck stack instead of leaving silence. + from condor.runtime import code_guard, loop_watchdog + + code_guard.install() + loop_watchdog.start(loop) + # Run uvicorn as a task web_task = asyncio.create_task(server.serve()) stop_task = asyncio.create_task(shutdown_event.wait()) diff --git a/tests/test_bot_performance.py b/tests/test_bot_performance.py index 95f6bad2f..23c0b451f 100644 --- a/tests/test_bot_performance.py +++ b/tests/test_bot_performance.py @@ -775,6 +775,131 @@ async def get_controller_performance_history(self, **_kw): assert fees == 0.0 +# ── A dead strategy's 0/0 readings (quad_threat_orchestrator, 2026-10-08) ── +# +# A bot whose strategy died stays listed and keeps reporting 0/0. Read as a +# cumulative, that is a fall back to zero, and session figures are differences +# of cumulatives — so a bot adopted mid-life read as -$78k volume and invented +# PnL, and a bot born and killed in one session vanished. + + +def _dead_bot_client(rows_by_ts: list[tuple[str, list[tuple[float, float]]]]): + """Per timestamp, one (realized, volume) per controller, as history rows. + + Built on :func:`_history_client` / :func:`_hist_row` further down. + """ + return _history_client( + [ + _hist_row(ts, f"c{i}", realized, volume) + for ts, controllers in rows_by_ts + for i, (realized, volume) in enumerate(controllers) + ] + ) + + +def _epoch(ts: str) -> float: + from condor.fetchers.bot_performance import _iso_to_epoch + + return _iso_to_epoch(ts) + + +_T0 = "2026-10-08T01:00:00+00:00" +_T1 = "2026-10-08T02:00:00+00:00" +_T2 = "2026-10-08T03:00:00+00:00" # strategy dies after this reading +_T3 = "2026-10-08T04:00:00+00:00" +_T4 = "2026-10-08T05:00:00+00:00" + + +def test_a_dead_strategys_zero_readings_carry_its_last_real_cumulative(): + from condor.fetchers.bot_performance import fetch_instance_history + + client = _dead_bot_client( + [ + (_T0, [(1.0, 100.0)]), + (_T2, [(5.0, 900.0)]), + (_T3, [(0.0, 0.0)]), + (_T4, [(0.0, 0.0)]), + ] + ) + + hist = asyncio.run(fetch_instance_history(client, "quad-e4")) + + assert [(r[1], r[2]) for r in hist] == [(1.0, 100.0), (5.0, 900.0)] + + +def test_a_bot_adopted_mid_life_that_dies_keeps_what_it_did_in_session(): + """The quad symptom: negative volume and invented PnL for the session.""" + from condor.fetchers.bot_performance import fetch_instance_history, slice_history + + client = _dead_bot_client( + [ + (_T0, [(1.0, 100.0)]), + (_T1, [(3.0, 500.0)]), # adopted here + (_T2, [(5.0, 900.0)]), + (_T3, [(0.0, 0.0)]), + ] + ) + hist = asyncio.run(fetch_instance_history(client, "quad-e4")) + + realized, volume, _, _ = slice_history([hist], _epoch(_T1), _epoch(_T4)) + + assert (realized, volume) == (2.0, 400.0) # was (-3.0, -500.0) + + +def test_a_bot_born_and_killed_in_one_session_does_not_vanish(): + from condor.fetchers.bot_performance import fetch_instance_history, slice_history + + client = _dead_bot_client([(_T1, [(4.0, 700.0)]), (_T3, [(0.0, 0.0)])]) + hist = asyncio.run(fetch_instance_history(client, "quad-e4")) + + realized, volume, _, _ = slice_history([hist], _epoch(_T0), _epoch(_T4)) + + assert (realized, volume) == (4.0, 700.0) # was (0.0, 0.0) + + +def test_zeros_before_a_controllers_first_trade_are_kept(): + from condor.fetchers.bot_performance import fetch_instance_history + + client = _dead_bot_client( + [(_T0, [(0.0, 0.0)]), (_T1, [(0.0, 0.0)]), (_T2, [(2.0, 50.0)])] + ) + + hist = asyncio.run(fetch_instance_history(client, "fresh-bot")) + + assert [r[2] for r in hist] == [0.0, 0.0, 50.0] + + +def test_only_the_dead_controller_is_held_its_live_sibling_keeps_moving(): + from condor.fetchers.bot_performance import fetch_instance_history + + client = _dead_bot_client( + [ + (_T1, [(1.0, 100.0), (2.0, 200.0)]), + (_T2, [(0.0, 0.0), (3.0, 300.0)]), # c0 died, c1 still trading + (_T3, [(0.0, 0.0), (4.0, 400.0)]), + ] + ) + + hist = asyncio.run(fetch_instance_history(client, "two-strategies")) + + assert [(r[1], r[2]) for r in hist] == [ + (3.0, 300.0), + (4.0, 400.0), + (5.0, 500.0), + ] + + +def test_a_real_reading_after_a_loss_to_breakeven_is_not_dropped(): + """Only an all-zero reading is dead: realized PnL back at 0 with volume is real.""" + from condor.fetchers.bot_performance import fetch_instance_history + + client = _dead_bot_client([(_T1, [(2.0, 100.0)]), (_T2, [(0.0, 300.0)])]) + + hist = asyncio.run(fetch_instance_history(client, "breakeven")) + + assert [(r[1], r[2]) for r in hist] == [(2.0, 100.0), (0.0, 300.0)] + + # ── Provider wiring ── diff --git a/tests/test_code_guard.py b/tests/test_code_guard.py new file mode 100644 index 000000000..d176b6ca7 --- /dev/null +++ b/tests/test_code_guard.py @@ -0,0 +1,476 @@ +"""Agent-written code cannot freeze or stop the main process (code_guard). + +Each case here is a way a snippet used to take all of Condor down with it — the +2026-10-07 freeze was a recursive glob from ``/`` that walked ``/sys`` on the +main thread for hours — and the property pinned is the same every time: the +snippet fails on its own line, the run is recorded, and the event loop is still +there afterwards. + +Sync tests driving coroutines with ``asyncio.run`` (``pytest-asyncio`` is not +installed in this venv). A ``SystemExit`` escaping a snippet would end +``asyncio.run`` and fail the test with it, which is the point of those cases. +""" + +import asyncio +import builtins +import logging +import os +import time + +import pytest + +from condor import code_runner +from condor.code_runs import CodeRunStore +from condor.runtime import code_guard, loop_watchdog + + +@pytest.fixture +def store(tmp_path, monkeypatch): + written = CodeRunStore(tmp_path / "code_runs") + monkeypatch.setattr(code_runner, "get_code_run_store", lambda: written) + + import config_manager + + async def _client(*a, **kw): + return None + + monkeypatch.setattr(config_manager, "get_client", _client) + return written + + +def run_alongside(code: str, timeout: float = 1.0) -> tuple[dict, int]: + """Run a snippet next to a ticking task; return the run and the ticks seen. + + Ticks after the run prove the loop survived it — a frozen or stopped loop + never comes back to them. + """ + + async def main(): + ticks = 0 + + async def ticker(): + nonlocal ticks + while True: + await asyncio.sleep(0.01) + ticks += 1 + + task = asyncio.create_task(ticker()) + out = await code_runner.execute_code(code, timeout=timeout) + before = ticks + await asyncio.sleep(0.1) + task.cancel() + return out, ticks - before + + return asyncio.run(main()) + + +# ── the deadline interrupts synchronous code ── + + +@pytest.mark.parametrize( + "code", + [ + "while True:\n pass\n", + "def spin():\n n = 0\n while True:\n n += 1\n\nspin()\n", + "[x for x in iter(int, 1)]\n", + ], + ids=["bare-loop", "loop-in-function", "endless-comprehension"], +) +def test_a_pure_cpu_loop_is_cut_at_its_budget(store, code): + started = time.monotonic() + out, ticks_after = run_alongside(code, timeout=1) + + assert out["status"] == "timeout" + assert "Timed out after 1s" in out["error"] + assert 'File ""' in out["traceback"] + assert time.monotonic() - started < 3 + assert ticks_after > 0 + + +def test_a_syscall_loop_is_cut_even_without_line_monitoring(store, monkeypatch): + """The audit hook alone stops it — the guard for library code the snippet calls.""" + monkeypatch.setattr(code_guard, "_tool_id", None) + + out, ticks_after = run_alongside("import os\nwhile True:\n os.listdir('.')\n") + + assert out["status"] == "timeout" + assert "interrupted at os.listdir" in out["error"] + assert ticks_after > 0 + + +def test_output_printed_before_the_cut_is_kept(store): + out, _ = run_alongside("print('started')\nwhile True:\n pass\n") + + assert out["status"] == "timeout" + assert out["stdout"] == "started\n" + + +def test_a_task_the_snippet_left_running_is_stopped_with_it(store): + """A spawned task carries the run's scope and dies with the run.""" + code = ( + "import builtins\n" + "async def forever():\n" + " while True:\n" + " await asyncio.sleep(0.01)\n" + "builtins._guard_orphan = asyncio.create_task(forever())\n" + ) + + async def main(): + out = await code_runner.execute_code("import asyncio\n" + code, timeout=5) + await asyncio.sleep(0.1) + return out, builtins._guard_orphan + + try: + out, orphan = asyncio.run(main()) + finally: + vars(builtins).pop("_guard_orphan", None) + + assert out["status"] == "ok" + assert orphan.done() + assert isinstance(orphan.exception(), code_guard.CodeRunDeadline) + + +_CLEANUP_SHAPES = { + # Timed out while awaiting: wait_for cancels it, not the guard, and the + # cleanup runs under the CancelledError — past the deadline all the same. + "await-finally": ( + "await lock.acquire()\n" + "try:\n" + " await asyncio.sleep(100)\n" + "finally:\n" + " lock.release()\n" + " print('cleanup ran')\n" + ), + "await-async-with": ( + "class Held:\n" + " async def __aenter__(self):\n" + " await lock.acquire()\n" + " async def __aexit__(self, *exc):\n" + " lock.release()\n" + " print('cleanup ran')\n" + "async with Held():\n" + " await asyncio.sleep(100)\n" + ), + "await-finally-does-io": ( + "import os, tempfile\n" + "await lock.acquire()\n" + "try:\n" + " await asyncio.sleep(100)\n" + "finally:\n" + " open(os.path.join(tempfile.gettempdir(), 'guard_io.txt'), 'w').close()\n" + " lock.release()\n" + " print('cleanup ran')\n" + ), + # Busy on the main thread: the guard itself interrupts it. + "finally": ( + "await lock.acquire()\n" + "try:\n" + " while True:\n" + " pass\n" + "finally:\n" + " lock.release()\n" + " print('cleanup ran')\n" + ), + "one-line-loop": ( + "await lock.acquire()\n" + "try:\n" + " while True: pass\n" + "finally:\n" + " lock.release()\n" + " print('cleanup ran')\n" + ), + "loop-in-function": ( + "def spin():\n" + " n = 0\n" + " while True:\n" + " n += 1\n" + "await lock.acquire()\n" + "try:\n" + " spin()\n" + "finally:\n" + " lock.release()\n" + " print('cleanup ran')\n" + ), + "except": ( + "await lock.acquire()\n" + "try:\n" + " while True:\n" + " pass\n" + "except BaseException:\n" + " lock.release()\n" + " print('cleanup ran')\n" + " raise\n" + ), + "with": ( + "import contextlib\n" + "@contextlib.contextmanager\n" + "def held():\n" + " try:\n" + " yield\n" + " finally:\n" + " lock.release()\n" + " print('cleanup ran')\n" + "await lock.acquire()\n" + "with held():\n" + " while True:\n" + " pass\n" + ), +} + + +@pytest.mark.parametrize("shape", list(_CLEANUP_SHAPES)) +def test_cleanup_still_runs_after_a_timeout(store, shape): + """Code handling the interruption gets to run: a lock it holds is released.""" + code = ( + "import asyncio, builtins\nlock = builtins._guard_lock\n" + + _CLEANUP_SHAPES[shape] + ) + + async def main(): + builtins._guard_lock = asyncio.Lock() + out = await code_runner.execute_code(code, timeout=1) + return out, builtins._guard_lock.locked() + + try: + out, still_locked = asyncio.run(main()) + finally: + vars(builtins).pop("_guard_lock", None) + + assert out["status"] == "timeout" + assert out["stdout"] == "cleanup ran\n" + assert not still_locked + + +def test_a_cleanup_that_hangs_is_cut_after_its_grace(store): + code = ( + "try:\n while True:\n pass\nfinally:\n while True:\n pass\n" + ) + started = time.monotonic() + out, ticks_after = run_alongside(code, timeout=1) + + assert out["status"] == "timeout" + assert time.monotonic() - started < 1 + code_guard.INTERRUPT_GRACE + 2 + assert ticks_after > 0 + + +def test_a_snippet_that_swallows_the_interruption_is_still_cut(store): + """The cleanup window is granted once, not per interruption.""" + code = ( + "while True:\n" + " try:\n" + " while True:\n" + " pass\n" + " except BaseException:\n" + " pass\n" + ) + started = time.monotonic() + out, ticks_after = run_alongside(code, timeout=1) + + assert out["status"] == "timeout" + assert time.monotonic() - started < 1 + code_guard.INTERRUPT_GRACE + 2 + assert ticks_after > 0 + + +def test_a_background_routine_outlives_the_snippet_that_started_it(store, monkeypatch): + """start_routine is a real background run: no snippet deadline, no orphan cut.""" + from pydantic import BaseModel + + from condor import routine_store + from condor.routine_store import RoutineStore + from routines.base import RoutineInfo + + class Config(BaseModel): + """A test routine.""" + + async def slow(config, context): + await asyncio.sleep(1.5) # past the 1s snippet budget, and its expiry + os.listdir(".") # an audited call: the guard would cut it here + return "finished" + + routines = RoutineStore() + monkeypatch.setattr(routine_store, "_store", routines) + monkeypatch.setattr( + routine_store, + "get_routine", + {"slow": RoutineInfo(name="slow", config_class=Config, run_fn=slow)}.get, + ) + + async def _quiet(self, *a, **kw): + return None + + monkeypatch.setattr(RoutineStore, "_fire_hooks", _quiet) + monkeypatch.setattr(RoutineStore, "_report_run", _quiet) + + async def main(): + out = await code_runner.execute_code( + "from condor import primitives\n" + "result = await primitives.start_routine('slow')\n", + timeout=1, + ) + for _ in range(100): + meta = routines.get_instance(out["result"]) + if meta and meta["status"] != "running": + return out, meta + await asyncio.sleep(0.05) + return out, routines.get_instance(out["result"]) + + out, meta = asyncio.run(main()) + + assert out["status"] == "ok" + assert meta["status"] == "completed" + assert "finished" in str(meta.get("result_summary") or meta) + + +# ── the freeze that started this ── + + +@pytest.mark.parametrize( + "code", + [ + "import glob\nglob.glob('/**/_theta_derive.py', recursive=True)\n", + "import pathlib\nlist(pathlib.Path('/').rglob('_theta_derive.py'))\n", + "import os\nfor _ in os.walk('/'):\n pass\n", + "import os\nos.listdir('/proc')\n", + "import os\nos.listdir('/sys/class')\n", + ], + ids=["glob-from-root", "rglob-from-root", "walk-from-root", "proc", "sys"], +) +def test_a_filesystem_search_from_the_root_is_refused_at_once(store, code): + started = time.monotonic() + out, ticks_after = run_alongside(code, timeout=60) + + assert out["status"] == "error" + assert "GuardViolation" in out["error"] + assert "exact path" in out["error"] + assert time.monotonic() - started < 2 + assert ticks_after > 0 + + +def test_listing_an_ordinary_directory_still_works(store, tmp_path): + (tmp_path / "a.txt").write_text("x") + + out, _ = run_alongside(f"import os\nresult = os.listdir({str(tmp_path)!r})\n") + + assert out["status"] == "ok" + assert "a.txt" in out["result"] + + +# ── ways to stop the process ── + + +@pytest.mark.parametrize( + "code", + [ + "exit()\n", + "import sys\nsys.exit(0)\n", + "raise SystemExit(3)\n", + "raise KeyboardInterrupt\n", + ], +) +def test_exiting_is_an_error_not_a_shutdown(store, code): + out, ticks_after = run_alongside(code) + + assert out["status"] == "error" + assert "would stop Condor" in out["error"] + assert ticks_after > 0 + + +@pytest.mark.parametrize( + "code", + [ + "import os\nos._exit(0)\n", + "from os import _exit\n_exit(0)\n", + "import os\nif False:\n os.abort()\n", + "import signal\nsignal.raise_signal(signal.SIGTERM)\n", + ], +) +def test_a_hard_exit_is_refused_before_anything_runs(store, code): + out, _ = run_alongside("print('ran')\n" + code) + + assert out["status"] == "error" + assert out["error"].startswith("Refused:") + assert out["stdout"] == "" + + +def test_signalling_condor_itself_is_refused(store): + out, _ = run_alongside("import os, signal\nos.kill(os.getpid(), signal.SIGTERM)\n") + + assert out["status"] == "error" + assert "own process" in out["error"] + + +# ── ways to stall the loop ── + + +def test_a_long_blocking_sleep_is_refused(store): + started = time.monotonic() + out, _ = run_alongside("import time\ntime.sleep(30)\n", timeout=60) + + assert out["status"] == "error" + assert "asyncio.sleep" in out["error"] + assert time.monotonic() - started < 2 + + +def test_a_short_blocking_sleep_is_allowed(store): + out, _ = run_alongside("import time\ntime.sleep(0.01)\nresult = 'slept'\n") + + assert out["status"] == "ok" + assert out["result"] == "slept" + + +def test_input_is_refused(store): + out, _ = run_alongside("input('continue? ')\n") + + assert out["status"] == "error" + assert "input()" in out["error"] + + +def test_help_prints_instead_of_starting_a_pager(store): + out, _ = run_alongside("import sys\nresult = sys.stdout.isatty()\nhelp(len)\n") + + assert out["status"] == "ok" + assert out["result"] == "False" + assert "len" in out["stdout"] + + +# ── nothing outside a guarded run is touched ── + + +def test_code_outside_a_run_is_unaffected(store): + run_alongside("result = 1\n") # the hooks are installed by now + + assert code_guard.current_scope() is None + assert os.listdir("/") + time.sleep(code_guard.MAX_BLOCKING_SLEEP + 0.05) # would raise inside a run + + +def test_a_routine_inherits_the_deadline_of_the_snippet_calling_it(): + parent = code_guard.GuardScope("snippet", deadline=time.monotonic() + 5, budget=5) + + assert code_guard.routine_scope("free").deadline is None + with code_guard.guarded(parent): + nested = code_guard.routine_scope("called") + assert nested.deadline == parent.deadline + + +# ── the watchdog names the stuck line ── + + +def test_the_watchdog_logs_where_the_loop_is_blocked(caplog): + def block_the_loop_here(): + time.sleep(0.8) + + async def main(): + watchdog = loop_watchdog.start(stall_after=0.3) + await asyncio.sleep(0.05) + block_the_loop_here() + await asyncio.sleep(1.5) + watchdog.stop() + + with caplog.at_level(logging.WARNING, logger=loop_watchdog.__name__): + asyncio.run(main()) + + blocked = [r for r in caplog.records if "Event loop blocked" in r.getMessage()] + assert len(blocked) == 1 + assert "block_the_loop_here" in blocked[0].getMessage() + assert any("unblocked" in r.getMessage() for r in caplog.records) diff --git a/tests/test_notifications.py b/tests/test_notifications.py index 5daa29af9..f9f28223f 100644 --- a/tests/test_notifications.py +++ b/tests/test_notifications.py @@ -121,6 +121,43 @@ async def three(): assert unread_count(USER_A) == 0 +def test_clear_empties_only_own_bucket(store): + async def go(): + await record(USER_A, "A1") + await record(USER_A, "A2") + await record(USER_B, "B1") + await notifications.clear(USER_A) + + asyncio.run(go()) + + assert list_for(USER_A) == [] + assert unread_count(USER_A) == 0 + assert [n.text for n in list_for(USER_B)] == ["B1"] + + +def test_clear_failure_is_reported_and_keeps_history(store, monkeypatch): + """A clear that could not be written must not report success.""" + from fastapi import HTTPException + + from condor.web.routes.notifications import clear_notifications + + asyncio.run(record(USER_A, "kept")) + + def broken(_data): + raise OSError("disk full") + + monkeypatch.setattr(notifications, "_write_all", broken) + + with pytest.raises(OSError): + asyncio.run(notifications.clear(USER_A)) + with pytest.raises(HTTPException) as exc: + asyncio.run(clear_notifications(user=WebUser(id=USER_A, role="user"))) + + assert exc.value.status_code == 500 + assert "disk full" not in exc.value.detail + assert [n.text for n in list_for(USER_A)] == ["kept"] + + def test_no_user_records_nothing(store): """A producer with nobody behind it is not an error.""" assert asyncio.run(record(0, "orphan")) is None