From dba7ac933a99c8b30fd31e05590a0cb3810ab25f Mon Sep 17 00:00:00 2001 From: Dmitry Volodchenkov Date: Wed, 17 Jun 2026 23:39:57 +0300 Subject: [PATCH 1/2] feat(runner): queue-on-running so back-to-back handoffs don't race MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Background ---------- An agent that calls `request_handoff(target_role='')` to schedule its own next step hits a race: Plane delivers the comment-webhook to conductor ~milliseconds before the previous subprocess has cleared `_active`. The old webhook path called `runner.spawn()` directly, which raised `SessionAlreadyRunningError` for the in-flight triple and recorded `skipped: already running` — the chain died there. This blocked the "step-execution discipline" we want to write into the coder / tester / SA prompts (one step per invocation + self-handoff to avoid 14-step token burn and hangs). A prompt rule that depends on a racy spawn would be a lie. Change ------ - New `Runner.request_spawn(...)`: spawn-or-queue. If the `(slug, nick, issue)` triple is already in flight, the request is stored in `_pending` keyed by the same triple (last-write-wins — multiple handoffs targeting the same triple coalesce, since the next agent reads state from Plane, not from queued trigger metadata). - `_supervise` flushes the queued spawn from its `finally` block, on a fresh task so it doesn't pin the dying supervisor's log fp / transport. - Direct `spawn()` keeps raising — webhook callers move to `request_spawn`, long-running callers (none today) can still opt in to the strict path. - Capacity cap stays a hard limit: queued spawns don't count toward the cap, and `request_spawn` re-raises `CapacityFullError` rather than queueing — same semantics as before for that case. - `wait_idle` now re-snapshots `_tasks` until the set drains, because flush tasks are added AFTER their parent `_supervise` is already done; a single `gather(*_tasks)` would miss them and leak the flushed log fp. - `_flush_pending` swallows errors with structured logs (the original webhook returned 200 to Plane long ago — there's no caller to surface to). Lost-race → `SessionAlreadyRunning` → info-log only. Tests ----- - `request_spawn` fires immediately on a free triple. - Second request for the same triple while active → `status: queued`, `pending_count == 1`. - Third request for the same triple → coalesces, still `pending_count == 1`. - Independent triples run in parallel (no false queueing across keys). - Queued spawn flushes after the active session for the same key exits. - Capacity full still raises (request_spawn does not bypass the cap). - Webhook stub updated to mock `request_spawn`. New webhook test covers the queued path returning `skipped: queued`. Old "skipped: already running" path retired (no caller for it). Out of scope ------------ Prompt patches for step-execution discipline (coders / testers / SA) ship in a follow-up PR on claude-sdlc-agents — they depend on this landing first. Co-Authored-By: Claude Opus 4.7 --- src/plane_conductor/runner.py | 197 ++++++++++++++++++++++++++++++--- src/plane_conductor/webhook.py | 22 ++-- tests/test_runner.py | 136 +++++++++++++++++++++++ tests/test_webhook.py | 25 +++-- 4 files changed, 344 insertions(+), 36 deletions(-) diff --git a/src/plane_conductor/runner.py b/src/plane_conductor/runner.py index d06e57d..053bd16 100644 --- a/src/plane_conductor/runner.py +++ b/src/plane_conductor/runner.py @@ -1,11 +1,18 @@ """Subprocess spawner — multi-workspace, defends against realistic failure modes. -- **Dedup**: the same `(workspace_slug, nickname, issue_uuid)` cannot be spawned - twice in this orchestrator. Plane delivers webhooks at-least-once and the - human can double-mention; without this, two agents race to create the same - artifact. The slug is included so two workspaces can share nicknames safely. +- **Dedup + queue-on-running**: the same `(workspace_slug, nickname, issue_uuid)` + cannot be spawned twice concurrently. Plane delivers webhooks at-least-once + and an agent can self-handoff via `request_handoff`; without dedup, two agents + race to create the same artifact. When a spawn for an in-flight triple is + requested via `request_spawn`, we **queue** it (last-write-wins) and fire it + the moment the active session exits — so step-execution chains + (`agent → request_handoff(target_role='') → next step`) don't lose the + re-entry to a race where the comment-webhook arrives before the previous + process has cleaned up. Direct `spawn()` callers still get the raise. - **Capacity cap**: `MAX_CONCURRENT_SESSIONS` is host-wide — it counts active agents across all workspaces. Stops a flood of mentions from melting the box. + Queued (pending) spawns do NOT count toward the cap; the cap is checked + again at flush time. - **Process group**: every subprocess is spawned in its own session (`start_new_session=True`). On timeout we `killpg(SIGTERM)` then SIGKILL the whole group, so descendants of `claude` (MCP servers, helper procs) die too. @@ -18,8 +25,11 @@ subprocess, then update that same comment when the agent exits. State kept in memory: `_tasks` (supervisor task pins), `_active` -(in-flight `(slug, nick, issue)` keys), `_procs` (process handles for shutdown). -All three are derivable from "what's running right now"; nothing is persisted. +(in-flight `(slug, nick, issue)` keys), `_procs` (process handles for shutdown), +`_pending` (queued spawns waiting for the corresponding `_active` key to clear). +All four are derivable from "what's running / queued right now"; nothing is +persisted — a conductor restart drops queued spawns the same way it drops +active ones (sentinels cover restart recovery for active sessions only). """ from __future__ import annotations @@ -30,6 +40,7 @@ import os import signal import time +from dataclasses import dataclass from datetime import UTC, datetime from pathlib import Path from typing import IO @@ -99,6 +110,22 @@ def _format_duration(seconds: float) -> str: return f"{h}h{m:02d}m" +@dataclass(frozen=True) +class _PendingSpawn: + """A spawn request deferred until the matching `_active` key clears. + + Holds the same arguments `spawn()` takes. The `PlaneClient` is the + long-lived per-workspace client owned by the router (it outlives any + individual webhook handler), so it's safe to capture by reference here. + """ + + workspace: WorkspaceConfig + plane: PlaneClient + nickname: str + issue_uuid: UUID + triggered_by_email: str | None + + class Runner: """Spawns claude subprocesses with dedup, capacity, and process-group control. @@ -113,11 +140,80 @@ def __init__(self, settings: Settings) -> None: self._active: set[tuple[str, str, UUID]] = set() # pid → Process. pid is also the pgid since we use start_new_session=True. self._procs: dict[int, asyncio.subprocess.Process] = {} + # Pending spawns keyed by (slug, nick, issue). Last-write-wins: + # multiple handoffs targeting the same triple while one is active + # coalesce — only one spawn fires at exit because the next agent + # reads actual state from Plane, not from queued trigger metadata. + self._pending: dict[tuple[str, str, UUID], _PendingSpawn] = {} @property def active_count(self) -> int: return len(self._active) + @property + def pending_count(self) -> int: + return len(self._pending) + + async def request_spawn( + self, + *, + workspace: WorkspaceConfig, + plane: PlaneClient, + nickname: str, + issue_uuid: UUID, + triggered_by_email: str | None = None, + ) -> dict[str, object]: + """Spawn-or-queue. Prefer this over `spawn()` for webhook-driven calls. + + If the `(slug, nick, issue)` triple is already in flight, the spawn is + **queued** and fires automatically when the active session exits. This + closes the race where an agent's `request_handoff(target_role='')` + comment-webhook arrives before the previous subprocess has finished + cleaning up — without it, the chain dies silently as a + `SessionAlreadyRunningError` log entry. + + Returns one of: + `{"status": "spawned", "log_path": Path}` — fresh spawn fired. + `{"status": "queued"}` — added to pending; fires on session exit. + If a pending entry for the triple already existed, it's + **overwritten** (last-write-wins; see class docstring). + + Raises: + CapacityFullError: host-wide cap reached at decision time. Pending + entries are NOT counted; the cap is rechecked at flush time + and queued spawns are dropped (with a log) if capacity is + still full then. + AgentSpawnError: claude binary missing (immediate spawn path only). + """ + slug = workspace.workspace_slug + key = (slug, nickname, issue_uuid) + if key in self._active: + already_queued = key in self._pending + self._pending[key] = _PendingSpawn( + workspace=workspace, + plane=plane, + nickname=nickname, + issue_uuid=issue_uuid, + triggered_by_email=triggered_by_email, + ) + log.info( + "spawn_queued", + workspace=slug, + nickname=nickname, + issue=str(issue_uuid), + replaced_existing=already_queued, + pending=len(self._pending), + ) + return {"status": "queued"} + log_path = await self.spawn( + workspace=workspace, + plane=plane, + nickname=nickname, + issue_uuid=issue_uuid, + triggered_by_email=triggered_by_email, + ) + return {"status": "spawned", "log_path": log_path} + async def spawn( self, *, @@ -129,6 +225,9 @@ async def spawn( ) -> Path: """Spawn `claude --agent ` for `(workspace, issue_uuid)`. + Direct-spawn path: raises rather than queues on conflict. Webhook + callers should prefer `request_spawn` to get queue-on-running. + Raises: SessionAlreadyRunningError: an agent for this triple is in flight. CapacityFullError: MAX_CONCURRENT_SESSIONS reached (host-wide). @@ -271,9 +370,20 @@ async def _supervise( if transport is not None: with contextlib.suppress(Exception): transport.close() - self._active.discard((workspace_slug, nickname, issue_uuid)) + key = (workspace_slug, nickname, issue_uuid) + self._active.discard(key) self._procs.pop(proc.pid, None) self._clear_sentinel(workspace_slug, nickname, issue_uuid) + pending = self._pending.pop(key, None) + if pending is not None: + # Schedule the queued spawn on a fresh task — don't await it + # in the supervise `finally` because that would keep this + # supervisor (and its log fp / transport refs) alive longer + # than necessary and could deadlock if the new spawn itself + # blocks on something this supervisor is holding. + flush_task = asyncio.create_task(self._flush_pending(pending)) + self._tasks.add(flush_task) + flush_task.add_done_callback(self._tasks.discard) duration = time.monotonic() - started_at log.info( @@ -303,6 +413,49 @@ async def _supervise( plane, project_id, nickname, issue_uuid, exit_code, timed_out ) + async def _flush_pending(self, pending: _PendingSpawn) -> None: + """Fire a queued spawn from `_supervise`'s exit hook. + + Errors are logged and swallowed — we never raise out of this path + because the original webhook that enqueued the pending spawn has + long since returned its 200 to Plane. There's no caller to surface + the error to. If capacity is now full, or the binary went missing, + we log it and the chain stops there; an initiator can re-trigger. + """ + try: + await self.spawn( + workspace=pending.workspace, + plane=pending.plane, + nickname=pending.nickname, + issue_uuid=pending.issue_uuid, + triggered_by_email=pending.triggered_by_email, + ) + except SessionAlreadyRunningError: + # Lost the race with a fresh direct trigger that landed between + # `_active.discard` and this flush. The direct trigger covers + # the same intent; safe to drop. + log.info( + "pending_flush_skipped_already_running", + workspace=pending.workspace.workspace_slug, + nickname=pending.nickname, + issue=str(pending.issue_uuid), + ) + except CapacityFullError: + log.warning( + "pending_flush_skipped_capacity_full", + workspace=pending.workspace.workspace_slug, + nickname=pending.nickname, + issue=str(pending.issue_uuid), + ) + except AgentSpawnError as exc: + log.error( + "pending_flush_failed", + workspace=pending.workspace.workspace_slug, + nickname=pending.nickname, + issue=str(pending.issue_uuid), + error=str(exc), + ) + async def _kill_group(self, proc: asyncio.subprocess.Process) -> int: """SIGTERM the process group; SIGKILL after 5s if it didn't exit.""" with contextlib.suppress(ProcessLookupError): @@ -423,14 +576,24 @@ def _clear_sentinel(self, workspace_slug: str, nickname: str, issue_uuid: UUID) # -- shutdown ------------------------------------------------------------- async def wait_idle(self, grace_seconds: float = 30.0) -> None: - """Wait for in-flight supervisors. After `grace_seconds`, kill all groups.""" - if not self._tasks: - return + """Wait for in-flight supervisors. After `grace_seconds`, kill all groups. + + Loops: a queued spawn flushed from an exiting supervisor adds a fresh + supervisor task, which `_flush_pending` schedules AFTER the parent + `_supervise` is already done — so a single `gather(*_tasks)` would + miss it. We re-snapshot until the set drains. + """ + deadline = asyncio.get_running_loop().time() + grace_seconds try: - await asyncio.wait_for( - asyncio.gather(*self._tasks, return_exceptions=True), - timeout=grace_seconds, - ) + while self._tasks: + remaining = deadline - asyncio.get_running_loop().time() + if remaining <= 0: + raise TimeoutError + snapshot = list(self._tasks) + await asyncio.wait_for( + asyncio.gather(*snapshot, return_exceptions=True), + timeout=remaining, + ) except TimeoutError: procs = list(self._procs.values()) log.warning("shutdown_grace_expired", killing_groups=len(procs)) @@ -440,7 +603,11 @@ async def wait_idle(self, grace_seconds: float = 30.0) -> None: for proc in procs: with contextlib.suppress(Exception): await proc.wait() - await asyncio.gather(*self._tasks, return_exceptions=True) + # Final drain — any flush tasks scheduled after we ran out of + # grace still get awaited so their file handles close. + while self._tasks: + snapshot = list(self._tasks) + await asyncio.gather(*snapshot, return_exceptions=True) # --------------------------------------------------------------------------- diff --git a/src/plane_conductor/webhook.py b/src/plane_conductor/webhook.py index 6c7e7af..5c57305 100644 --- a/src/plane_conductor/webhook.py +++ b/src/plane_conductor/webhook.py @@ -32,7 +32,6 @@ AgentSpawnError, CapacityFullError, PlaneAPIError, - SessionAlreadyRunningError, ) from plane_conductor.logging_config import get_logger from plane_conductor.plane_client import PlaneClient @@ -325,22 +324,23 @@ async def receive( # type: ignore[no-untyped-def] continue try: - await runner.spawn( + spawn_result = await runner.request_spawn( workspace=workspace, plane=plane, nickname=nickname, issue_uuid=issue_uuid, triggered_by_email=email, ) - spawned.append(nickname) - except SessionAlreadyRunningError: - log.info( - "duplicate_trigger", - workspace=slug, - nickname=nickname, - issue=str(issue_uuid), - ) - skipped.append({"nickname": nickname, "reason": "already running"}) + if spawn_result.get("status") == "queued": + log.info( + "spawn_deferred_until_active_clears", + workspace=slug, + nickname=nickname, + issue=str(issue_uuid), + ) + skipped.append({"nickname": nickname, "reason": "queued"}) + else: + spawned.append(nickname) except CapacityFullError: log.warning("capacity_full", workspace=slug, nickname=nickname) skipped.append({"nickname": nickname, "reason": "capacity full"}) diff --git a/tests/test_runner.py b/tests/test_runner.py index 84a239b..e5be75e 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -302,6 +302,142 @@ async def test_runner_capacity_cap( await runner.wait_idle(grace_seconds=0.1) +async def test_request_spawn_immediate_when_not_active( + settings: Settings, workspace_config: WorkspaceConfig, tmp_path: Path +) -> None: + """When the triple isn't running, request_spawn fires immediately.""" + s = settings.model_copy(update={"claude_binary": "/bin/true", "log_dir": tmp_path / "logs"}) + runner = Runner(settings=s) + + result = await runner.request_spawn( + workspace=workspace_config, plane=StubPlane(), nickname="rinzler", issue_uuid=ISSUE + ) # type: ignore[arg-type] + assert result["status"] == "spawned" + assert "log_path" in result + + await runner.wait_idle() + assert runner.active_count == 0 + assert runner.pending_count == 0 + + +async def test_request_spawn_queues_when_active( + settings: Settings, workspace_config: WorkspaceConfig, tmp_path: Path +) -> None: + """Second request for the same triple queues instead of raising.""" + fake_claude = tmp_path / "fake_claude" + fake_claude.write_text("#!/bin/sh\nsleep 5\n") + fake_claude.chmod(0o755) + s = settings.model_copy( + update={"claude_binary": str(fake_claude), "log_dir": tmp_path / "logs"} + ) + runner = Runner(settings=s) + + first = await runner.request_spawn( + workspace=workspace_config, plane=StubPlane(), nickname="rinzler", issue_uuid=ISSUE + ) # type: ignore[arg-type] + assert first["status"] == "spawned" + + second = await runner.request_spawn( + workspace=workspace_config, plane=StubPlane(), nickname="rinzler", issue_uuid=ISSUE + ) # type: ignore[arg-type] + assert second["status"] == "queued" + assert runner.pending_count == 1 + + third = await runner.request_spawn( + workspace=workspace_config, plane=StubPlane(), nickname="rinzler", issue_uuid=ISSUE + ) # type: ignore[arg-type] + assert third["status"] == "queued" + assert runner.pending_count == 1 # coalesced — same key, overwrites + + await runner.wait_idle(grace_seconds=0.1) + + +async def test_request_spawn_flushes_pending_on_active_exit( + settings: Settings, workspace_config: WorkspaceConfig, tmp_path: Path +) -> None: + """A queued spawn fires after the active session for the same key exits.""" + # First session uses a short-lived /bin/true so it exits ~immediately. + s = settings.model_copy(update={"claude_binary": "/bin/true", "log_dir": tmp_path / "logs"}) + runner = Runner(settings=s) + + plane = StubPlane() + first = await runner.request_spawn( + workspace=workspace_config, plane=plane, nickname="rinzler", issue_uuid=ISSUE + ) # type: ignore[arg-type] + assert first["status"] == "spawned" + + # Queue a second spawn for the same triple before the first finishes. + second = await runner.request_spawn( + workspace=workspace_config, plane=plane, nickname="rinzler", issue_uuid=ISSUE + ) # type: ignore[arg-type] + assert second["status"] == "queued" + + # First session exits → pending must flush → second session runs → both + # eventually drain. wait_idle waits for ALL active + flushed sessions. + await runner.wait_idle(grace_seconds=0.2) + assert runner.active_count == 0 + assert runner.pending_count == 0 + + +async def test_request_spawn_capacity_full_still_raises( + settings: Settings, workspace_config: WorkspaceConfig, tmp_path: Path +) -> None: + """Capacity is a hard cap — request_spawn raises rather than queueing.""" + from plane_conductor.exceptions import CapacityFullError + + fake_claude = tmp_path / "fake_claude" + fake_claude.write_text("#!/bin/sh\nsleep 5\n") + fake_claude.chmod(0o755) + s = settings.model_copy( + update={ + "claude_binary": str(fake_claude), + "max_concurrent_sessions": 1, + "log_dir": tmp_path / "logs", + } + ) + runner = Runner(settings=s) + + await runner.request_spawn( + workspace=workspace_config, plane=StubPlane(), nickname="a", issue_uuid=UUID(int=1) + ) # type: ignore[arg-type] + with pytest.raises(CapacityFullError): + await runner.request_spawn( + workspace=workspace_config, plane=StubPlane(), nickname="b", issue_uuid=UUID(int=2) + ) # type: ignore[arg-type] + + await runner.wait_idle(grace_seconds=0.1) + + +async def test_request_spawn_independent_triples_run_in_parallel( + settings: Settings, workspace_config: WorkspaceConfig, tmp_path: Path +) -> None: + """request_spawn for a different (nick, issue) doesn't queue against an unrelated active.""" + fake_claude = tmp_path / "fake_claude" + fake_claude.write_text("#!/bin/sh\nsleep 5\n") + fake_claude.chmod(0o755) + s = settings.model_copy( + update={ + "claude_binary": str(fake_claude), + "max_concurrent_sessions": 4, + "log_dir": tmp_path / "logs", + } + ) + runner = Runner(settings=s) + plane = StubPlane() + + a = await runner.request_spawn( + workspace=workspace_config, plane=plane, nickname="a", issue_uuid=UUID(int=1) + ) # type: ignore[arg-type] + b = await runner.request_spawn( + workspace=workspace_config, plane=plane, nickname="b", issue_uuid=UUID(int=2) + ) # type: ignore[arg-type] + assert a["status"] == b["status"] == "spawned" + assert runner.active_count == 2 + assert runner.pending_count == 0 + + await runner.wait_idle(grace_seconds=0.1) + + async def test_active_set_clears_after_exit( settings: Settings, workspace_config: WorkspaceConfig, tmp_path: Path ) -> None: diff --git a/tests/test_webhook.py b/tests/test_webhook.py index 3c9cf23..811bac9 100644 --- a/tests/test_webhook.py +++ b/tests/test_webhook.py @@ -47,8 +47,9 @@ class StubRunner: def __init__(self) -> None: self.calls: list[dict[str, Any]] = [] - async def spawn(self, **kwargs: Any) -> None: + async def request_spawn(self, **kwargs: Any) -> dict[str, Any]: self.calls.append(kwargs) + return {"status": "spawned"} def _sign(secret: str, body: bytes) -> str: @@ -453,7 +454,7 @@ def test_webhook_handles_spawn_failure( from plane_conductor.exceptions import AgentSpawnError class CrashingRunner: - async def spawn(self, **kwargs: Any) -> None: + async def request_spawn(self, **kwargs: Any) -> dict[str, Any]: raise AgentSpawnError("boom") plane = StubPlane({SARK: {"email": "sark@example.io"}}) @@ -492,25 +493,29 @@ def _comment_body(mention_uuid: str) -> bytes: ).encode() -def test_webhook_skipped_on_dedup(settings: Settings, workspace_config: WorkspaceConfig) -> None: - from plane_conductor.exceptions import SessionAlreadyRunningError +def test_webhook_queues_when_runner_reports_already_running( + settings: Settings, workspace_config: WorkspaceConfig +) -> None: + """When request_spawn reports queued (because the triple is already in + flight), webhook records it as `skipped: queued` — the runner will fire + the spawn itself when the active session exits.""" - class DupRunner: - async def spawn(self, **kwargs: Any) -> None: - raise SessionAlreadyRunningError("already running") + class QueueingRunner: + async def request_spawn(self, **kwargs: Any) -> dict[str, Any]: + return {"status": "queued"} plane = StubPlane({SARK: {"email": "sark@example.io"}}) - client = TestClient(_app(settings, workspace_config, plane, DupRunner())) + client = TestClient(_app(settings, workspace_config, plane, QueueingRunner())) resp = _send(client, settings, workspace_config, _comment_body(SARK)) assert resp.status_code == 200 - assert resp.json()["skipped"] == [{"nickname": "sark", "reason": "already running"}] + assert resp.json()["skipped"] == [{"nickname": "sark", "reason": "queued"}] def test_webhook_skipped_on_capacity(settings: Settings, workspace_config: WorkspaceConfig) -> None: from plane_conductor.exceptions import CapacityFullError class FullRunner: - async def spawn(self, **kwargs: Any) -> None: + async def request_spawn(self, **kwargs: Any) -> dict[str, Any]: raise CapacityFullError("full") plane = StubPlane({SARK: {"email": "sark@example.io"}}) From d09188498de6117799a47547e7f0fc70ecb60f6c Mon Sep 17 00:00:00 2001 From: Dmitry Volodchenkov Date: Thu, 18 Jun 2026 16:41:54 +0300 Subject: [PATCH 2/2] fix: pin httpx2 for starlette 1.x TestClient + tighten flush error handling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI on Python 3.11 broke on collection because starlette 1.x's TestClient imports `httpx2` and emits StarletteDeprecationWarning when falling back to httpx — strict-mode pytest (`filterwarnings = ["error"]`) turns the warning into a collection error. Locally the .venv still had starlette 1.0.0 cached; reproduced by upgrading to 1.3.1 and confirmed the same failure. Adding `httpx2` to dev-deps lets TestClient pick its preferred backend and the warning never fires. Once fastapi pulls in httpx2 by default this entry can drop out — leaving a comment. Also addresses two CodeRabbit findings from PR #24 review: - `_flush_pending`: add a catch-all `Exception` arm. The flush task runs detached from any webhook caller, so an uncaught OSError / PermissionError / unexpected subprocess failure would only surface in asyncio's unraisable hook. Log it structurally instead and keep the contract that the flush never escapes. - `wait_idle` post-kill drain: bound the final loop with a 5s deadline. A flushed spawn that itself blocks (broken plane API, hung child) could otherwise pin shutdown indefinitely. Timeout → warning + break. Skipped: CodeRabbit asked to re-add a `SessionAlreadyRunningError` catch in the webhook and a matching test. `request_spawn` contractually does NOT raise that error (it queues), so the catch would be dead code and the test would assert a path that can't be reached. Left a thread reply explaining. Tests: 224 passed, 7 skipped (e2e). ruff + ruff-format + mypy clean. Co-Authored-By: Claude Opus 4.7 --- pyproject.toml | 4 ++++ src/plane_conductor/runner.py | 36 +++++++++++++++++++++++++++++++++-- 2 files changed, 38 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 8ec90e0..b10fbb1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -45,6 +45,10 @@ dev = [ "mypy>=1.10", "types-PyYAML", "mcp>=1.0", + # starlette 1.0 switched its TestClient backend to httpx2; the deprecation + # warning hits strict-mode pytest at collection. Pinning httpx2 here lets + # TestClient pick it up without us touching production deps. + "httpx2>=0.1", ] mcp = [ "mcp>=1.0", diff --git a/src/plane_conductor/runner.py b/src/plane_conductor/runner.py index 053bd16..ff29409 100644 --- a/src/plane_conductor/runner.py +++ b/src/plane_conductor/runner.py @@ -455,6 +455,18 @@ async def _flush_pending(self, pending: _PendingSpawn) -> None: issue=str(pending.issue_uuid), error=str(exc), ) + except Exception as exc: + # Catch-all: this runs detached from any webhook caller, so an + # uncaught OSError / PermissionError / unexpected subprocess + # failure would just go into asyncio's unraisable handler with + # no operator-visible signal. Log it structurally instead. + log.exception( + "pending_flush_unexpected_error", + workspace=pending.workspace.workspace_slug, + nickname=pending.nickname, + issue=str(pending.issue_uuid), + error=str(exc), + ) async def _kill_group(self, proc: asyncio.subprocess.Process) -> int: """SIGTERM the process group; SIGKILL after 5s if it didn't exit.""" @@ -604,10 +616,30 @@ async def wait_idle(self, grace_seconds: float = 30.0) -> None: with contextlib.suppress(Exception): await proc.wait() # Final drain — any flush tasks scheduled after we ran out of - # grace still get awaited so their file handles close. + # grace still get awaited so their file handles close. Bounded + # so a misbehaving flush (queued spawn that itself blocks on a + # broken plane API, etc.) can't park shutdown indefinitely. + post_kill_deadline = asyncio.get_running_loop().time() + 5.0 while self._tasks: + remaining = post_kill_deadline - asyncio.get_running_loop().time() + if remaining <= 0: + log.warning( + "wait_idle_post_kill_drain_timeout", + leftover_tasks=len(self._tasks), + ) + break snapshot = list(self._tasks) - await asyncio.gather(*snapshot, return_exceptions=True) + try: + await asyncio.wait_for( + asyncio.gather(*snapshot, return_exceptions=True), + timeout=remaining, + ) + except TimeoutError: + log.warning( + "wait_idle_post_kill_drain_timeout", + leftover_tasks=len(self._tasks), + ) + break # ---------------------------------------------------------------------------