Skip to content

feat(runner): queue-on-running so back-to-back handoffs don't race #24

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
volodchenkov merged 2 commits into from
Jun 18, 2026
Merged

feat(runner): queue-on-running so back-to-back handoffs don't race #24

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
dba7ac9
feat(runner): queue-on-running so back-to-back handoffs don't race
Jun 17, 2026
d091884
fix: pin httpx2 for starlette 1.x TestClient + tighten flush error ha…
Jun 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions pyproject.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Comment on lines +48 to +51

@coderabbitai coderabbitai Bot Jun 18, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Check if httpx2 exists on PyPI
echo "=== Checking httpx2 on PyPI ==="
curl -s https://pypi.org/pypi/httpx2/json | jq -r '.info.version // "PACKAGE NOT FOUND"'

echo -e "\n=== Checking httpx (the standard library) on PyPI ==="
curl -s https://pypi.org/pypi/httpx/json | jq -r '.info.version'

echo -e "\n=== Checking current Starlette dependency ==="
rg -n "starlette" pyproject.toml

Repository: volodchenkov/plane-conductor

Length of output: 301

🌐 Web query:

What HTTP client backend does Starlette TestClient use in version 1.0 and later?

💡 Result:

In Starlette version 1.0 and later, the TestClient uses the httpx library as its underlying HTTP client backend [1][2][3]. Prior to this version, Starlette's TestClient originally used the requests library. The transition from requests to httpx was implemented in Starlette 0.21.0 [4][5], ensuring that by the time Starlette reached its 1.0 stable release in March 2026, the TestClient was fully powered by httpx [4][5]. The TestClient provides an interface compatible with standard httpx sessions while adding specialized support for testing ASGI applications and WebSocket connections [2][3][6].

Citations:

Replace httpx2 with httpx—Starlette's TestClient backend is httpx, not httpx2.

Although httpx2 is a real package on PyPI, Starlette's TestClient uses the standard httpx library. The comment incorrectly states that Starlette 1.0 switched to httpx2; it actually uses httpx. Pinning the wrong package will not address the deprecation warning workaround as intended.

Diff 
-    # starlette 1.0 switched its TestClient backend to httpx2; the deprecation
+    # starlette 1.0 switched its TestClient backend to httpx; the deprecation
     # warning hits strict-mode pytest at collection. Pinning httpx here lets
     # TestClient pick it up without us touching production deps.
-    "httpx2>=0.1",
+    "httpx>=0.23.0",
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@pyproject.toml` around lines 48 - 51, The dependency pinned in the test
requirements is incorrect. Replace the package name from httpx2 to httpx in the
test dependencies section. The comment references Starlette 1.0's TestClient
backend switch, but Starlette actually uses the standard httpx library, not
httpx2. Update "httpx2>=0.1" to use the correct httpx package to properly
address the deprecation warning workaround as intended.

]
mcp = [
"mcp>=1.0",
Expand Down
229 changes: 214 additions & 15 deletions src/plane_conductor/runner.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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='<self>') → 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.
Expand All @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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.

Expand All @@ -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='<self>')`
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,
*,
Expand All @@ -129,6 +225,9 @@ async def spawn(
) -> Path:
"""Spawn `claude --agent <nickname>` 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).
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -303,6 +413,61 @@ 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),
)
Comment thread
coderabbitai[bot] marked this conversation as resolved.
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."""
with contextlib.suppress(ProcessLookupError):
Expand Down Expand Up @@ -423,14 +588,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))
Expand All @@ -440,7 +615,31 @@ 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. 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)
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


# ---------------------------------------------------------------------------
Expand Down
22 changes: 11 additions & 11 deletions src/plane_conductor/webhook.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,6 @@
AgentSpawnError,
CapacityFullError,
PlaneAPIError,
SessionAlreadyRunningError,
)
from plane_conductor.logging_config import get_logger
from plane_conductor.plane_client import PlaneClient
Expand Down Expand Up @@ -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"})
Expand Down
Loading