diff --git a/docs/memory/feature-flows.md b/docs/memory/feature-flows.md index a3f8f6caa..020c1bda7 100644 --- a/docs/memory/feature-flows.md +++ b/docs/memory/feature-flows.md @@ -11,6 +11,7 @@ | Date | ID | Feature | Flow | |------|-----|---------|------| +| 2026-04-26 | #498 | Sync `/task` long-poll on backlog — sync parallel calls at capacity now spill to BACKLOG-001 (same backlog as async) and long-poll the open HTTP connection until terminal status (cap `2 × effective_timeout`); new `services/sync_waiter.py` owns the in-process registry + event/poll-fallback wait helper | [persistent-task-backlog.md](feature-flows/persistent-task-backlog.md), [parallel-headless-execution.md](feature-flows/parallel-headless-execution.md) | | 2026-04-24 | WEBHOOK-001 (#291) | Webhook triggers — token-authenticated public URL fires schedule executions | [webhook-triggers.md](feature-flows/webhook-triggers.md) | | 2026-04-25 | #487 | Telegram file upload Phase 2 — workspace delivery hardening: NFKC sanitizer with collision dedup, spec injection format `[File uploaded by {uploader}]: {name} ({size}) saved to {path}`, all-writes-failed channel error + abort. Same code path benefits Slack inbound. | [telegram-integration.md](feature-flows/telegram-integration.md), [slack-file-sharing.md](feature-flows/slack-file-sharing.md) | | 2026-04-23 | #476 | SQLite lexicographic cutoff bug fix — new `iso_cutoff(hours)` helper replaces `datetime('now', ...)` in 15 sites across rate-limit / dashboard / schedules; `max_retries` default flipped `1 → 0`; `cleanup_old_rate_limit_events` wired into `CleanupService` (phase 6, hourly) | [subscription-auto-switch.md](feature-flows/subscription-auto-switch.md), [cleanup-service.md](feature-flows/cleanup-service.md), [scheduler-service.md](feature-flows/scheduler-service.md) | diff --git a/docs/memory/feature-flows/parallel-headless-execution.md b/docs/memory/feature-flows/parallel-headless-execution.md index a9927ee11..4f5edb9bb 100644 --- a/docs/memory/feature-flows/parallel-headless-execution.md +++ b/docs/memory/feature-flows/parallel-headless-execution.md @@ -3,13 +3,14 @@ > **Requirement**: 12.1 - Parallel Headless Execution > **Status**: Implemented > **Created**: 2025-12-22 -> **Updated**: 2026-04-20 (Issue #418 inter-agent timeout fix) +> **Updated**: 2026-04-26 (Issue #498 sync long-poll on backlog) > **Verified**: 2026-02-05 ## Revision History | Date | Changes | |------|---------| +| 2026-04-26 | **Issue #498 - Sync long-poll on backlog**: Sync `/task` calls (`async_mode=false`) at capacity used to fail terminally with HTTP 429. They now spill to the same persistent backlog (BACKLOG-001) the async path uses and long-poll on the open HTTP connection until the queued execution reaches a terminal status. Total connection hold capped at `2 × effective_timeout` (queue wait + execution). Router (`chat.py:1007-1144`) pre-acquires the slot mirroring the async path, then on at-capacity calls `backlog.enqueue()` followed by `wait_for_sync_terminal()`. New module `services/sync_waiter.py` owns the in-process registry, `signal_sync_waiter()`, and `wait_for_sync_terminal()` (event + 5s DB-poll fallback). The drain reuses `_run_async_task_with_persistence` unchanged — it now calls `signal_sync_waiter` from its `finally` block to wake any sync waiter. See [persistent-task-backlog.md](persistent-task-backlog.md) for the full backlog flow. | | 2026-04-20 | **Issue #418 - Inter-agent timeout ceiling fix**: Removed hardcoded 600s timeout assumption from the MCP parallel/task path so per-agent `execution_timeout_seconds` (TIMEOUT-001, default 900s, max 7200s) is honored end-to-end. `src/mcp-server/src/tools/chat.ts` — `chat_with_agent` Zod schema no longer applies `.default(600)` to `timeout_seconds`; when callers omit it, `undefined` now flows through to the backend, which resolves the target agent's configured timeout. `src/mcp-server/src/client.ts:563-565` — `client.task()` HTTP fetch ceiling changed from `(timeout_seconds \|\| 600) + 10` to `(timeout_seconds ?? 7200) + 60`, so the fetch client doesn't abort before a long-running agent-configured task finishes. Async mode still uses a fixed 30s HTTP ceiling (unchanged). | | 2026-04-17 | **Issue #361 - Max-turns error fix**: Fixed max_turns termination being misclassified as authentication failure. Added detection for `terminal_reason="max_turns"` and `subtype="error_max_turns"` in result messages (`claude_code.py:329-336`). Max-turns errors now return HTTP 422 with clear "Task exceeded turn limit" message instead of HTTP 503 "Authentication failure". Also raised `max_turns_task` default from 20 to 50 in both `claude_code.py:52` and `guardrails-baseline.json:65`. | | 2026-03-26 | **Line number refresh**: Updated all file/line references to match current codebase after upstream shifts (~92 lines in backend `chat.py`, model extraction in `models.py`, agent server reorganisation). | @@ -221,13 +222,14 @@ As of EXEC-024, the sync and async execution paths diverge: | Aspect | Sync (`async_mode=false`) | Async (`async_mode=true`) | |--------|---------------------------|---------------------------| -| Execution logic | `TaskExecutionService.execute_task()` in `services/task_execution_service.py` | `_run_async_task_with_persistence()` inline in `routers/chat.py:438-650` | -| Slot management | Service acquires/releases slots internally | Router acquires slot before spawning; background task releases in `finally` | +| Execution logic | `TaskExecutionService.execute_task()` in `services/task_execution_service.py` | `_run_async_task_with_persistence()` inline in `routers/chat.py` | +| Slot management | Router pre-acquires (issue #498); service releases in `finally` (`slot_already_held=True`) | Router pre-acquires; background task releases in `finally` | +| At-capacity behavior | Spills to backlog (BACKLOG-001), long-polls on the open HTTP connection until terminal status (issue #498). Total hold ≤ `2 × effective_timeout`. | Spills to backlog, returns HTTP 202 with `execution_id`, caller polls. | | Activity tracking | Service tracks start/completion internally | Router tracks start; background task completes activities | | Result handling | Returns `TaskExecutionResult`; router translates to HTTP | Background task updates DB directly | | HTTP helper | `agent_post_with_retry()` defined in service, called internally | Same function imported from service into `chat.py` | -The router (`chat.py:652-917`) still handles: container validation, execution record creation (early), collaboration tracking (WebSocket events), async mode branching, session persistence (`save_to_session`), and translating `TaskExecutionResult.status == "failed"` to HTTP error codes (429/504/503). +The router (`chat.py`) still handles: container validation, execution record creation (early), collaboration tracking (WebSocket events), async mode branching, session persistence (`save_to_session`), and translating `TaskExecutionResult.status == "failed"` to HTTP error codes (429/504/503). For sync at-capacity, the router additionally calls `backlog.enqueue()` and `wait_for_sync_terminal()` (services/sync_waiter.py); on wake it either returns the inline result (drain happy path) or reconstructs a minimal `TaskExecutionResult` from the DB row (poll-fallback for non-drain terminal flips). ## API Specifications diff --git a/docs/memory/feature-flows/persistent-task-backlog.md b/docs/memory/feature-flows/persistent-task-backlog.md index 4be2ede05..8c8b84aa0 100644 --- a/docs/memory/feature-flows/persistent-task-backlog.md +++ b/docs/memory/feature-flows/persistent-task-backlog.md @@ -1,20 +1,28 @@ # Feature Flow: Persistent Task Backlog -> **Requirement**: BACKLOG-001 — Persistent async task backlog for over-capacity requests +> **Requirement**: BACKLOG-001 — Persistent task backlog for over-capacity requests > **Status**: Implemented > **Created**: 2026-04-13 -> **GitHub Issue**: [#260](https://github.com/abilityai/trinity/issues/260) +> **GitHub Issue**: [#260](https://github.com/abilityai/trinity/issues/260), extended by [#498](https://github.com/abilityai/trinity/issues/498) (sync long-poll) > **Priority**: P1 > **Related**: [parallel-capacity.md](parallel-capacity.md), [task-execution-service.md](task-execution-service.md), [parallel-headless-execution.md](parallel-headless-execution.md) ## Overview -When `async_mode=true` arrives at `POST /api/agents/{name}/task` and all of the -agent's parallel execution slots (CAPACITY-001) are occupied, the request is -spilled into a durable SQLite-backed FIFO backlog instead of returning HTTP -429. When a slot frees, the oldest queued item for that agent is drained -automatically via a `SlotService` release callback. True HTTP 429 is only -returned when the backlog itself is also at its configured depth. +When a `POST /api/agents/{name}/task` request arrives and all of the agent's +parallel execution slots (CAPACITY-001) are occupied, the request is spilled +into a durable SQLite-backed FIFO backlog instead of returning HTTP 429. When +a slot frees, the oldest queued item for that agent is drained automatically +via a `SlotService` release callback. True HTTP 429 is only returned when the +backlog itself is also at its configured depth. + +Both modes share the same backlog (issue #498): +- **Async (`async_mode=true`)**: Caller gets HTTP 202 with `execution_id` + immediately and polls for the result. The backlog drains in the background. +- **Sync (`async_mode=false`)**: Caller's HTTP connection is held open and + long-polls until the queued execution reaches a terminal status, then the + result is returned inline on the same connection. Total connection hold is + bounded by `2 × effective_timeout` (queue wait + execution). Queued rows survive backend restarts. A 60-second maintenance task in the backend process expires rows older than 24 hours and drains orphans left @@ -22,20 +30,29 @@ behind when a release callback couldn't fire (e.g. process crash). ## Problem Statement -Before this change, `async_mode=true` requests at capacity were dropped on -the floor with a 429 response. Bursty MCP fan-out scenarios (agents +Before BACKLOG-001 (#260), `async_mode=true` requests at capacity were dropped +on the floor with a 429 response. Bursty MCP fan-out scenarios (agents orchestrating other agents via `chat_with_agent(async=true)`) routinely hit the 3-slot default cap and lost work. Clients had to implement their own retry-with-backoff logic, and there was no first-class backpressure signal. +Before #498, sync calls (`async_mode=false`) bypassed the backlog entirely — +hitting capacity returned a terminal 429 even though the backlog could have +absorbed the overflow. Observed in production: ~40% terminal-failure rate +from one MCP fan-out caller (214 capacity rejections / 24h, 0 enqueues from +the same caller across 541 dispatches). #498 closed that gap by spilling +sync calls to the same backlog and long-polling on the open HTTP connection. + The backlog gives Trinity: -- **Spill-over by default** for async mode — no lost requests below the - backlog depth cap +- **Spill-over by default** for both sync and async — no lost requests below + the backlog depth cap - **Restart durability** — queued items survive backend restarts via SQLite - **Bounded resource envelope** — per-agent `max_backlog_depth` (default 50, hard cap 200) + 24h stale expiry - **Transparent to pollers** — existing `GET /api/agents/{name}/executions/{id}` returns `status=queued` while the row waits to drain +- **Transparent to sync callers** — same response shape as immediate-slot path, + just with extra wait time ## Architecture Diagram @@ -114,6 +131,46 @@ Parallel path (safety net): ▼ ``` +### Sync long-poll path (issue #498) + +``` + POST /api/agents/{name}/task + async_mode=false + │ + router pre-acquires slot + │ + ┌──────────────────┴───────────────────┐ + │ │ + slot acquired slot full + │ │ + ▼ ▼ + execute_task(slot_already_held=True) backlog.enqueue() + → return inline result │ + ┌───────────┴───────────┐ + │ │ + depth < cap depth >= cap + │ │ + ▼ ▼ + wait_for_sync_terminal HTTP 429 + (event + 5s DB-poll fallback) + │ + ┌─────────────┼─────────────┐ + │ │ │ + signaled by poll detects timeout + drain finally terminal flip (2 × effective_timeout) + │ │ │ + ▼ ▼ ▼ + return inline result reconstruct HTTP 504 + (full TaskExecResult) from DB row (execution may + → return still complete + in background) +``` + +The drain machinery is shared with the async path — `_run_async_task_with_persistence` +runs the queued task identically, then signals `_sync_waiters` from its `finally` +block. Sync waiters wake on the same event the async chat-session-persistence +broadcast fires on. + ## Database Schema Migration `backlog_support` (append-only, reuses existing table): @@ -296,6 +353,31 @@ After stopping the container and deleting schedules, the delete path calls `backlog.cancel_all_backlog(agent_name, reason="agent_deleted")` so orphan queued rows don't linger in the database. +### Sync Waiter — `src/backend/services/sync_waiter.py` (NEW, #498) + +In-process registry that lets sync HTTP callers long-poll a queued execution +on the same connection. Two primitives: + +- `signal_sync_waiter(execution_id, result, chat_session_id)` — called from + `_run_async_task_with_persistence` finally block. Looks up the registered + future and completes it with the rich `TaskExecutionResult`. No-op when no + waiter is registered (the normal async fire-and-forget path) or when the + caller already cancelled. +- `wait_for_sync_terminal(execution_id, timeout)` — registers a future, + starts a 5s DB-poll fallback task, then `asyncio.wait(FIRST_COMPLETED)`s + on either signal. Returns the rich payload on signal, returns `None` on + poll-fallback hit (caller reconstructs from DB row), raises `TimeoutError` + if neither fires. + +The registry is in-process — multi-worker deployments would need pubsub to +fan signals across processes; that's not the current backend shape (single +worker). + +The poll fallback covers terminal-flip sites that don't go through the drain: +corrupt-metadata in `_spawn_drain`, `expire_stale_queued`, `cancel_all_backlog`, +and `cleanup_service` recovery. Latency cost is bounded at one poll interval +(default 5s). + ## Configuration Per-agent backlog depth is stored in `agent_ownership.max_backlog_depth`: diff --git a/src/backend/routers/chat.py b/src/backend/routers/chat.py index f3f89df4f..5f7a30fc6 100644 --- a/src/backend/routers/chat.py +++ b/src/backend/routers/chat.py @@ -42,6 +42,12 @@ _websocket_manager = None +# Sync HTTP long-poll primitives live in services/sync_waiter.py so they're +# importable from tests without pulling in the full router/auth chain. +# (Issue #498) +from services.sync_waiter import signal_sync_waiter, wait_for_sync_terminal + + def set_websocket_manager(manager): """Set WebSocket manager for broadcasting collaboration events.""" global _websocket_manager @@ -591,157 +597,165 @@ async def _run_async_task_with_persistence( task_service = get_task_execution_service() triggered_by = "agent" if x_source_agent else "manual" - # Service tracks CHAT_START with parent_activity_id=collaboration_activity_id - # and merges extra_activity_details (parallel_mode/async_mode) so the Network - # view filter at src/frontend/src/stores/network.js:255 still includes this - # execution. - result = await task_service.execute_task( - agent_name=agent_name, - message=request.message, - triggered_by=triggered_by, - source_user_id=user_id, - source_user_email=user_email, - source_agent_name=x_source_agent, - model=request.model, - timeout_seconds=request.timeout_seconds, - resume_session_id=request.resume_session_id, - allowed_tools=request.allowed_tools, - system_prompt=request.system_prompt, - execution_id=execution_id, - subscription_id=subscription_id, - parent_activity_id=collaboration_activity_id, - extra_activity_details={ - "parallel_mode": True, - "async_mode": True, - "model": request.model, - "timeout_seconds": request.timeout_seconds, - }, - slot_already_held=True, # Router pre-acquired to preserve 429-upfront contract - ) - - execution_time_ms = int((datetime.utcnow() - start_time).total_seconds() * 1000) - - # ---- Post-task: chat session persistence (THINK-001) ---- + # Outer try/finally so a sync long-poll waiter (issue #498) is always + # signaled even if the post-task side effects below raise. + result = None chat_session_id = None - if request.save_to_session and user_id and user_email: - chat_session_id = await _persist_chat_session( + try: + # Service tracks CHAT_START with parent_activity_id=collaboration_activity_id + # and merges extra_activity_details (parallel_mode/async_mode) so the Network + # view filter at src/frontend/src/stores/network.js:255 still includes this + # execution. + result = await task_service.execute_task( agent_name=agent_name, - request=request, - result=result, - user_id=user_id, - user_email=user_email, + message=request.message, + triggered_by=triggered_by, + source_user_id=user_id, + source_user_email=user_email, + source_agent_name=x_source_agent, + model=request.model, + timeout_seconds=request.timeout_seconds, + resume_session_id=request.resume_session_id, + allowed_tools=request.allowed_tools, + system_prompt=request.system_prompt, + execution_id=execution_id, subscription_id=subscription_id, - execution_time_ms=execution_time_ms, + parent_activity_id=collaboration_activity_id, + extra_activity_details={ + "parallel_mode": True, + "async_mode": True, + "model": request.model, + "timeout_seconds": request.timeout_seconds, + }, + slot_already_held=True, # Router pre-acquired to preserve 429-upfront contract ) - if chat_session_id and _websocket_manager: - try: - await _websocket_manager.broadcast(json.dumps({ - "type": "chat_response_ready", - "execution_id": execution_id, - "agent_name": agent_name, - "chat_session_id": chat_session_id, - "timestamp": utc_now_iso(), - })) - except Exception as e: - logger.warning(f"[Task Async] chat_response_ready broadcast failed: {e}") - # ---- Post-task: complete collaboration activity ---- - if collaboration_activity_id: - try: - await activity_service.complete_activity( - activity_id=collaboration_activity_id, - status=( - ActivityState.COMPLETED - if result.status == TaskExecutionStatus.SUCCESS - else ActivityState.FAILED - ), - details={ - "response_length": len(result.response or ""), - "execution_time_ms": execution_time_ms, - "execution_id": execution_id, - }, - error=(result.error if result.status == TaskExecutionStatus.FAILED else None), - ) - except Exception as e: - logger.warning(f"[Task Async] collaboration activity completion failed: {e}") - - # ---- Post-task: complete self-task activity and inject result (SELF-EXEC-001) ---- - if is_self_task and self_task_activity_id: - activity_status = ( - ActivityState.COMPLETED - if result.status == TaskExecutionStatus.SUCCESS - else ActivityState.FAILED - ) + execution_time_ms = int((datetime.utcnow() - start_time).total_seconds() * 1000) - # Complete the self-task activity - try: - await activity_service.complete_activity( - activity_id=self_task_activity_id, - status=activity_status, - details={ - "response_length": len(result.response or ""), - "execution_time_ms": execution_time_ms, - "execution_id": execution_id, - "inject_result": request.inject_result, - }, - error=(result.error if result.status == TaskExecutionStatus.FAILED else None), + # ---- Post-task: chat session persistence (THINK-001) ---- + if request.save_to_session and user_id and user_email: + chat_session_id = await _persist_chat_session( + agent_name=agent_name, + request=request, + result=result, + user_id=user_id, + user_email=user_email, + subscription_id=subscription_id, + execution_time_ms=execution_time_ms, ) - except Exception as e: - logger.warning(f"[Task Async] self-task activity completion failed: {e}") - - # Inject result into chat session if requested - if request.inject_result and request.chat_session_id and result.status == TaskExecutionStatus.SUCCESS: + if chat_session_id and _websocket_manager: + try: + await _websocket_manager.broadcast(json.dumps({ + "type": "chat_response_ready", + "execution_id": execution_id, + "agent_name": agent_name, + "chat_session_id": chat_session_id, + "timestamp": utc_now_iso(), + })) + except Exception as e: + logger.warning(f"[Task Async] chat_response_ready broadcast failed: {e}") + + # ---- Post-task: complete collaboration activity ---- + if collaboration_activity_id: try: - # Validate session exists and belongs to user - session = db.get_chat_session(request.chat_session_id) - if session and session.get("user_id") == user_id: - # Add self-task result as a chat message - db.add_chat_message( - session_id=request.chat_session_id, - agent_name=agent_name, - user_id=user_id, - user_email=user_email or "", - role="assistant", - content=result.response or "", - cost=result.cost, - context_used=result.context_used, - context_max=result.context_max, - execution_time_ms=execution_time_ms, - source="self_task", # Mark as self-task result - ) - logger.info(f"[Self-Task] Injected result into chat session {request.chat_session_id}") - else: - logger.warning(f"[Self-Task] Cannot inject result: session {request.chat_session_id} not found or not owned by user") + await activity_service.complete_activity( + activity_id=collaboration_activity_id, + status=( + ActivityState.COMPLETED + if result.status == TaskExecutionStatus.SUCCESS + else ActivityState.FAILED + ), + details={ + "response_length": len(result.response or ""), + "execution_time_ms": execution_time_ms, + "execution_id": execution_id, + }, + error=(result.error if result.status == TaskExecutionStatus.FAILED else None), + ) except Exception as e: - logger.warning(f"[Self-Task] Failed to inject result into chat session: {e}") + logger.warning(f"[Task Async] collaboration activity completion failed: {e}") + + # ---- Post-task: complete self-task activity and inject result (SELF-EXEC-001) ---- + if is_self_task and self_task_activity_id: + activity_status = ( + ActivityState.COMPLETED + if result.status == TaskExecutionStatus.SUCCESS + else ActivityState.FAILED + ) - # Broadcast self-task completion event - if _websocket_manager: + # Complete the self-task activity try: - await _websocket_manager.broadcast(json.dumps({ - "type": "agent_activity", - "agent_name": agent_name, - "activity_type": "self_task", - "activity_state": "completed" if result.status == TaskExecutionStatus.SUCCESS else "failed", - "action": f"Background task completed", - "timestamp": utc_now_iso(), - "details": { - "execution_id": execution_id, - "chat_session_id": request.chat_session_id, - "cost_usd": result.cost, + await activity_service.complete_activity( + activity_id=self_task_activity_id, + status=activity_status, + details={ + "response_length": len(result.response or ""), "execution_time_ms": execution_time_ms, - "response_preview": (result.response or "")[:200], + "execution_id": execution_id, "inject_result": request.inject_result, - "result_injected": request.inject_result and request.chat_session_id is not None, - } - })) + }, + error=(result.error if result.status == TaskExecutionStatus.FAILED else None), + ) except Exception as e: - logger.warning(f"[Self-Task] WebSocket broadcast failed: {e}") + logger.warning(f"[Task Async] self-task activity completion failed: {e}") + + # Inject result into chat session if requested + if request.inject_result and request.chat_session_id and result.status == TaskExecutionStatus.SUCCESS: + try: + # Validate session exists and belongs to user + session = db.get_chat_session(request.chat_session_id) + if session and session.get("user_id") == user_id: + # Add self-task result as a chat message + db.add_chat_message( + session_id=request.chat_session_id, + agent_name=agent_name, + user_id=user_id, + user_email=user_email or "", + role="assistant", + content=result.response or "", + cost=result.cost, + context_used=result.context_used, + context_max=result.context_max, + execution_time_ms=execution_time_ms, + source="self_task", # Mark as self-task result + ) + logger.info(f"[Self-Task] Injected result into chat session {request.chat_session_id}") + else: + logger.warning(f"[Self-Task] Cannot inject result: session {request.chat_session_id} not found or not owned by user") + except Exception as e: + logger.warning(f"[Self-Task] Failed to inject result into chat session: {e}") + + # Broadcast self-task completion event + if _websocket_manager: + try: + await _websocket_manager.broadcast(json.dumps({ + "type": "agent_activity", + "agent_name": agent_name, + "activity_type": "self_task", + "activity_state": "completed" if result.status == TaskExecutionStatus.SUCCESS else "failed", + "action": f"Background task completed", + "timestamp": utc_now_iso(), + "details": { + "execution_id": execution_id, + "chat_session_id": request.chat_session_id, + "cost_usd": result.cost, + "execution_time_ms": execution_time_ms, + "response_preview": (result.response or "")[:200], + "inject_result": request.inject_result, + "result_injected": request.inject_result and request.chat_session_id is not None, + } + })) + except Exception as e: + logger.warning(f"[Self-Task] WebSocket broadcast failed: {e}") - logger.info( - f"[Task Async] Completed background task for agent '{agent_name}', " - f"execution_id={execution_id}, status={result.status}" - ) + logger.info( + f"[Task Async] Completed background task for agent '{agent_name}', " + f"execution_id={execution_id}, status={result.status}" + ) + finally: + # Issue #498: signal any sync HTTP caller waiting on this execution. + # No-op when no waiter is registered (the common async path). + signal_sync_waiter(execution_id, result, chat_session_id) @router.post("/{name}/task") @@ -990,7 +1004,144 @@ def _on_task_done(task: asyncio.Task): "async_mode": True, } - # ---- Sync mode: delegate to TaskExecutionService (EXEC-024) ---- + # ---- Sync mode: pre-acquire slot to mirror async branch (issue #498). + # On success, delegate to TaskExecutionService with slot_already_held=True + # so service finally still releases. On at-capacity, spill to the same + # backlog the async path uses and long-poll on this connection until the + # execution reaches a terminal status. + sync_slot_service = get_slot_service() + sync_max_parallel_tasks = db.get_max_parallel_tasks(name) + sync_effective_timeout = request.timeout_seconds + if sync_effective_timeout is None: + sync_effective_timeout = db.get_execution_timeout(name) + + sync_slot_acquired = await sync_slot_service.acquire_slot( + agent_name=name, + execution_id=execution_id or f"temp-{datetime.utcnow().timestamp()}", + max_parallel_tasks=sync_max_parallel_tasks, + message_preview=request.message[:100] if request.message else "", + timeout_seconds=sync_effective_timeout, + ) + + if not sync_slot_acquired: + # Issue #498: spill sync calls to the SAME backlog the async path uses + # (BACKLOG-001), then await on the open HTTP connection. The drain + # callback fires _run_async_task_with_persistence; that helper signals + # _sync_waiters from its finally so we wake immediately on terminal. + from services.backlog_service import get_backlog_service + sync_backlog = get_backlog_service() + sync_enqueued = await sync_backlog.enqueue( + agent_name=name, + execution_id=execution_id, + request=request, + effective_timeout=sync_effective_timeout, + user_id=current_user.id, + user_email=current_user.email or current_user.username, + subscription_id=_task_subscription_id, + x_source_agent=x_source_agent, + x_mcp_key_id=x_mcp_key_id, + x_mcp_key_name=x_mcp_key_name, + triggered_by=triggered_by, + collaboration_activity_id=collaboration_activity_id, + is_self_task=is_self_task, + self_task_activity_id=self_task_activity_id, + ) + if not sync_enqueued: + # Backlog ALSO full → preserve existing terminal-failure semantics. + if execution_id: + db.update_execution_status( + execution_id=execution_id, + status=TaskExecutionStatus.FAILED, + error=f"Agent at capacity ({sync_max_parallel_tasks}/{sync_max_parallel_tasks} parallel tasks running) and backlog is full", + ) + raise HTTPException( + status_code=429, + detail=( + f"Agent '{name}' is at capacity ({sync_max_parallel_tasks} parallel tasks) " + f"and its backlog is full. Try again later." + ), + ) + + # Long-poll cap: queue wait + execution time, both bounded individually + # by effective_timeout via slot TTL and TaskExecutionService internals. + # Total connection hold ≤ 2 * effective_timeout (Policy B). + sync_wait_cap = 2 * sync_effective_timeout + logger.info( + f"[Task Sync] Agent '{name}' at capacity — execution {execution_id} " + f"queued to backlog; long-polling up to {sync_wait_cap}s" + ) + try: + wait_payload = await wait_for_sync_terminal( + execution_id, timeout=sync_wait_cap + ) + except asyncio.TimeoutError: + raise HTTPException( + status_code=504, + detail=( + f"Sync task on agent '{name}' did not complete within " + f"{sync_wait_cap}s. Execution {execution_id} may still be " + f"running; poll GET /api/agents/{name}/executions/{execution_id}." + ), + ) + + if wait_payload is not None and wait_payload.get("result") is not None: + # Drain happy path — full TaskExecutionResult is available. + result = wait_payload["result"] + sync_chat_session_id = wait_payload.get("chat_session_id") + else: + # Polling fallback caught a non-drain terminal flip (corrupt + # metadata, expire_stale, cleanup recovery). Reconstruct a + # minimal result from the row so the failure-translation block + # below still works. + row = db.get_execution(execution_id) + if row is None: + raise HTTPException( + status_code=503, + detail=f"Execution {execution_id} disappeared while waiting", + ) + from services.task_execution_service import TaskExecutionResult + result = TaskExecutionResult( + execution_id=execution_id, + status=row.status, + response=row.response or "", + cost=row.cost, + context_used=row.context_used, + context_max=row.context_max, + session_id=row.claude_session_id, + error=row.error, + raw_response={ + "response": row.response or "", + "cost": row.cost, + "execution_id": execution_id, + "claude_session_id": row.claude_session_id, + }, + ) + sync_chat_session_id = None + + # Side effects (collaboration activity, chat session persist) were + # handled by _run_async_task_with_persistence inside the drain — do + # NOT repeat them. Just translate failure and build the response. + if result.status == "failed": + if "at capacity" in (result.error or ""): + raise HTTPException( + status_code=429, + detail=f"Agent '{name}' is at capacity. Try again later.", + ) + elif "timed out" in (result.error or ""): + raise HTTPException(status_code=504, detail=result.error) + else: + raise HTTPException( + status_code=503, + detail=result.error or "Failed to execute task. The agent may be unavailable.", + ) + + sync_response_data = result.raw_response or {} + if sync_chat_session_id: + sync_response_data["chat_session_id"] = sync_chat_session_id + sync_response_data["task_execution_id"] = execution_id + return sync_response_data + + # ---- Slot acquired immediately — existing sync path (EXEC-024) ---- task_execution_service = get_task_execution_service() result = await task_execution_service.execute_task( agent_name=name, @@ -1007,6 +1158,7 @@ def _on_task_done(task: asyncio.Task): allowed_tools=request.allowed_tools, system_prompt=request.system_prompt, execution_id=execution_id, + slot_already_held=True, # Issue #498: router pre-acquired ) # Complete collaboration activity based on result diff --git a/src/backend/services/sync_waiter.py b/src/backend/services/sync_waiter.py new file mode 100644 index 000000000..3e6793f12 --- /dev/null +++ b/src/backend/services/sync_waiter.py @@ -0,0 +1,126 @@ +""" +Sync HTTP long-poll waiter (issue #498). + +When a sync `/task` (parallel=true, async=false) call hits an at-capacity agent, +it spills to the same persistent backlog the async path uses (BACKLOG-001) and +holds the HTTP connection open until the execution reaches a terminal status. +This module owns the in-process registry and signal/wait primitives the chat +router uses to coordinate that. + +Key invariants: +- Registry is in-process — multi-worker deployments would need pubsub to fan + signals across processes; that's not the current backend shape. +- Signal is a no-op when no waiter is registered (the common async path + doesn't register one). +- Wait combines an asyncio.Future (set by the drain happy path) with a 5s + DB-poll fallback so terminal flips that don't go through the drain + (corrupt-metadata, expire_stale, cleanup recovery) still wake the caller. +- Registry is cleaned in the wait helper's finally — caller cancellation, + timeout, and normal completion all leave the registry empty. +""" + +from __future__ import annotations + +import asyncio +import logging +from typing import Any, Dict, Optional + +from models import TaskExecutionStatus + +logger = logging.getLogger(__name__) + + +# Module-level state --------------------------------------------------------- + +# execution_id -> Future that resolves to {"result": ..., "chat_session_id": ...} +_sync_waiters: Dict[str, asyncio.Future] = {} + +# DB-poll cadence safety net for terminal flips that don't signal directly. +# Module-level constant so tests can monkeypatch a tighter interval. +SYNC_WAITER_POLL_INTERVAL = 5.0 # seconds + +# Anything that's NOT queued/running counts as terminal for sync waiter purposes. +TERMINAL_TASK_STATUSES = frozenset( + { + TaskExecutionStatus.SUCCESS, + TaskExecutionStatus.FAILED, + TaskExecutionStatus.CANCELLED, + TaskExecutionStatus.SKIPPED, + } +) + + +def signal_sync_waiter( + execution_id: Optional[str], + result: Any, + chat_session_id: Optional[str], +) -> None: + """Notify a sync HTTP caller that the execution has reached terminal state. + + Called from the drain happy path (`_run_async_task_with_persistence` + finally). Safe no-op when no waiter is registered (the normal async fire- + and-forget path does not register one), the future is already done + (caller cancelled), or the execution_id is missing. + """ + if not execution_id: + return + fut = _sync_waiters.get(execution_id) + if fut is None or fut.done(): + return + try: + fut.set_result({"result": result, "chat_session_id": chat_session_id}) + except asyncio.InvalidStateError: + # Already cancelled between the .done() check and set_result. + # Safe to ignore — caller is gone. + pass + + +async def wait_for_sync_terminal( + execution_id: str, + timeout: float, +) -> Optional[Dict[str, Any]]: + """Wait for an execution to reach terminal status. + + Returns: + - {"result": TaskExecutionResult, "chat_session_id": Optional[str]} + when the drain happy path completed and signaled directly. + - None when the polling fallback caught a non-drain terminal flip + (caller must reconstruct response from the DB row). + + Raises: + asyncio.TimeoutError if neither fires within `timeout` seconds. + """ + # Late import keeps this module dependency-light for testing. + from database import db + + fut = asyncio.get_running_loop().create_future() + _sync_waiters[execution_id] = fut + + async def _poll_db(): + # Cheap safety net: peek at the row directly every poll interval. + # Catches terminal flips fired by code paths that don't (or can't) + # signal the waiter — drain spawn-failure, expire_stale, cleanup + # service recovery. Latency cost is bounded at one poll interval. + while True: + await asyncio.sleep(SYNC_WAITER_POLL_INTERVAL) + row = db.get_execution(execution_id) + if row is not None and row.status in TERMINAL_TASK_STATUSES: + return None # signals: poll-fallback hit, no rich result + + poll_task = asyncio.create_task(_poll_db()) + try: + done, pending = await asyncio.wait( + [fut, poll_task], + timeout=timeout, + return_when=asyncio.FIRST_COMPLETED, + ) + if not done: + raise asyncio.TimeoutError() + winner = next(iter(done)) + return winner.result() + finally: + _sync_waiters.pop(execution_id, None) + if not poll_task.done(): + poll_task.cancel() + if not fut.done(): + fut.cancel() diff --git a/tests/unit/test_chat_sync_backlog.py b/tests/unit/test_chat_sync_backlog.py new file mode 100644 index 000000000..be0aa4cf0 --- /dev/null +++ b/tests/unit/test_chat_sync_backlog.py @@ -0,0 +1,307 @@ +""" +Sync `/task` Long-Poll on Backlog Tests (issue #498) + +Sync parallel calls (parallel=true, async=false) used to fail terminally with +429 when the agent was at capacity. After #498 they spill to the SAME backlog +the async path uses (BACKLOG-001) and long-poll on the open HTTP connection +until the execution reaches a terminal status. + +These tests cover the in-process primitives in services/sync_waiter.py: +- `_sync_waiters` registry semantics +- `signal_sync_waiter` — fires registered waiter, no-op otherwise +- `wait_for_sync_terminal` — event happy path, DB-poll fallback, timeout + +The full router integration (pre-acquire → enqueue → wait → return) requires +FastAPI dependency injection plumbing and is left to integration tests; here +we verify the in-process primitives the router relies on. +""" + +from __future__ import annotations + +import asyncio +import sys +import types +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + + +# --------------------------------------------------------------------------- +# Bootstrap: same path-shadow protection used by test_backlog.py — pytest +# auto-adds tests/ which has its own utils package. Backend code does +# `from utils.helpers ...` and we need that to resolve to src/backend/utils. +# --------------------------------------------------------------------------- + +_THIS = Path(__file__).resolve() +_BACKEND = _THIS.parent.parent.parent / "src" / "backend" +_BACKEND_STR = str(_BACKEND) +for _shadow in ("utils", "utils.api_client", "utils.assertions", "utils.cleanup"): + sys.modules.pop(_shadow, None) +while _BACKEND_STR in sys.path: + sys.path.remove(_BACKEND_STR) +sys.path.insert(0, _BACKEND_STR) + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def waiter_module(monkeypatch): + """Import services.sync_waiter against a stub `database` module. + + The wait helper does a late `from database import db` to avoid a load- + time cycle. We install a fake module up-front so the import resolves. + """ + fake_db = MagicMock() + fake_db_module = types.SimpleNamespace(db=fake_db) + monkeypatch.setitem(sys.modules, "database", fake_db_module) + + # Reset cached sync_waiter module so the fake_db patch takes effect. + sys.modules.pop("services.sync_waiter", None) + import services.sync_waiter as sw + + sw._sync_waiters.clear() + # Expose the fake db on the module so tests can stub get_execution per-test. + sw._test_db = fake_db + return sw + + +@pytest.fixture +def fast_poll(monkeypatch, waiter_module): + """Shorten the DB-poll interval so timeout tests don't wall-clock-sleep.""" + monkeypatch.setattr(waiter_module, "SYNC_WAITER_POLL_INTERVAL", 0.05) + return waiter_module + + +# --------------------------------------------------------------------------- +# signal_sync_waiter +# --------------------------------------------------------------------------- + + +class TestSignalSyncWaiter: + def test_noop_when_no_waiter_registered(self, waiter_module): + """Async fire-and-forget path doesn't register a waiter — must be safe.""" + waiter_module.signal_sync_waiter( + "never-registered", result=MagicMock(), chat_session_id=None + ) + assert "never-registered" not in waiter_module._sync_waiters + + def test_noop_when_execution_id_empty(self, waiter_module): + # Either empty string or None must not raise. + waiter_module.signal_sync_waiter("", result=MagicMock(), chat_session_id="cs-1") + waiter_module.signal_sync_waiter(None, result=MagicMock(), chat_session_id="cs-1") + + @pytest.mark.asyncio + async def test_fires_registered_waiter_with_payload(self, waiter_module): + loop = asyncio.get_running_loop() + fut = loop.create_future() + waiter_module._sync_waiters["exec-x"] = fut + + sentinel_result = MagicMock(name="TaskExecutionResult") + waiter_module.signal_sync_waiter( + "exec-x", result=sentinel_result, chat_session_id="cs-9" + ) + + payload = await asyncio.wait_for(fut, timeout=0.1) + assert payload == {"result": sentinel_result, "chat_session_id": "cs-9"} + # signal helper itself doesn't pop the registry (the wait helper does). + waiter_module._sync_waiters.pop("exec-x", None) + + @pytest.mark.asyncio + async def test_silent_when_waiter_already_done(self, waiter_module): + """If the caller disconnected and cancelled the future, signal must not crash.""" + loop = asyncio.get_running_loop() + fut = loop.create_future() + fut.cancel() + waiter_module._sync_waiters["exec-cancelled"] = fut + + # Must not raise InvalidStateError despite cancelled future. + waiter_module.signal_sync_waiter( + "exec-cancelled", result=None, chat_session_id=None + ) + + waiter_module._sync_waiters.pop("exec-cancelled", None) + + +# --------------------------------------------------------------------------- +# wait_for_sync_terminal — event happy path +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +class TestWaitForSyncTerminalEvent: + async def test_returns_payload_when_signaled(self, waiter_module): + """Drain happy path: fired by signal_sync_waiter — wait returns payload.""" + sentinel = object() + + async def _fire_after_delay(): + await asyncio.sleep(0.05) + waiter_module.signal_sync_waiter( + "exec-1", result=sentinel, chat_session_id="cs-1" + ) + + # No DB needed for the event path; stub get_execution to be safe. + waiter_module._test_db.get_execution = MagicMock(return_value=None) + + firer = asyncio.create_task(_fire_after_delay()) + try: + result = await waiter_module.wait_for_sync_terminal( + "exec-1", timeout=2.0 + ) + assert result == {"result": sentinel, "chat_session_id": "cs-1"} + finally: + firer.cancel() + # Registry cleaned up on exit. + assert "exec-1" not in waiter_module._sync_waiters + + async def test_signal_with_none_result_still_returns_payload(self, waiter_module): + """Signal with None result still wakes — the dict envelope is what matters.""" + waiter_module._test_db.get_execution = MagicMock(return_value=None) + + async def _fire(): + await asyncio.sleep(0.02) + waiter_module.signal_sync_waiter( + "exec-2", result=None, chat_session_id=None + ) + + firer = asyncio.create_task(_fire()) + try: + payload = await waiter_module.wait_for_sync_terminal( + "exec-2", timeout=1.0 + ) + assert payload == {"result": None, "chat_session_id": None} + finally: + firer.cancel() + + +# --------------------------------------------------------------------------- +# wait_for_sync_terminal — DB-poll fallback +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +class TestWaitForSyncTerminalPollFallback: + async def test_poll_returns_none_when_db_shows_terminal(self, fast_poll): + """When a non-drain code path flips the row terminal (corrupt + metadata, expire_stale, cleanup recovery), the event is never set. + The poll fallback must catch it and return None to signal the + caller to reconstruct the response from the DB row. + """ + sw = fast_poll + running_row = MagicMock(status="running") + failed_row = MagicMock(status="failed") + sw._test_db.get_execution = MagicMock(side_effect=[running_row, failed_row]) + + result = await sw.wait_for_sync_terminal("exec-poll-1", timeout=2.0) + assert result is None + assert "exec-poll-1" not in sw._sync_waiters + + async def test_poll_recognizes_all_terminal_statuses(self, fast_poll): + """success, failed, cancelled all qualify as terminal.""" + sw = fast_poll + for status in ("success", "failed", "cancelled"): + sw._sync_waiters.clear() + sw._test_db.get_execution = MagicMock(return_value=MagicMock(status=status)) + result = await sw.wait_for_sync_terminal(f"exec-{status}", timeout=1.0) + assert result is None, f"status={status} should be terminal" + + async def test_poll_ignores_non_terminal_statuses(self, fast_poll): + """queued and running are NOT terminal — wait should NOT return on them.""" + sw = fast_poll + sw._test_db.get_execution = MagicMock(return_value=MagicMock(status="queued")) + with pytest.raises(asyncio.TimeoutError): + await sw.wait_for_sync_terminal("exec-stuck", timeout=0.3) + + +# --------------------------------------------------------------------------- +# wait_for_sync_terminal — timeout & cleanup +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +class TestWaitForSyncTerminalTimeout: + async def test_timeout_when_neither_event_nor_poll_fires(self, fast_poll): + sw = fast_poll + sw._test_db.get_execution = MagicMock(return_value=MagicMock(status="running")) + with pytest.raises(asyncio.TimeoutError): + await sw.wait_for_sync_terminal("exec-timeout", timeout=0.2) + # Registry MUST be cleaned even on timeout. + assert "exec-timeout" not in sw._sync_waiters + + async def test_registry_cleaned_on_caller_cancellation(self, fast_poll): + """If the HTTP request is cancelled mid-wait, the wait coroutine + raises CancelledError. Registry must still be cleaned (no leak). + """ + sw = fast_poll + sw._test_db.get_execution = MagicMock(return_value=MagicMock(status="running")) + + async def _wait(): + return await sw.wait_for_sync_terminal("exec-cancel", timeout=10) + + task = asyncio.create_task(_wait()) + await asyncio.sleep(0.05) # let it register + assert "exec-cancel" in sw._sync_waiters + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert "exec-cancel" not in sw._sync_waiters + + +# --------------------------------------------------------------------------- +# Concurrent waiters +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +class TestConcurrentWaiters: + async def test_signals_dont_cross_fire(self, waiter_module): + """Two concurrent waiters for different execution_ids: signaling one + must not wake the other. + """ + sw = waiter_module + sw._test_db.get_execution = MagicMock(return_value=None) + + async def _wait(eid): + return await sw.wait_for_sync_terminal(eid, timeout=2.0) + + t1 = asyncio.create_task(_wait("exec-A")) + t2 = asyncio.create_task(_wait("exec-B")) + await asyncio.sleep(0.05) + assert "exec-A" in sw._sync_waiters + assert "exec-B" in sw._sync_waiters + + sw.signal_sync_waiter("exec-A", result="result-A", chat_session_id="cs-A") + payload_a = await asyncio.wait_for(t1, timeout=1.0) + assert payload_a == {"result": "result-A", "chat_session_id": "cs-A"} + + # exec-B is still waiting; must not have fired. + assert not t2.done() + assert "exec-B" in sw._sync_waiters + + sw.signal_sync_waiter("exec-B", result="result-B", chat_session_id=None) + payload_b = await asyncio.wait_for(t2, timeout=1.0) + assert payload_b == {"result": "result-B", "chat_session_id": None} + + +# --------------------------------------------------------------------------- +# Regression: terminal status frozenset matches the enum +# --------------------------------------------------------------------------- + + +def test_terminal_statuses_match_enum(waiter_module): + """If TaskExecutionStatus grows a new terminal value (e.g. EXPIRED), the + poll fallback must include it or sync waiters miss the wake-up. This + regression test forces a deliberate choice when the enum changes. + """ + from models import TaskExecutionStatus + + non_terminal = {TaskExecutionStatus.QUEUED, TaskExecutionStatus.RUNNING} + expected_terminal = set(TaskExecutionStatus) - non_terminal + assert waiter_module.TERMINAL_TASK_STATUSES == frozenset(expected_terminal), ( + "TaskExecutionStatus enum gained a new value; update " + "TERMINAL_TASK_STATUSES in services/sync_waiter.py to decide whether " + "sync waiters should wake on it." + )