Gateway shutdown draining of in-flight cron work

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #58992 fix(gateway): drain in-flight cron delivery on restart instead of dropping it (#58818)

  merged to main

  files: gateway/run.py, tests/gateway/test_cron_shutdown_drain.py

  Show the diff

  • gateway/run.py

    @@ -19737,6 +19737,48 @@ def _start_cron_ticker(stop_event: threading.Event, adapters=None, loop=None, in
         InProcessCronScheduler().start(stop_event, adapters=adapters, loop=loop, interval=interval)
     
     
    +# Upper bound for cooperatively draining the cron ticker on shutdown. The cron
    +# thread delivers via ``safe_schedule_threadsafe`` and blocks on
    +# ``future.result(timeout=60)`` (see cron/scheduler.py::_deliver_result), so a
    +# single in-flight delivery unblocks within ~60s. The extra margin covers the
    +# hop back through run_one_job's bookkeeping.
    +_CRON_SHUTDOWN_DRAIN_TIMEOUT = 65.0
    +
    +# Upper bound for cooperatively draining the housekeeping ticker on shutdown.
    +# Housekeeping periodically refreshes the channel directory via
    +# ``safe_schedule_threadsafe(build_channel_directory(...), loop)`` and blocks on
    +# ``fut.result(timeout=30)`` (see ``_start_gateway_housekeeping``) — the same
    +# loop-scheduled-future pattern as cron. So the cooperative bound must cover
    +# that 30s future (plus margin) rather than the old 5s join, otherwise a
    +# channel-directory refresh in flight at shutdown gets abandoned mid-resolve.
    +# Unlike a dropped cron delivery this is not user-facing (it self-heals on the
    +# next tick), but bounding it correctly keeps the drain honest.
    +_HOUSEKEEPING_SHUTDOWN_DRAIN_TIMEOUT = 35.0
    +
    +
    +async def _await_thread_exit(
    +    thread: Optional[threading.Thread], timeout: float, poll: float = 0.1
    +) -> bool:
    +    """Wait for a daemon thread to exit WITHOUT blocking the event loop.
    +
    +    A synchronous ``thread.join()`` here would freeze the event loop — fatal
    +    for the cron ticker, whose in-flight delivery is a coroutine scheduled onto
    +    *this* loop via ``safe_schedule_threadsafe``. Blocking the loop deadlocks
    +    that delivery (the loop can never run it), so ``join(timeout=5)`` always
    +    times out and the message is silently dropped on restart (#58818).
    +
    +    Polling ``is_alive()`` with ``await asyncio.sleep`` keeps the loop running
    +    so the pending delivery completes, then the ticker sees ``stop_event`` and
    +    exits. Returns True if the thread exited within ``timeout``.
    +    """
    +    if thread is None:
    +        return True
    +    deadline = asyncio.get_running_loop().time() + max(0.0, timeout)
    +    while thread.is_alive() and asyncio.get_running_loop().time() < deadline:
    +        await asyncio.sleep(poll)
    +    return not thread.is_alive()
    +
    +
     async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool = False, verbosity: Optional[int] = 0) -> bool:
         """
         Start the gateway and run until interrupted.
    @@ -20231,14 +20273,28 @@ def restart_signal_handler():
                 logger.error("Gateway exiting with failure: %s", runner.exit_reason)
             return False
         
    -    # Stop cron scheduler + housekeeping cleanly
    +    # Stop cron scheduler + housekeeping cleanly.
    +    #
    +    # These MUST be awaited cooperatively, not join()ed. A cron delivery in
    +    # flight when the gateway restarts is a coroutine scheduled onto THIS event
    +    # loop (safe_schedule_threadsafe); the ticker thread is blocked on its
    +    # future.result(). A synchronous cron_thread.join() would block the loop,
    +    # so that delivery could never run — it timed out and the message was
    +    # silently dropped (#58818). Awaiting keeps the loop alive so the in-flight
    +    # delivery finishes before we tear down.
         cron_stop.set()
         try:
             cron_provider.stop()
         except Exception as e:
             logger.debug("Cron provider stop() error: %s", e)
    -    cron_thread.join(timeout=5)
    -    housekeeping_thread.join(timeout=5)
    +    if not await _await_thread_exit(cron_thread, timeout=_CRON_SHUTDOWN_DRAIN_TIMEOUT):
    +        logger.warning(
    +            "Cron ticker did not exit within %.0fs of shutdown — an in-flight "
    +            "delivery may have been dropped.", _CRON_SHUTDOWN_DRAIN_TIMEOUT,
    +        )
    +    await _await_thread_exit(
    +        housekeeping_thread, timeout=_HOUSEKEEPING_SHUTDOWN_DRAIN_TIMEOUT
    +    )
     
         # Stop the planned-stop watcher (daemon=True so this is belt-and-suspenders).
         _planned_stop_watcher_stop.set()

  • tests/gateway/test_cron_shutdown_drain.py

    @@ -0,0 +1,73 @@
    +"""Regression tests for #58818.
    +
    +On restart the gateway must drain an in-flight cron delivery instead of
    +dropping it. A cron delivery is a coroutine scheduled onto the gateway event
    +loop (``safe_schedule_threadsafe``) while the ticker thread blocks on its
    +future. The shutdown wait therefore must NOT block the loop with a synchronous
    +``thread.join()`` — doing so deadlocks the delivery (the loop can never run it)
    +and the message is silently lost. ``_await_thread_exit`` waits cooperatively so
    +the pending delivery completes first.
    +"""
    +import asyncio
    +import threading
    +
    +import pytest
    +
    +import gateway.run as gateway_run
    +
    +
    [redacted:email]
    +async def test_await_thread_exit_lets_loop_scheduled_delivery_complete():
    +    # Reproduces the drop: the worker schedules a coroutine onto THIS loop and
    +    # blocks on its result, exactly like cron/_deliver_result. A blocking join
    +    # would deadlock it; the cooperative wait lets it finish.
    +    loop = asyncio.get_running_loop()
    +    delivered = threading.Event()
    +    worker_done = threading.Event()
    +
    +    async def _delivery():
    +        await asyncio.sleep(0.05)
    +        delivered.set()
    +        return "ok"
    +
    +    def _cron_worker():
    +        fut = asyncio.run_coroutine_threadsafe(_delivery(), loop)
    +        fut.result(timeout=10)
    +        worker_done.set()
    +
    +    thread = threading.Thread(target=_cron_worker, daemon=True)
    +    thread.start()
    +
    +    exited = await gateway_run._await_thread_exit(thread, timeout=10)
    +
    +    assert exited is True
    +    assert delivered.is_set(), "in-flight delivery coroutine never ran (loop was blocked)"
    +    assert worker_done.is_set()
    +
    +
    [redacted:email]
    +async def test_await_thread_exit_returns_false_on_timeout():
    +    keep_alive = threading.Event()
    +
    +    def _spin():
    +        keep_alive.wait(5)
    +
    +    thread = threading.Thread(target=_spin, daemon=True)
    +    thread.start()
    +    try:
    +        exited = await gateway_run._await_thread_exit(thread, timeout=0.2, poll=0.02)
    +        assert exited is False
    +        assert thread.is_alive()
    +    finally:
    +        keep_alive.set()
    +        thread.join(timeout=2)
    +
    +
    [redacted:email]
    +async def test_await_thread_exit_handles_none_and_dead_thread():
    +    assert await gateway_run._await_thread_exit(None, timeout=1) is True
    +
    +    thread = threading.Thread(target=lambda: None, daemon=True)
    +    thread.start()
    +    thread.join(timeout=2)
    +    assert await gateway_run._await_thread_exit(thread, timeout=1) is True

• #60711 fix(gateway,cron): make shutdown drain see in-flight cron work (#60432)

  merged to main

  files: cron/scheduler.py, gateway/run.py, tests/cron/test_shutdown_interrupt.py, tests/gateway/test_cron_active_work_drain.py, tests/gateway/test_shutdown_cache_cleanup.py, tests/gateway/test_update_cron_drain.py

  Show the diff

  • cron/scheduler.py

    @@ -298,6 +298,102 @@ def _is_token(line: str) -> bool:
     _running_job_ids: set = set()
     _running_lock = threading.Lock()
     
    +# Job IDs the gateway shutdown path force-killed the tool subprocess of
    +# while still in ``_running_job_ids`` (see ``mark_running_jobs_interrupted``
    +# below). ``run_one_job``'s own completion path checks this set before
    +# writing its own ``last_status`` so a cron agent thread that keeps running
    +# in-process after its tool was killed out from under it — and produces a
    +# plausible-looking final response from truncated output — can never
    +# overwrite the interrupted status with a false "ok" (#60432).
    +_interrupted_job_ids: set = set()
    +
    +
    +def get_running_job_ids() -> "frozenset[str]":
    +    """Thread-safe snapshot of cron job IDs currently executing.
    +
    +    A job ID is a member from the moment ``_submit_with_guard`` dispatches
    +    it onto the parallel/sequential pool until ``_process_job`` returns —
    +    i.e. for the job's *entire* run, tool calls included, not just the
    +    ticker's dispatch instant.
    +
    +    The gateway shutdown path (``gateway/run.py::GatewayRunner.
    +    _drain_active_agents``) reads this to treat in-flight cron work as
    +    active the same way it already treats in-flight chat sessions via
    +    ``_running_agents`` — cron jobs run through their own thread pool here,
    +    entirely outside that dict, so without this the drain is structurally
    +    blind to them (#60432).
    +    """
    +    with _running_lock:
    +        return frozenset(_running_job_ids)
    +
    +
    +def mark_running_jobs_interrupted(reason: str) -> list:
    +    """Best-effort: mark every currently in-flight cron job interrupted.
    +
    +    Called by the gateway shutdown path immediately after it force-kills
    +    tool subprocesses (``process_registry.kill_all()``). A job whose tool
    +    subprocess was just killed out from under it must never be allowed to
    +    report success — even though its agent thread is still alive in this
    +    same process and may go on to produce a plausible-looking final
    +    response from the now-truncated tool output.
    +
    +    Records the job IDs in ``_interrupted_job_ids`` BEFORE writing
    +    ``last_status`` so ``run_one_job``'s own eventual completion for the
    +    same job (racing in its own thread) sees the flag and skips its normal
    +    write instead of clobbering this one — see the check near the end of
    +    ``run_one_job``. This does not attempt to correlate the killed
    +    subprocess PID to a specific job ID (the process registry tracks PIDs,
    +    not cron job IDs); any job still dispatched at the moment of a forced
    +    kill is treated as interrupted, matching the coarser precedent already
    +    set by ``GatewayRunner._interrupt_running_agents``, which interrupts
    +    every entry in ``_running_agents`` on a drain timeout without
    +    per-agent correlation either.
    +
    +    Returns the list of job IDs marked, for the caller to log.
    +    """
    +    with _running_lock:
    +        job_ids = list(_running_job_ids)
    +        _interrupted_job_ids.update(job_ids)
    +    marked = []
    +    for job_id in job_ids:
    +        try:
    +            mark_job_run(job_id, False, reason)
    +            marked.append(job_id)
    +        except Exception as e:
    +            logger.warning("Failed to mark job %s interrupted: %s", job_id, e)
    +    return marked
    +
    +
    +def _is_interrupted(job_id: str) -> bool:
    +    """Non-destructive peek at whether the shutdown path has marked
    +    ``job_id`` interrupted (see ``mark_running_jobs_interrupted``).
    +
    +    Called by ``run_one_job`` BEFORE it decides what to deliver — a job
    +    whose tool subprocess was killed mid-flight may still produce a
    +    plausible-looking ``final_response`` from the truncated output, and
    +    that must not go out to the user as if it were a normal result.
    +    Unlike ``_consume_interrupted_flag`` below, this does not clear the
    +    flag: the later, authoritative check (right before ``last_status`` is
    +    written) still needs to see it."""
    +    with _running_lock:
    +        return job_id in _interrupted_job_ids
    +
    +
    +def _consume_interrupted_flag(job_id: str) -> bool:
    +    """Return True and clear the flag if the shutdown path already marked
    +    ``job_id`` interrupted (see ``mark_running_jobs_interrupted``).
    +
    +    Called by ``run_one_job`` right before it would otherwise write its own
    +    ``last_status``. Consuming (discarding) rather than just checking keeps
    +    the flag from leaking across a later, unrelated run of the same job ID
    +    (recurring jobs reuse their ID every fire)."""
    +    with _running_lock:
    +        if job_id in _interrupted_job_ids:
    +            _interrupted_job_ids.discard(job_id)
    +            return True
    +        return False
    +
    +
     # Sequential (env-mutating) cron jobs — workdir jobs that touch
     # process-global runtime state — must run one at a time, but must NOT block the
     # ticker thread.  A persistent single-thread executor preserves ordering across
    @@ -3332,6 +3428,20 @@ def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) -
                 if verbose:
                     logger.info("Output saved to: %s", output_file)
     
    +            # If the gateway shutdown killed this job's tool subprocess
    +            # mid-flight (#60432), the agent may still have produced a
    +            # plausible-looking final_response from the truncated output --
    +            # force the failure path so the delivered message is an honest
    +            # "this run was interrupted" summary instead of that response.
    +            # Peek-only: the flag stays set for the authoritative check
    +            # right before mark_job_run below.
    +            if success and _is_interrupted(job["id"]):
    +                success = False
    +                error = (
    +                    "Interrupted by gateway shutdown before the run finished "
    +                    "(tool subprocess was killed mid-flight)."
    +                )
    +
                 # Deliver the final response to the origin/target chat.
                 # If the agent responded with [SILENT], skip delivery (but
                 # output is already saved above).  Failed jobs always deliver.
    @@ -3370,12 +3480,14 @@ def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) -
                 success = False
                 error = "Agent completed but produced empty response (model error, timeout, or misconfiguration)"
     
    -        mark_job_run(job["id"], success, error, delivery_error=delivery_error)
    +        if not _consume_interrupted_flag(job["id"]):
    +            mark_job_run(job["id"], success, error, delivery_error=delivery_error)
             return True
     
         except Exception as e:
             logger.error("Error processing job %s: %s", job['id'], e)
    -        mark_job_run(job["id"], False, str(e))
    +        if not _consume_interrupted_flag(job["id"]):
    +            mark_job_run(job["id"], False, str(e))
             return False
     
     

  • gateway/run.py

    @@ -4080,6 +4080,26 @@ def _request_clean_exit(self, reason: str) -> None:
         def _running_agent_count(self) -> int:
             return len(self._running_agents)
     
    +    def _active_cron_job_count(self) -> int:
    +        """Count of cron jobs currently executing, from the cron scheduler's
    +        own in-flight tracking (``cron.scheduler._running_job_ids``).
    +
    +        Cron jobs run through a standalone ``AIAgent`` on the scheduler's own
    +        thread pool (``cron/scheduler.py::run_job``), entirely outside
    +        ``self._running_agents`` — the dict every OTHER active-work check on
    +        this class (``_running_agent_count``, ``_drain_active_agents``) reads.
    +        Without this, the shutdown drain is structurally blind to in-flight
    +        cron work: it can report ``active_at_start=0`` and proceed straight
    +        to killing tool subprocesses while a cron job's terminal command is
    +        still running (#60432). Best-effort: returns 0 if the cron module
    +        can't be imported (e.g. a minimal test double for this class).
    +        """
    +        try:
    +            from cron.scheduler import get_running_job_ids
    +            return len(get_running_job_ids())
    +        except Exception:
    +            return 0
    +
         # ── scale-to-zero idle detection / dormant-quiesce (Phase 0) ──────────────
         # The gateway-side BEHAVIOUR that consumes the relay scale-to-zero primitives
         # (gateway-gateway Phase 5). Pure logic lives in gateway/scale_to_zero.py; the
    @@ -5550,18 +5570,31 @@ async def _handle_active_session_busy_message(self, event: MessageEvent, session
         async def _drain_active_agents(self, timeout: float) -> tuple[Dict[str, Any], bool]:
             snapshot = self._snapshot_running_agents()
             last_active_count = self._running_agent_count()
    +        last_cron_count = self._active_cron_job_count()
             last_status_at = 0.0
     
             def _maybe_update_status(force: bool = False) -> None:
    -            nonlocal last_active_count, last_status_at
    +            nonlocal last_active_count, last_cron_count, last_status_at
                 now = asyncio.get_running_loop().time()
                 active_count = self._running_agent_count()
    -            if force or active_count != last_active_count or (now - last_status_at) >= 1.0:
    +            cron_count = self._active_cron_job_count()
    +            if (
    +                force
    +                or active_count != last_active_count
    +                or cron_count != last_cron_count
    +                or (now - last_status_at) >= 1.0
    +            ):
                     self._update_runtime_status("draining")
                     last_active_count = active_count
    +                last_cron_count = cron_count
                     last_status_at = now
     
    -        if not self._running_agents:
    +        # Cron jobs run on the scheduler's own thread pool, outside
    +        # ``self._running_agents`` — fold their in-flight count into the
    +        # same wait/timeout this method already applies to chat sessions,
    +        # or a cron job's tool work gets killed with zero warning the
    +        # instant it's the only active thing running (#60432).
    +        if not self._running_agents and last_cron_count == 0:
                 _maybe_update_status(force=True)
                 return snapshot, False
     
    @@ -5570,10 +5603,13 @@ def _maybe_update_status(force: bool = False) -> None:
                 return snapshot, True
     
             deadline = asyncio.get_running_loop().time() + timeout
    -        while self._running_agents and asyncio.get_running_loop().time() < deadline:
    +        while (
    +            (self._running_agents or self._active_cron_job_count())
    +            and asyncio.get_running_loop().time() < deadline
    +        ):
                 _maybe_update_status()
                 await asyncio.sleep(0.1)
    -        timed_out = bool(self._running_agents)
    +        timed_out = bool(self._running_agents) or bool(self._active_cron_job_count())
             _maybe_update_status(force=True)
             return snapshot, timed_out
     
    @@ -7944,6 +7980,27 @@ def _kill_tool_subprocesses(phase: str) -> None:
                             )
                     except Exception as _e:
                         logger.debug("process_registry.kill_all (%s) error: %s", phase, _e)
    +                try:
    +                    # Any cron job still dispatched at this instant just had
    +                    # its tool subprocess killed above (kill_all() has no
    +                    # per-job-ID targeting — it's a global sweep). Its agent
    +                    # thread is still alive in this process and may go on to
    +                    # produce a plausible-looking final response from the
    +                    # now-truncated tool output; mark the run interrupted so
    +                    # the scheduler can never report that as success (#60432).
    +                    # No-op when no cron job is in flight.
    +                    from cron.scheduler import mark_running_jobs_interrupted
    +                    _interrupted = mark_running_jobs_interrupted(
    +                        f"Gateway shutdown ({phase}) killed the job's tool "
    +                        "subprocess before the run finished."
    +                    )
    +                    if _interrupted:
    +                        logger.warning(
    +                            "Shutdown (%s): marked %d in-flight cron job(s) interrupted: %s",
    +                            phase, len(_interrupted), ", ".join(_interrupted),
    +                        )
    +                except Exception as _e:
    +                    logger.debug("mark_running_jobs_interrupted (%s) error: %s", phase, _e)
                     try:
                         from tools.async_delegation import interrupt_all as _interrupt_async
                         _async_n = _interrupt_async(reason=f"gateway shutdown ({phase})")
    @@ -8004,16 +8061,20 @@ def _phase_elapsed() -> float:
                     except Exception as _e:
                         logger.debug("pre-drain mark_resume_pending failed for %s: %s", _sk, _e)
     
    +            _cron_at_start = self._active_cron_job_count()
                 _drain_started_at = time.monotonic()
                 active_agents, timed_out = await self._drain_active_agents(timeout)
                 logger.info(
                     "Shutdown phase: drain done at +%.2fs (drain took %.2fs, "
    -                "timed_out=%s, active_at_start=%d, active_now=%d)",
    +                "timed_out=%s, active_at_start=%d, active_now=%d, "
    +                "cron_at_start=%d, cron_now=%d)",
                     _phase_elapsed(),
                     time.monotonic() - _drain_started_at,
                     timed_out,
                     len(active_agents),
                     self._running_agent_count(),
    +                _cron_at_start,
    +                self._active_cron_job_count(),
                 )
     
                 if not timed_out:
    @@ -8032,9 +8093,11 @@ def _phase_elapsed() -> float:
     
                 if timed_out:
                     logger.warning(
    -                    "Gateway drain timed out after %.1fs with %d active agent(s); interrupting remaining work.",
    +                    "Gateway drain timed out after %.1fs with %d active agent(s) "
    +                    "and %d in-flight cron job(s); interrupting remaining work.",
                         timeout,
                         self._running_agent_count(),
    +                    self._active_cron_job_count(),
                     )
                     # Mark forcibly-interrupted sessions as resume_pending BEFORE
                     # interrupting the agents.  This preserves each session's

  • tests/cron/test_shutdown_interrupt.py

    @@ -0,0 +1,278 @@
    +"""Tests for #60432: cron jobs must not be silently invisible to gateway
    +shutdown, and a job whose tool subprocess got killed by shutdown must
    +never be reported as a successful run.
    +
    +Covers the cron/scheduler.py primitives directly:
    +  - get_running_job_ids() -- thread-safe snapshot the gateway drain reads
    +  - mark_running_jobs_interrupted() -- called by the gateway right after
    +    it force-kills tool subprocesses
    +  - the interrupted-flag race guard in run_one_job(), which must win over
    +    the job's own thread finishing normally with a plausible-looking
    +    result AFTER its tool was already killed out from under it
    +"""
    +
    +from unittest.mock import patch
    +
    +import pytest
    +
    +
    [redacted:email](autouse=True)
    +def _reset_scheduler_state():
    +    """Every test starts from a clean slate and leaves one behind, since
    +    these sets are module-level globals shared across the test process."""
    +    import cron.scheduler as sched
    +
    +    sched._running_job_ids.clear()
    +    sched._interrupted_job_ids.clear()
    +    yield
    +    sched._running_job_ids.clear()
    +    sched._interrupted_job_ids.clear()
    +
    +
    +class TestGetRunningJobIds:
    +    def test_empty_when_nothing_running(self):
    +        import cron.scheduler as sched
    +
    +        assert sched.get_running_job_ids() == frozenset()
    +
    +    def test_reflects_in_flight_jobs(self):
    +        import cron.scheduler as sched
    +
    +        sched._running_job_ids.add("job-1")
    +        sched._running_job_ids.add("job-2")
    +
    +        result = sched.get_running_job_ids()
    +
    +        assert result == frozenset({"job-1", "job-2"})
    +
    +    def test_snapshot_is_immutable_and_independent(self):
    +        """Mutating _running_job_ids after the call must not change the
    +        already-returned snapshot -- callers (the gateway drain loop) rely
    +        on this to safely count in a tight polling loop."""
    +        import cron.scheduler as sched
    +
    +        sched._running_job_ids.add("job-1")
    +        snapshot = sched.get_running_job_ids()
    +        sched._running_job_ids.add("job-2")
    +
    +        assert snapshot == frozenset({"job-1"})
    +
    +
    +class TestMarkRunningJobsInterrupted:
    +    def test_no_op_when_nothing_running(self):
    +        import cron.scheduler as sched
    +
    +        with patch("cron.scheduler.mark_job_run") as mock_mark:
    +            marked = sched.mark_running_jobs_interrupted("shutdown")
    +
    +        assert marked == []
    +        mock_mark.assert_not_called()
    +
    +    def test_marks_every_in_flight_job(self):
    +        import cron.scheduler as sched
    +
    +        sched._running_job_ids.update({"job-1", "job-2"})
    +
    +        with patch("cron.scheduler.mark_job_run") as mock_mark:
    +            marked = sched.mark_running_jobs_interrupted("gateway shutdown (final-cleanup)")
    +
    +        assert sorted(marked) == ["job-1", "job-2"]
    +        assert mock_mark.call_count == 2
    +        called_ids = {c.args[0] for c in mock_mark.call_args_list}
    +        assert called_ids == {"job-1", "job-2"}
    +        for c in mock_mark.call_args_list:
    +            # success must be False -- an interrupted run is never "ok".
    +            assert c.args[1] is False
    +            assert "gateway shutdown" in c.args[2]
    +
    +    def test_sets_interrupted_flag_for_consumption_by_run_one_job(self):
    +        import cron.scheduler as sched
    +
    +        sched._running_job_ids.add("job-1")
    +
    +        with patch("cron.scheduler.mark_job_run"):
    +            sched.mark_running_jobs_interrupted("shutdown")
    +
    +        assert "job-1" in sched._interrupted_job_ids
    +
    +    def test_one_job_marking_failure_does_not_block_the_others(self):
    +        """mark_job_run raising for one job (e.g. a jobs.json write race)
    +        must not prevent the rest from being marked -- this runs during
    +        shutdown, there's no retry window."""
    +        import cron.scheduler as sched
    +
    +        sched._running_job_ids.update({"job-1", "job-2"})
    +
    +        def _side_effect(job_id, success, reason, **kwargs):
    +            if job_id == "job-1":
    +                raise OSError("disk full")
    +
    +        with patch("cron.scheduler.mark_job_run", side_effect=_side_effect):
    +            marked = sched.mark_running_jobs_interrupted("shutdown")
    +
    +        assert marked == ["job-2"]
    +
    +
    +class TestIsInterrupted:
    +    """Peek-only check used at the delivery gate -- must NOT clear the
    +    flag, unlike _consume_interrupted_flag."""
    +
    +    def test_false_when_not_marked(self):
    +        import cron.scheduler as sched
    +
    +        assert sched._is_interrupted("job-1") is False
    +
    +    def test_true_when_marked(self):
    +        import cron.scheduler as sched
    +
    +        sched._interrupted_job_ids.add("job-1")
    +
    +        assert sched._is_interrupted("job-1") is True
    +
    +    def test_does_not_clear_the_flag(self):
    +        import cron.scheduler as sched
    +
    +        sched._interrupted_job_ids.add("job-1")
    +
    +        sched._is_interrupted("job-1")
    +
    +        # Still set -- the later, authoritative check before mark_job_run
    +        # must still see it.
    +        assert "job-1" in sched._interrupted_job_ids
    +        assert sched._is_interrupted("job-1") is True
    +
    +
    +class TestConsumeInterruptedFlag:
    +    def test_false_when_not_marked(self):
    +        import cron.scheduler as sched
    +
    +        assert sched._consume_interrupted_flag("job-1") is False
    +
    +    def test_true_and_clears_when_marked(self):
    +        import cron.scheduler as sched
    +
    +        sched._interrupted_job_ids.add("job-1")
    +
    +        assert sched._consume_interrupted_flag("job-1") is True
    +        # Consumed -- a second check (e.g. a later, unrelated fire of the
    +        # same recurring job ID) must not still read as interrupted.
    +        assert sched._consume_interrupted_flag("job-1") is False
    +
    +
    +class TestRunOneJobHonoursInterruptedFlag:
    +    """run_one_job() must not let a job's own completion overwrite a
    +    status the shutdown path already wrote for the same run."""
    +
    +    def _make_job(self, job_id="job-1"):
    +        return {"id": job_id, "name": "test job", "prompt": "do work"}
    +
    +    def test_success_path_skipped_when_interrupted(self):
    +        import cron.scheduler as sched
    +
    +        job = self._make_job()
    +        sched._interrupted_job_ids.add(job["id"])
    +
    +        with patch("cron.scheduler.claim_dispatch", return_value=True), \
    +             patch("agent.secret_scope.set_secret_scope", return_value=None), \
    +             patch("agent.secret_scope.build_profile_secret_scope", return_value=None), \
    +             patch("agent.secret_scope.reset_secret_scope"), \
    +             patch(
    +                 "cron.scheduler.run_job",
    +                 return_value=(True, "full output", "final response", None),
    +             ), \
    +             patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \
    +             patch("cron.scheduler._is_cron_silence_response", return_value=False), \
    +             patch("cron.scheduler._deliver_result", return_value=None), \
    +             patch("cron.scheduler.mark_job_run") as mock_mark:
    +            result = sched.run_one_job(job)
    +
    +        assert result is True
    +        # The would-be "success" write must NOT happen -- the shutdown
    +        # path already wrote the authoritative interrupted status.
    +        mock_mark.assert_not_called()
    +        # Flag is consumed so a later, unrelated fire of the same job ID
    +        # isn't permanently silenced.
    +        assert job["id"] not in sched._interrupted_job_ids
    +
    +    def test_interrupted_job_delivers_failure_summary_not_raw_response(self):
    +        """The status-write guard alone isn't enough: delivery happens
    +        BEFORE mark_job_run in run_one_job's own flow, so a job that kept
    +        running post-kill and produced a plausible-looking final_response
    +        must not have that response sent to the user just because the
    +        eventual status write gets suppressed. Interrupted jobs must route
    +        through the same failure-summary delivery path a real failure
    +        would."""
    +        import cron.scheduler as sched
    +
    +        job = self._make_job()
    +        sched._interrupted_job_ids.add(job["id"])
    +
    +        with patch("cron.scheduler.claim_dispatch", return_value=True), \
    +             patch("agent.secret_scope.set_secret_scope", return_value=None), \
    +             patch("agent.secret_scope.build_profile_secret_scope", return_value=None), \
    +             patch("agent.secret_scope.reset_secret_scope"), \
    +             patch(
    +                 "cron.scheduler.run_job",
    +                 return_value=(True, "full output", "a plausible final response", None),
    +             ), \
    +             patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \
    +             patch(
    +                 "cron.scheduler._summarize_cron_failure_for_delivery",
    +                 return_value="This run was interrupted.",
    +             ) as mock_summarize, \
    +             patch("cron.scheduler._is_cron_silence_response", return_value=False), \
    +             patch("cron.scheduler._deliver_result", return_value=None) as mock_deliver, \
    +             patch("cron.scheduler.mark_job_run"):
    +            result = sched.run_one_job(job)
    +
    +        assert result is True
    +        mock_summarize.assert_called_once()
    +        # The summarizer's error argument must mention the interruption,
    +        # not be silently None / the agent's own (possibly absent) error.
    +        assert "interrupt" in mock_summarize.call_args.args[1].lower()
    +        delivered_content = mock_deliver.call_args.args[1]
    +        assert delivered_content == "This run was interrupted."
    +        assert "plausible final response" not in delivered_content
    +
    +    def test_success_path_writes_normally_when_not_interrupted(self):
    +        """Control case: the guard must not swallow ordinary, un-interrupted
    +        completions -- only ones the shutdown path explicitly flagged."""
    +        import cron.scheduler as sched
    +
    +        job = self._make_job()
    +
    +        with patch("cron.scheduler.claim_dispatch", return_value=True), \
    +             patch("agent.secret_scope.set_secret_scope", return_value=None), \
    +             patch("agent.secret_scope.build_profile_secret_scope", return_value=None), \
    +             patch("agent.secret_scope.reset_secret_scope"), \
    +             patch(
    +                 "cron.scheduler.run_job",
    +                 return_value=(True, "full output", "final response", None),
    +             ), \
    +             patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \
    +             patch("cron.scheduler._is_cron_silence_response", return_value=False), \
    +             patch("cron.scheduler._deliver_result", return_value=None), \
    +             patch("cron.scheduler.mark_job_run") as mock_mark:
    +            result = sched.run_one_job(job)
    +
    +        assert result is True
    +        mock_mark.assert_called_once()
    +        assert mock_mark.call_args.args[0] == job["id"]
    +        assert mock_mark.call_args.args[1] is True  # success
    +
    +    def test_exception_path_also_honours_interrupted_flag(self):
    +        import cron.scheduler as sched
    +
    +        job = self._make_job()
    +        sched._interrupted_job_ids.add(job["id"])
    +
    +        with patch("cron.scheduler.claim_dispatch", return_value=True), \
    +             patch("agent.secret_scope.set_secret_scope", return_value=None), \
    +             patch("agent.secret_scope.build_profile_secret_scope", return_value=None), \
    +             patch("agent.secret_scope.reset_secret_scope"), \
    +             patch("cron.scheduler.run_job", side_effect=RuntimeError("boom")), \
    +             patch("cron.scheduler.mark_job_run") as mock_mark:
    +            result = sched.run_one_job(job)
    +
    +        assert result is False
    +        mock_mark.assert_not_called()

  • tests/gateway/test_cron_active_work_drain.py

    @@ -0,0 +1,185 @@
    +"""Tests for #60432: the gateway shutdown drain was structurally blind to
    +in-flight cron work. Cron jobs run through cron/scheduler.py's own thread
    +pool, entirely outside ``GatewayRunner._running_agents`` -- the dict every
    +other active-work check on this class reads. A shutdown (``/update``,
    +``/restart``, SIGUSR1 -- they all funnel through the same ``stop()``) could
    +report ``active_at_start=0`` and immediately kill tool subprocesses while a
    +cron job's terminal command was still running.
    +
    +These tests cover the gateway side of the fix:
    +  - _active_cron_job_count() reads cron.scheduler's in-flight job set
    +  - _drain_active_agents() waits for cron work the same way it already
    +    waits for chat sessions
    +  - the final tool-subprocess kill marks any still-in-flight cron job
    +    interrupted
    +
    +See tests/cron/test_shutdown_interrupt.py for the cron-side primitives
    +this relies on (get_running_job_ids, mark_running_jobs_interrupted).
    +"""
    +
    +import asyncio
    +from unittest.mock import MagicMock, patch
    +
    +import pytest
    +
    +from tests.gateway.restart_test_helpers import make_restart_runner
    +
    +
    [redacted:email](autouse=True)
    +def _reset_cron_running_set():
    +    import cron.scheduler as sched
    +
    +    sched._running_job_ids.clear()
    +    sched._interrupted_job_ids.clear()
    +    yield
    +    sched._running_job_ids.clear()
    +    sched._interrupted_job_ids.clear()
    +
    +
    +def _make_async_noop():
    +    async def _noop(*args, **kwargs):
    +        return None
    +
    +    return _noop
    +
    +
    +class TestActiveCronJobCount:
    +    def test_zero_when_no_cron_jobs_running(self):
    +        runner, _adapter = make_restart_runner()
    +        assert runner._active_cron_job_count() == 0
    +
    +    def test_reflects_cron_scheduler_state(self):
    +        import cron.scheduler as sched
    +
    +        runner, _adapter = make_restart_runner()
    +        sched._running_job_ids.add("job-1")
    +
    +        assert runner._active_cron_job_count() == 1
    +
    +    def test_never_raises_if_cron_module_unavailable(self):
    +        """Best-effort: a broken/absent import must not take shutdown
    +        counting down with it."""
    +        runner, _adapter = make_restart_runner()
    +
    +        with patch(
    +            "cron.scheduler.get_running_job_ids", side_effect=ImportError("boom")
    +        ):
    +            assert runner._active_cron_job_count() == 0
    +
    +
    +class TestDrainWaitsForCronWork:
    +    @pytest.mark.asyncio
    +    async def test_drain_returns_immediately_when_nothing_active(self):
    +        runner, _adapter = make_restart_runner()
    +
    +        _snapshot, timed_out = await runner._drain_active_agents(5.0)
    +
    +        assert timed_out is False
    +
    +    @pytest.mark.asyncio
    +    async def test_drain_waits_for_in_flight_cron_job(self):
    +        """Before this fix, a cron-only workload made active_at_start=0
    +        and the drain returned instantly -- this is the exact repro from
    +        the issue (a `sleep 1800` cron job in flight during /update)."""
    +        import cron.scheduler as sched
    +
    +        runner, _adapter = make_restart_runner()
    +        sched._running_job_ids.add("job-1")
    +
    +        async def finish_job():
    +            await asyncio.sleep(0.12)
    +            sched._running_job_ids.discard("job-1")
    +
    +        task = asyncio.create_task(finish_job())
    +        _snapshot, timed_out = await runner._drain_active_agents(2.0)
    +        await task
    +
    +        assert timed_out is False, (
    +            "drain must wait for the cron job to finish, not report "
    +            "active_at_start=0 and return instantly"
    +        )
    +
    +    @pytest.mark.asyncio
    +    async def test_drain_times_out_if_cron_job_outlives_the_window(self):
    +        import cron.scheduler as sched
    +
    +        runner, _adapter = make_restart_runner()
    +        sched._running_job_ids.add("job-1")  # never removed within the window
    +
    +        _snapshot, timed_out = await runner._drain_active_agents(0.1)
    +
    +        assert timed_out is True
    +
    +    @pytest.mark.asyncio
    +    async def test_drain_still_waits_for_chat_sessions_unchanged(self):
    +        """Regression guard: folding cron into the check must not break
    +        the pre-existing chat-session drain behavior."""
    +        runner, _adapter = make_restart_runner()
    +        runner._running_agents = {"session-1": MagicMock()}
    +
    +        async def finish_agent():
    +            await asyncio.sleep(0.12)
    +            runner._running_agents.clear()
    +
    +        task = asyncio.create_task(finish_agent())
    +        _snapshot, timed_out = await runner._drain_active_agents(2.0)
    +        await task
    +
    +        assert timed_out is False
    +
    +
    +class TestKillToolSubprocessesMarksCronInterrupted:
    +    @pytest.mark.asyncio
    +    async def test_in_flight_cron_job_marked_interrupted_on_forced_kill(self, monkeypatch):
    +        import cron.scheduler as sched
    +        import tools.process_registry as _pr
    +        import tools.terminal_tool as _tt
    +        import tools.browser_tool as _bt
    +
    +        runner, adapter = make_restart_runner()
    +        runner._restart_drain_timeout = 0.01  # force the timeout path
    +        adapter.disconnect = _make_async_noop()
    +
    +        sched._running_job_ids.add("job-1")
    +
    +        monkeypatch.setattr(_pr.process_registry, "kill_all", lambda task_id=None: 1)
    +        monkeypatch.setattr(_tt, "cleanup_all_environments", lambda: None)
    +        monkeypatch.setattr(_bt, "cleanup_all_browsers", lambda: None)
    +
    +        marked_calls = []
    +        real_mark = sched.mark_running_jobs_interrupted
    +
    +        def _spy(reason):
    +            result = real_mark(reason)
    +            marked_calls.append((reason, result))
    +            return result
    +
    +        monkeypatch.setattr(sched, "mark_running_jobs_interrupted", _spy)
    +
    +        with patch("gateway.status.remove_pid_file"), patch("gateway.status.write_runtime_status"), \
    +             patch("cron.scheduler.mark_job_run"):
    +            await runner.stop()
    +
    +        assert marked_calls, "mark_running_jobs_interrupted was never called during shutdown"
    +        assert any(result == ["job-1"] for _reason, result in marked_calls)
    +
    +    @pytest.mark.asyncio
    +    async def test_no_cron_jobs_running_is_a_silent_no_op(self, monkeypatch):
    +        """Graceful shutdown with nothing in flight must not spuriously
    +        mark or log anything cron-related."""
    +        import tools.process_registry as _pr
    +        import tools.terminal_tool as _tt
    +        import tools.browser_tool as _bt
    +
    +        runner, adapter = make_restart_runner()
    +        adapter.disconnect = _make_async_noop()
    +
    +        monkeypatch.setattr(_pr.process_registry, "kill_all", lambda task_id=None: 0)
    +        monkeypatch.setattr(_tt, "cleanup_all_environments", lambda: None)
    +        monkeypatch.setattr(_bt, "cleanup_all_browsers", lambda: None)
    +
    +        with patch("gateway.status.remove_pid_file"), patch("gateway.status.write_runtime_status"), \
    +             patch("cron.scheduler.mark_job_run") as mock_mark:
    +            await runner.stop()
    +
    +        mock_mark.assert_not_called()

  • tests/gateway/test_shutdown_cache_cleanup.py

    @@ -54,6 +54,12 @@ def __init__(self):
         def _running_agent_count(self):
             return len(self._running_agents)
     
    +    def _active_cron_job_count(self):
    +        # stop() reads this alongside _running_agent_count when logging the
    +        # drain snapshot (#60432) -- this fake has no cron scheduler, so
    +        # there's never in-flight cron work to report.
    +        return 0
    +
         def _update_runtime_status(self, *_a, **_kw):
             pass
     

  • tests/gateway/test_update_cron_drain.py

    @@ -0,0 +1,86 @@
    +"""Regression tests for #60432.
    +
    +``/update`` (and other gateway shutdown paths) must drain in-flight cron jobs
    +before ``process_registry.kill_all()`` runs in final-cleanup.  Cron work runs on
    +a thread-pool worker and is tracked in ``cron.scheduler._running_job_ids``, not
    +in ``GatewayRunner._running_agents`` — so a zero-agent drain must still wait
    +for cron to finish (or time out and take the interrupt/kill path).
    +"""
    +import asyncio
    +from unittest.mock import AsyncMock, MagicMock, patch
    +
    +import pytest
    +
    +from tests.gateway.restart_test_helpers import make_restart_runner
    +
    +
    [redacted:email]
    +async def test_drain_active_agents_waits_for_in_flight_cron_jobs():
    +    runner, _adapter = make_restart_runner()
    +    runner._running_agents = {}
    +
    +    cron_count = [1]
    +
    +    def _cron_in_flight():
    +        return frozenset(f"job-{i}" for i in range(cron_count[0]))
    +
    +    async def finish_cron():
    +        await asyncio.sleep(0.15)
    +        cron_count[0] = 0
    +
    +    with patch("cron.scheduler.get_running_job_ids", side_effect=_cron_in_flight):
    +        task = asyncio.create_task(finish_cron())
    +        _snapshot, timed_out = await runner._drain_active_agents(1.0)
    +        await task
    +
    +    assert timed_out is False
    +    assert _snapshot == {}
    +
    +
    [redacted:email]
    +async def test_drain_active_agents_times_out_when_cron_still_running():
    +    runner, _adapter = make_restart_runner()
    +    runner._running_agents = {}
    +
    +    with patch("cron.scheduler.get_running_job_ids", return_value=frozenset({"job-1"})):
    +        _snapshot, timed_out = await runner._drain_active_agents(0.05)
    +
    +    assert timed_out is True
    +    assert _snapshot == {}
    +
    +
    [redacted:email]
    +async def test_gateway_stop_waits_for_cron_before_final_tool_kill():
    +    """Graceful cron completion must finish before final-cleanup kill_all."""
    +    runner, adapter = make_restart_runner()
    +    runner._restart_drain_timeout = 1.0
    +
    +    cron_count = [1]
    +    call_order: list[str] = []
    +
    +    def _cron_in_flight():
    +        return frozenset(f"job-{i}" for i in range(cron_count[0]))
    +
    +    def _fake_kill_all(task_id=None):
    +        call_order.append("kill_all")
    +        return 0
    +
    +    async def finish_cron():
    +        await asyncio.sleep(0.12)
    +        cron_count[0] = 0
    +
    +    with (
    +        patch("cron.scheduler.get_running_job_ids", side_effect=_cron_in_flight),
    +        patch("gateway.status.remove_pid_file"),
    +        patch("gateway.status.write_runtime_status"),
    +        patch("agent.auxiliary_client.shutdown_cached_clients"),
    +        patch("tools.process_registry.process_registry") as registry_mock,
    +    ):
    +        registry_mock.kill_all.side_effect = _fake_kill_all
    +        adapter.disconnect = AsyncMock()
    +
    +        cron_task = asyncio.create_task(finish_cron())
    +        await runner.stop()
    +        await cron_task
    +
    +    assert call_order == ["kill_all"]