From b5d2be8a393cc9bec1025967115e14435ae656f3 Mon Sep 17 00:00:00 2001 From: Andrii Pasternak Date: Sat, 25 Apr 2026 19:44:46 +0100 Subject: [PATCH 1/4] fix(backlog): repair drain spawn after #95 rename (#496) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `services/backlog_service.py:_spawn_drain` was lazy-importing `_execute_task_background` from `routers.chat`, but #95 (PR #316) deleted that function and replaced it with `_run_async_task_with_persistence`. Every backlog drain raised `ImportError`, was caught at line 218-228, and silently marked queued executions FAILED — leaving BACKLOG-001 (#260) non-functional whenever an agent hit capacity. Rewire the lazy import to the new helper and adjust the call shape: - drop `task_activity_id` (not in new signature; chat router already passes None at enqueue) - drop `release_slot=True` (the wrapper passes `slot_already_held=True` to TaskExecutionService, which manages release in its finally block) - derive `is_self_task` from x_source_agent vs agent_name - pass `self_task_activity_id=None` (queued items don't carry one; separate gap, not in scope here) Add `tests/test_backlog_drain_unit.py` with five regression checks: two AST-based contract tests that pin the function name and signature in `routers/chat.py` (would have caught the original break), and three runtime spy tests covering the kwarg shape `_spawn_drain` forwards. The existing `tests/unit/test_backlog.py::test_drain_happy_path_spawns_background` is updated to match the new contract. Sync the BACKLOG-001 and TaskExecutionService feature-flow docs to reference the renamed helper. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../feature-flows/persistent-task-backlog.md | 18 +- .../feature-flows/task-execution-service.md | 4 +- src/backend/services/backlog_service.py | 24 +- tests/test_backlog_drain_unit.py | 355 ++++++++++++++++++ tests/unit/test_backlog.py | 15 +- 5 files changed, 396 insertions(+), 20 deletions(-) create mode 100644 tests/test_backlog_drain_unit.py diff --git a/docs/memory/feature-flows/persistent-task-backlog.md b/docs/memory/feature-flows/persistent-task-backlog.md index f368aab18..25bbe79ae 100644 --- a/docs/memory/feature-flows/persistent-task-backlog.md +++ b/docs/memory/feature-flows/persistent-task-backlog.md @@ -57,7 +57,7 @@ The backlog gives Trinity: slot acquired slot full │ │ ▼ ▼ - _execute_task_background() backlog.enqueue() + _run_async_task_with_persistence() backlog.enqueue() │ │ ▼ ┌───────────┴───────────┐ finally: release_slot │ │ @@ -87,9 +87,10 @@ The backlog gives Trinity: │ ▼ asyncio.create_task( - _execute_task_background( - release_slot=True, + _run_async_task_with_persistence( identity from backlog_metadata, + # slot release happens inside TaskExecutionService + # via slot_already_held=True (set by the wrapper) ) ) ``` @@ -222,7 +223,7 @@ if _exec_row and _exec_row.status == TaskExecutionStatus.QUEUED: | Method | Purpose | |---|---| | `enqueue(...)` | Check depth, persist `backlog_metadata`, flip row to QUEUED. Returns False if at cap. | -| `drain_next(agent_name)` | Acquire sentinel slot → atomically claim row → swap to real execution_id slot → reconstruct `ParallelTaskRequest` → spawn `_execute_task_background`. | +| `drain_next(agent_name)` | Acquire sentinel slot → atomically claim row → swap to real execution_id slot → reconstruct `ParallelTaskRequest` → spawn `_run_async_task_with_persistence`. | | `on_slot_released(agent_name)` | Callback registered with SlotService. Tries `drain_next` once per release. | | `expire_stale(max_age_hours=24)` | Maintenance: mark old queued rows as FAILED. | | `drain_orphans_all()` | Maintenance: iterate agents with queued work, drain one item each. | @@ -231,8 +232,11 @@ if _exec_row and _exec_row.status == TaskExecutionStatus.QUEUED: Design invariants: - Slot acquired **before** row is claimed — prevents RUNNING-without-slot orphans. - Single-statement `UPDATE ... WHERE id=(SELECT ... LIMIT 1) RETURNING` — atomic claim. -- `_execute_task_background` is late-imported inside `_spawn_drain` to avoid a - `routers.chat` ↔ `services.backlog_service` cycle. +- `_run_async_task_with_persistence` is late-imported inside `_spawn_drain` + to avoid a `routers.chat` ↔ `services.backlog_service` cycle. (Issue #496: + this lazy import was previously named `_execute_task_background` and silently + ImportError'd after #95 renamed the helper; now pinned by + `tests/test_backlog_drain_unit.py`.) - Identity replayed from `backlog_metadata`; no re-auth at drain time. ### Slot Service — `src/backend/services/slot_service.py` @@ -316,7 +320,7 @@ page if demand emerges. | Corrupt `backlog_metadata` JSON | Row marked FAILED with reason, slot released, drain continues with next item. | | Slot acquisition fails after claim | Row released back to QUEUED via `release_claim_to_queued`; next callback retries. | | Backend crash mid-drain | Row stays RUNNING with no Claude session ID — existing cleanup service recovers it within the timeout window. New queued rows are drained by the 60s maintenance loop on restart. | -| Agent container gone when drain fires | `_execute_task_background` surfaces an HTTP error, row marked FAILED. | +| Agent container gone when drain fires | `_run_async_task_with_persistence` surfaces an HTTP error, row marked FAILED. | | Concurrent drains on same agent | Atomic UPDATE guarantees only one callback wins the row; others get None and release their sentinel slots. | | Cancel-while-queued | Terminate endpoint short-circuits, row moves to CANCELLED. The claim SQL's `WHERE status='queued'` filter naturally skips cancelled rows, so the drain path is race-safe. | diff --git a/docs/memory/feature-flows/task-execution-service.md b/docs/memory/feature-flows/task-execution-service.md index 418302c77..fd596ec22 100644 --- a/docs/memory/feature-flows/task-execution-service.md +++ b/docs/memory/feature-flows/task-execution-service.md @@ -58,7 +58,7 @@ Callers inspect `result.status` to decide HTTP response. Status values come from Moved from `routers/chat.py`. Module-level async function. Used by: - `TaskExecutionService.execute_task()` internally (line 249) -- `routers/chat.py` for `/chat` endpoint (line 248) and `_execute_task_background` (line 477) +- `routers/chat.py` for `/chat` endpoint and `_run_async_task_with_persistence` (the async-mode wrapper introduced by #95; previously named `_execute_task_background`) ```python async def agent_post_with_retry( @@ -163,7 +163,7 @@ The endpoint handles: 2. Determine `triggered_by` from headers (lines 686-691) 3. Create execution record early (lines 694-705) -- passed to service as `execution_id` 4. Collaboration tracking for agent-to-agent (lines 710-732) -- stays in router -5. **Async mode branch** (lines 735-808) -- spawns `_execute_task_background()`, does NOT use service +5. **Async mode branch** -- pre-acquires capacity slot, then spawns `_run_async_task_with_persistence()` which delegates to `task_execution_service.execute_task(slot_already_held=True)` (post-#95) 6. **Sync mode branch** (lines 810-827) -- delegates to `task_execution_service.execute_task()` 7. Collaboration activity completion (lines 830-839) 8. Error translation to HTTP exceptions (lines 842-857) diff --git a/src/backend/services/backlog_service.py b/src/backend/services/backlog_service.py index 082bcf984..b5b2f15fc 100644 --- a/src/backend/services/backlog_service.py +++ b/src/backend/services/backlog_service.py @@ -16,8 +16,8 @@ drain), the slot we just acquired is immediately released. - Claim uses a single atomic UPDATE ... WHERE id = (SELECT ... ORDER BY queued_at LIMIT 1) RETURNING so concurrent drains can't double-claim. -- Drain imports `_execute_task_background` lazily to avoid a circular import - with routers/chat.py. +- Drain imports `_run_async_task_with_persistence` lazily to avoid a + circular import with routers/chat.py. - Credentials are never stored in backlog_metadata — only opaque references (subscription_id, user_id, mcp key id). """ @@ -138,7 +138,7 @@ async def drain_next(self, agent_name: str) -> bool: 2. Acquire a slot up-front (using current agent capacity & timeout). 3. Atomically claim the oldest queued row. 4. On any failure after (2), release the slot we grabbed. - 5. Spawn `_execute_task_background` on the reconstituted request. + 5. Spawn `_run_async_task_with_persistence` on the reconstituted request. Returns True if a row was drained, False otherwise. """ @@ -237,7 +237,12 @@ async def _spawn_drain( existing background execution helper. Late-imported to avoid the chat.py <-> backlog_service.py cycle. """ - from routers.chat import _execute_task_background + # Issue #95 renamed this helper from `_execute_task_background` to + # `_run_async_task_with_persistence`. The wrapper passes + # `slot_already_held=True` to TaskExecutionService internally, so the + # drain no longer needs a `release_slot` flag — slot release happens + # in the service's finally block. + from routers.chat import _run_async_task_with_persistence request = ParallelTaskRequest( message=metadata.get("message") or "", @@ -254,18 +259,21 @@ async def _spawn_drain( resume_session_id=metadata.get("resume_session_id"), ) + x_source_agent = metadata.get("x_source_agent") + is_self_task = bool(x_source_agent) and x_source_agent == agent_name + task = asyncio.create_task( - _execute_task_background( + _run_async_task_with_persistence( agent_name=agent_name, request=request, execution_id=execution_id, - task_activity_id=metadata.get("task_activity_id"), collaboration_activity_id=metadata.get("collaboration_activity_id"), - x_source_agent=metadata.get("x_source_agent"), - release_slot=True, + x_source_agent=x_source_agent, user_id=metadata.get("user_id"), user_email=metadata.get("user_email"), subscription_id=metadata.get("subscription_id"), + is_self_task=is_self_task, + self_task_activity_id=None, ) ) diff --git a/tests/test_backlog_drain_unit.py b/tests/test_backlog_drain_unit.py new file mode 100644 index 000000000..ba1e3cab8 --- /dev/null +++ b/tests/test_backlog_drain_unit.py @@ -0,0 +1,355 @@ +""" +Backlog drain spawn unit tests (test_backlog_drain_unit.py) + +Issue #496. Pins two contracts that, if either drifts, silently break +BACKLOG-001: + +1. **Import contract** — `routers.chat` must define a public-enough + `_run_async_task_with_persistence` symbol that `BacklogService._spawn_drain` + can import. The previous symbol (`_execute_task_background`) was deleted by + #95 (PR #316) without updating the lazy import here, leaving every drain + raising `ImportError` and silently marking queued executions as failed. + +2. **Signature contract** — `_spawn_drain` must call the helper with the + kwargs the helper actually accepts. Drift here would also be caught only + at runtime, behind the same exception swallow at + `services/backlog_service.py:218-228`. + +Pure unit test — no backend, no live database, no router import side effects. + +The contract checks (tests 1 + 2) are static AST scans on the source files +and need no stubbing at all. The runtime spy tests (3 + 4 + 5) build their +stubs inside fixtures with `monkeypatch.setitem` so sys.modules is restored +between tests and other unit-style test files in the suite are not polluted. +""" + +from __future__ import annotations + +import ast +import asyncio +import importlib.util +import os +import sys +import types +from datetime import datetime +from typing import Any, Dict, List +from unittest.mock import AsyncMock, MagicMock + +import pytest + +# Make src/backend importable for the AST-based tests (cheap, side-effect-free). +_BACKEND_PATH = os.path.abspath( + os.path.join(os.path.dirname(__file__), "..", "src", "backend") +) +if _BACKEND_PATH not in sys.path: + sys.path.insert(0, _BACKEND_PATH) + + +# Override the backend-requiring autouse fixtures from the package conftest. +@pytest.fixture(scope="session") +def api_client(): + yield None + + +@pytest.fixture(autouse=True) +def cleanup_after_test(): + yield + + +# --------------------------------------------------------------------------- +# Test 1 — Import contract (AST, no imports of routers/chat at runtime) +# --------------------------------------------------------------------------- + + +def test_routers_chat_defines_run_async_task_with_persistence(): + """`_run_async_task_with_persistence` must remain defined at the top of + routers/chat.py. A pure-text/AST check is used (not a real import) so the + test stays fast and self-contained — importing routers/chat would pull in + the full backend dependency graph. If the function is renamed or removed, + this test fails immediately rather than waiting for a capacity-overflow + scenario in production. + """ + chat_path = os.path.join(_BACKEND_PATH, "routers", "chat.py") + with open(chat_path, "r", encoding="utf-8") as f: + tree = ast.parse(f.read(), filename=chat_path) + + names = { + node.name + for node in ast.walk(tree) + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + } + assert "_run_async_task_with_persistence" in names, ( + "`_run_async_task_with_persistence` must remain defined in " + "routers/chat.py — BacklogService._spawn_drain depends on it via " + "lazy import. If this helper is renamed, update " + "services/backlog_service.py at the same time." + ) + + # Negative guard: the prior name must not come back without an + # explicit migration of the drain. + assert "_execute_task_background" not in names, ( + "`_execute_task_background` was deleted by issue #95. If a function " + "with this name is reintroduced, ensure backlog_service.py is " + "updated and remove this guard." + ) + + +# --------------------------------------------------------------------------- +# Test 2 — Signature contract (AST) +# --------------------------------------------------------------------------- + + +def test_run_async_task_with_persistence_signature_includes_drain_kwargs(): + """The helper must accept every kwarg `_spawn_drain` passes. Mirrors the + call-site at `services/backlog_service.py:_spawn_drain`. If the helper + drops or renames any of these parameters without a coordinated update, + this test fails before the runtime ImportError-equivalent shows up. + """ + chat_path = os.path.join(_BACKEND_PATH, "routers", "chat.py") + with open(chat_path, "r", encoding="utf-8") as f: + tree = ast.parse(f.read(), filename=chat_path) + + target = next( + ( + node + for node in ast.walk(tree) + if isinstance(node, ast.AsyncFunctionDef) + and node.name == "_run_async_task_with_persistence" + ), + None, + ) + assert target is not None, "Function not found (covered by other test)." + + params = {a.arg for a in target.args.args} + required = { + "agent_name", + "request", + "execution_id", + "collaboration_activity_id", + "x_source_agent", + "user_id", + "user_email", + "subscription_id", + "is_self_task", + "self_task_activity_id", + } + missing = required - params + assert not missing, ( + f"_run_async_task_with_persistence is missing kwargs that " + f"BacklogService._spawn_drain passes: {sorted(missing)}. " + "Coordinated update needed." + ) + + +# --------------------------------------------------------------------------- +# Test 3+ — Runtime spy on _spawn_drain +# +# We need a real `BacklogService` instance plus a fake `routers.chat` module +# the drain can lazy-import. All sys.modules manipulation is fixture-scoped +# via monkeypatch.setitem so other unit tests in the suite don't see leaked +# stubs. +# --------------------------------------------------------------------------- + + +@pytest.fixture +def backlog_module(monkeypatch): + """Load services.backlog_service in isolation with stubbed dependencies. + + Stubs are installed via monkeypatch (auto-restored after each test) and + the module itself is loaded via importlib so a real backend isn't needed. + """ + # Stub `database` for the late `from database import db` calls in + # backlog_service. The fake_db is configured per-test if needed. + fake_db = MagicMock() + fake_database = types.SimpleNamespace(db=fake_db) + monkeypatch.setitem(sys.modules, "database", fake_database) + + # Stub `utils.helpers` — only `utc_now_iso` is consumed at module load. + if "utils.helpers" not in sys.modules: + helpers = types.ModuleType("utils.helpers") + helpers.utc_now_iso = lambda: datetime.utcnow().isoformat() + "Z" + monkeypatch.setitem(sys.modules, "utils.helpers", helpers) + if "utils" not in sys.modules: + monkeypatch.setitem(sys.modules, "utils", types.ModuleType("utils")) + + # Stub `models` with the symbols backlog_service imports. Use setdefault + # so a richer stub installed by another test (e.g. with `User`) is + # preserved. + if "models" not in sys.modules: + models_mod = types.ModuleType("models") + + class _StubParallelTaskRequest: + def __init__(self, **kwargs): + self.__dict__.update(kwargs) + + models_mod.ParallelTaskRequest = _StubParallelTaskRequest + models_mod.TaskExecutionStatus = MagicMock() + models_mod.User = type("User", (), {}) # forward-compat for other unit tests + monkeypatch.setitem(sys.modules, "models", models_mod) + else: + # Ensure the existing module satisfies what backlog_service imports. + existing = sys.modules["models"] + if not hasattr(existing, "ParallelTaskRequest"): + class _StubParallelTaskRequest: + def __init__(self, **kwargs): + self.__dict__.update(kwargs) + + existing.ParallelTaskRequest = _StubParallelTaskRequest + if not hasattr(existing, "TaskExecutionStatus"): + existing.TaskExecutionStatus = MagicMock() + + # Stub services.slot_service to avoid importing redis. + services_pkg = sys.modules.get("services") or types.ModuleType("services") + monkeypatch.setitem(sys.modules, "services", services_pkg) + + fake_slot = types.ModuleType("services.slot_service") + fake_slot.get_slot_service = lambda: MagicMock() + monkeypatch.setitem(sys.modules, "services.slot_service", fake_slot) + + # Load services.backlog_service via importlib (bypasses services/__init__.py). + bs_path = os.path.join(_BACKEND_PATH, "services", "backlog_service.py") + spec = importlib.util.spec_from_file_location( + "_test_backlog_service", bs_path + ) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) # type: ignore[union-attr] + return module + + +def _install_chat_module_stub(monkeypatch, spy: AsyncMock) -> None: + """Install a fake `routers.chat` exposing `_run_async_task_with_persistence`.""" + if "routers" not in sys.modules: + monkeypatch.setitem(sys.modules, "routers", types.ModuleType("routers")) + fake_chat = types.ModuleType("routers.chat") + fake_chat._run_async_task_with_persistence = spy + monkeypatch.setitem(sys.modules, "routers.chat", fake_chat) + + +def _make_metadata(x_source_agent: Any = None) -> Dict[str, Any]: + return { + "message": "hello", + "model": "claude-sonnet-4-6", + "allowed_tools": ["Read"], + "system_prompt": None, + "timeout_seconds": 600, + "max_turns": None, + "save_to_session": False, + "user_message": None, + "create_new_session": False, + "chat_session_id": None, + "resume_session_id": None, + "user_id": 42, + "user_email": "user@example.com", + "subscription_id": "sub-1", + "x_source_agent": x_source_agent, + "x_mcp_key_id": None, + "x_mcp_key_name": None, + "triggered_by": "agent", + "collaboration_activity_id": "collab-1", + "task_activity_id": "task-1", # ignored by the new helper + } + + +def _run_drain(backlog_module, monkeypatch, metadata, agent_name, execution_id): + """Execute `_spawn_drain` and the spawned coroutine, returning captured kwargs.""" + captured_kwargs: List[Dict[str, Any]] = [] + + async def _spy(**kwargs): + captured_kwargs.append(kwargs) + + spy = AsyncMock(side_effect=_spy) + _install_chat_module_stub(monkeypatch, spy) + + captured_coros: List[Any] = [] + + def _fake_create_task(coro): + captured_coros.append(coro) + + class _T: + def add_done_callback(self, *_a, **_kw): + pass + + def cancel(self): + pass + + return _T() + + monkeypatch.setattr( + backlog_module.asyncio, "create_task", _fake_create_task + ) + + service = backlog_module.BacklogService() + asyncio.run( + service._spawn_drain( + agent_name=agent_name, + execution_id=execution_id, + metadata=metadata, + ) + ) + assert len(captured_coros) == 1, ( + "Expected exactly one asyncio.create_task() call from _spawn_drain" + ) + asyncio.run(captured_coros[0]) + assert len(captured_kwargs) == 1, ( + "Expected exactly one call to _run_async_task_with_persistence" + ) + return captured_kwargs[0] + + +def test_spawn_drain_calls_helper_with_expected_kwargs(backlog_module, monkeypatch): + """Runtime contract: _spawn_drain forwards the right kwargs.""" + kwargs = _run_drain( + backlog_module, + monkeypatch, + _make_metadata(x_source_agent="caller-agent"), + agent_name="target-agent", + execution_id="exec-1", + ) + + assert kwargs["agent_name"] == "target-agent" + assert kwargs["execution_id"] == "exec-1" + assert kwargs["collaboration_activity_id"] == "collab-1" + assert kwargs["x_source_agent"] == "caller-agent" + assert kwargs["user_id"] == 42 + assert kwargs["user_email"] == "user@example.com" + assert kwargs["subscription_id"] == "sub-1" + assert kwargs["is_self_task"] is False # caller != target + assert kwargs["self_task_activity_id"] is None + + request = kwargs["request"] + assert request.message == "hello" + assert request.async_mode is True + assert request.timeout_seconds == 600 + + # Ensure no stale params from the deleted helper leak through. + assert "task_activity_id" not in kwargs + assert "release_slot" not in kwargs + + +def test_spawn_drain_marks_self_task_when_source_matches_target( + backlog_module, monkeypatch +): + """is_self_task is derived from x_source_agent == agent_name.""" + kwargs = _run_drain( + backlog_module, + monkeypatch, + _make_metadata(x_source_agent="self-agent"), + agent_name="self-agent", + execution_id="exec-2", + ) + assert kwargs["is_self_task"] is True + + +def test_spawn_drain_marks_non_self_when_source_missing( + backlog_module, monkeypatch +): + """No x_source_agent → not a self-task.""" + kwargs = _run_drain( + backlog_module, + monkeypatch, + _make_metadata(x_source_agent=None), + agent_name="target-agent", + execution_id="exec-3", + ) + assert kwargs["is_self_task"] is False + assert kwargs["x_source_agent"] is None diff --git a/tests/unit/test_backlog.py b/tests/unit/test_backlog.py index 5308d03fa..ef808aa1d 100644 --- a/tests/unit/test_backlog.py +++ b/tests/unit/test_backlog.py @@ -673,8 +673,12 @@ async def _fake_bg(**kwargs): spawned.update(kwargs) # Install a fake routers.chat module so the late import inside - # _spawn_drain picks up our stub instead of the real one. - fake_chat = types.SimpleNamespace(_execute_task_background=_fake_bg) + # _spawn_drain picks up our stub instead of the real one. Issue #496: + # the helper was renamed from `_execute_task_background` to + # `_run_async_task_with_persistence` by #95; the stub must match. + fake_chat = types.SimpleNamespace( + _run_async_task_with_persistence=_fake_bg, + ) monkeypatch.setitem(sys.modules, "routers.chat", fake_chat) metadata = { @@ -700,8 +704,13 @@ async def _fake_bg(**kwargs): await asyncio.sleep(0) assert spawned["agent_name"] == "alpha" assert spawned["execution_id"] == "exec-7" - assert spawned["release_slot"] is True assert spawned["user_id"] == 5 + # `release_slot` was removed in #95 — slot release is now always + # handled inside TaskExecutionService via slot_already_held=True. + assert "release_slot" not in spawned + # Self-task derivation: no x_source_agent → not a self-task. + assert spawned["is_self_task"] is False + assert spawned["self_task_activity_id"] is None # --------------------------------------------------------------------------- From 427ddb32e2e6bde87eb81adfe45568989be84e08 Mon Sep 17 00:00:00 2001 From: Andrii Pasternak Date: Sat, 25 Apr 2026 22:13:07 +0100 Subject: [PATCH 2/4] docs(feature-flows): sync index + parallel-capacity for #496 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add #496 entry to feature-flows.md Recent Updates. - Fix two more stale `release_slot=True` references in parallel-capacity.md left over from #95 — the param never existed on `_run_async_task_with_persistence` (slot release happens inside TaskExecutionService via slot_already_held=True). Other stale `release_slot=True` references in authenticated-chat-tab.md and parallel-headless-execution.md are deeper drift (separate flows, not touched by #496) — leave for a follow-up doc-cleanup pass. Co-Authored-By: Claude Opus 4.7 (1M context) --- docs/memory/feature-flows.md | 1 + docs/memory/feature-flows/parallel-capacity.md | 6 +++--- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/docs/memory/feature-flows.md b/docs/memory/feature-flows.md index a3f8f6caa..7b00ded58 100644 --- a/docs/memory/feature-flows.md +++ b/docs/memory/feature-flows.md @@ -12,6 +12,7 @@ | Date | ID | Feature | Flow | |------|-----|---------|------| | 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 | #496 | Backlog drain spawn fix — repair `_spawn_drain` lazy import after #95 renamed `_execute_task_background` → `_run_async_task_with_persistence`; AST-based regression tests pin the contract | [persistent-task-backlog.md](feature-flows/persistent-task-backlog.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) | | 2026-04-22 | #458 | `.gitignore` init fix — `initialize_git_in_container` now appends missing patterns instead of truncate-and-write; adds `.env`, `.env.*`, `.mcp.json` to the default list and runs for both `/home/developer` and legacy `/home/developer/workspace` (stops credential leak on first GitHub sync) | [github-repo-initialization.md](feature-flows/github-repo-initialization.md) | diff --git a/docs/memory/feature-flows/parallel-capacity.md b/docs/memory/feature-flows/parallel-capacity.md index 5f6d6aa8e..3abdd730e 100644 --- a/docs/memory/feature-flows/parallel-capacity.md +++ b/docs/memory/feature-flows/parallel-capacity.md @@ -61,8 +61,8 @@ The frontend displays slot usage as a vertical capacity meter bar on the Agents │ │ 1. Create execution record in database (chat.py:602-613) │ │ │ │ 2. Router acquires slot directly (chat.py:644-651) │ │ │ │ 3. If full → 429 response (chat.py:653-663) │ │ -│ │ 4. Spawn _run_async_task_with_persistence() with release_slot=True │ │ -│ │ 5. Background task releases slot in finally (chat.py:554-557) │ │ +│ │ 4. Spawn _run_async_task_with_persistence() (router pre-acquired)│ │ +│ │ 5. TaskExecutionService releases slot in finally (slot_already_held=True)│ │ │ │ │ │ │ PUBLIC path (public.py:315-322 → task_execution_service.py): │ │ │ │ 1. Delegate to TaskExecutionService.execute_task() │ │ @@ -137,7 +137,7 @@ POST /api/agents/{name}/task (async) │ 1. db.get_max_parallel_tasks(name) │ │ 2. slot_service.acquire_slot(...) │ │ 3. If not acquired → 429 Too Many Requests │ - │ 4. Spawn background task with release_slot=True │ + │ 4. Spawn `_run_async_task_with_persistence()` │ └─────────────────────────────────────────────────┘ ``` From 1ce8af43fba957527a9c822dd325c0b347776bbd Mon Sep 17 00:00:00 2001 From: Andrii Pasternak Date: Sat, 25 Apr 2026 22:19:18 +0100 Subject: [PATCH 3/4] test(catalog): register test_backlog_drain_unit.py (#496) Adds the new BACKLOG-001 regression test file to tests/registry.json so it shows up in the catalog alongside test_event_bus.py and unit/test_backlog.py. Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/registry.json | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tests/registry.json b/tests/registry.json index b67be9c94..ebdaa7bce 100644 --- a/tests/registry.json +++ b/tests/registry.json @@ -180,6 +180,13 @@ "added": "2026-04-21", "categories": ["backend", "unit", "websocket", "redis-streams"], "description": "Unit tests for Redis Streams event bus (#306): last-event-id validation + id comparison, scope visibility (SCOPE_ALL vs SCOPE_SCOPED with accessible_agents), EventBus XADD envelope (dict + legacy JSON string + inferred agent_name), Redis-unavailable fallback buffer, StreamDispatcher 3-failure client eviction, slow-consumer queue overflow triggers resync marker, update_accessible_agents mutation, invalid last-event-id queues resync_required" + }, + { + "file": "test_backlog_drain_unit.py", + "feature": "Issue #496", + "added": "2026-04-25", + "categories": ["backend", "unit", "backlog", "executions"], + "description": "Regression tests for BACKLOG-001 drain spawn (#496). Two AST-based contract tests pin the function name and signature in routers/chat.py (would have caught #95 silent rename). Three runtime spy tests verify _spawn_drain forwards the correct kwargs and derives is_self_task from x_source_agent." } ] } From 715ce357bf8961edebd8550858353e37a730b5a8 Mon Sep 17 00:00:00 2001 From: Andrii Pasternak Date: Sat, 25 Apr 2026 22:49:16 +0100 Subject: [PATCH 4/4] test: drop redundant test_backlog_drain_unit.py PR #500 (which superseded the original #496 fix scope) shipped equivalent contract coverage with a more robust setup: - `TestLazyImportTarget` (AST guard for the lazy-import target) - `test_drain_threads_self_task_fields` (round-trip via real BacklogService against sqlite) The local file used sys.modules stubs which were strictly weaker. Keeping it would only add maintenance burden for duplicate coverage, so drop the file and its registry entry. Net effect on PR #503 is that it becomes a small, focused docs-cleanup PR (parallel-capacity.md and task-execution-service.md drift from #95, plus the missing Recent Updates entry for #496). Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/registry.json | 7 - tests/test_backlog_drain_unit.py | 361 ------------------------------- 2 files changed, 368 deletions(-) delete mode 100644 tests/test_backlog_drain_unit.py diff --git a/tests/registry.json b/tests/registry.json index 1de807242..bc53ba922 100644 --- a/tests/registry.json +++ b/tests/registry.json @@ -181,13 +181,6 @@ "added": "2026-04-21", "categories": ["backend", "unit", "websocket", "redis-streams"], "description": "Unit tests for Redis Streams event bus (#306): last-event-id validation + id comparison, scope visibility (SCOPE_ALL vs SCOPE_SCOPED with accessible_agents), EventBus XADD envelope (dict + legacy JSON string + inferred agent_name), Redis-unavailable fallback buffer, StreamDispatcher 3-failure client eviction, slow-consumer queue overflow triggers resync marker, update_accessible_agents mutation, invalid last-event-id queues resync_required" - }, - { - "file": "test_backlog_drain_unit.py", - "feature": "Issue #496", - "added": "2026-04-25", - "categories": ["backend", "unit", "backlog", "executions"], - "description": "Regression tests for BACKLOG-001 drain spawn (#496). Two AST-based contract tests pin the function name and signature in routers/chat.py (would have caught #95 silent rename). Three runtime spy tests verify _spawn_drain forwards the correct kwargs and derives is_self_task from x_source_agent." } ] } diff --git a/tests/test_backlog_drain_unit.py b/tests/test_backlog_drain_unit.py deleted file mode 100644 index 7e5a1aecc..000000000 --- a/tests/test_backlog_drain_unit.py +++ /dev/null @@ -1,361 +0,0 @@ -""" -Backlog drain spawn unit tests (test_backlog_drain_unit.py) - -Issue #496. Pins two contracts that, if either drifts, silently break -BACKLOG-001: - -1. **Import contract** — `routers.chat` must define a public-enough - `_run_async_task_with_persistence` symbol that `BacklogService._spawn_drain` - can import. The previous symbol (`_execute_task_background`) was deleted by - #95 (PR #316) without updating the lazy import here, leaving every drain - raising `ImportError` and silently marking queued executions as failed. - -2. **Signature contract** — `_spawn_drain` must call the helper with the - kwargs the helper actually accepts. Drift here would also be caught only - at runtime, behind the same exception swallow at - `services/backlog_service.py:218-228`. - -Pure unit test — no backend, no live database, no router import side effects. - -The contract checks (tests 1 + 2) are static AST scans on the source files -and need no stubbing at all. The runtime spy tests (3 + 4 + 5) build their -stubs inside fixtures with `monkeypatch.setitem` so sys.modules is restored -between tests and other unit-style test files in the suite are not polluted. -""" - -from __future__ import annotations - -import ast -import asyncio -import importlib.util -import os -import sys -import types -from datetime import datetime -from typing import Any, Dict, List -from unittest.mock import AsyncMock, MagicMock - -import pytest - -# Make src/backend importable for the AST-based tests (cheap, side-effect-free). -_BACKEND_PATH = os.path.abspath( - os.path.join(os.path.dirname(__file__), "..", "src", "backend") -) -if _BACKEND_PATH not in sys.path: - sys.path.insert(0, _BACKEND_PATH) - - -# Override the backend-requiring autouse fixtures from the package conftest. -@pytest.fixture(scope="session") -def api_client(): - yield None - - -@pytest.fixture(autouse=True) -def cleanup_after_test(): - yield - - -# --------------------------------------------------------------------------- -# Test 1 — Import contract (AST, no imports of routers/chat at runtime) -# --------------------------------------------------------------------------- - - -def test_routers_chat_defines_run_async_task_with_persistence(): - """`_run_async_task_with_persistence` must remain defined at the top of - routers/chat.py. A pure-text/AST check is used (not a real import) so the - test stays fast and self-contained — importing routers/chat would pull in - the full backend dependency graph. If the function is renamed or removed, - this test fails immediately rather than waiting for a capacity-overflow - scenario in production. - """ - chat_path = os.path.join(_BACKEND_PATH, "routers", "chat.py") - with open(chat_path, "r", encoding="utf-8") as f: - tree = ast.parse(f.read(), filename=chat_path) - - names = { - node.name - for node in ast.walk(tree) - if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) - } - assert "_run_async_task_with_persistence" in names, ( - "`_run_async_task_with_persistence` must remain defined in " - "routers/chat.py — BacklogService._spawn_drain depends on it via " - "lazy import. If this helper is renamed, update " - "services/backlog_service.py at the same time." - ) - - # Negative guard: the prior name must not come back without an - # explicit migration of the drain. - assert "_execute_task_background" not in names, ( - "`_execute_task_background` was deleted by issue #95. If a function " - "with this name is reintroduced, ensure backlog_service.py is " - "updated and remove this guard." - ) - - -# --------------------------------------------------------------------------- -# Test 2 — Signature contract (AST) -# --------------------------------------------------------------------------- - - -def test_run_async_task_with_persistence_signature_includes_drain_kwargs(): - """The helper must accept every kwarg `_spawn_drain` passes. Mirrors the - call-site at `services/backlog_service.py:_spawn_drain`. If the helper - drops or renames any of these parameters without a coordinated update, - this test fails before the runtime ImportError-equivalent shows up. - """ - chat_path = os.path.join(_BACKEND_PATH, "routers", "chat.py") - with open(chat_path, "r", encoding="utf-8") as f: - tree = ast.parse(f.read(), filename=chat_path) - - target = next( - ( - node - for node in ast.walk(tree) - if isinstance(node, ast.AsyncFunctionDef) - and node.name == "_run_async_task_with_persistence" - ), - None, - ) - assert target is not None, "Function not found (covered by other test)." - - params = {a.arg for a in target.args.args} - required = { - "agent_name", - "request", - "execution_id", - "collaboration_activity_id", - "x_source_agent", - "user_id", - "user_email", - "subscription_id", - "is_self_task", - "self_task_activity_id", - } - missing = required - params - assert not missing, ( - f"_run_async_task_with_persistence is missing kwargs that " - f"BacklogService._spawn_drain passes: {sorted(missing)}. " - "Coordinated update needed." - ) - - -# --------------------------------------------------------------------------- -# Test 3+ — Runtime spy on _spawn_drain -# -# We need a real `BacklogService` instance plus a fake `routers.chat` module -# the drain can lazy-import. All sys.modules manipulation is fixture-scoped -# via monkeypatch.setitem so other unit tests in the suite don't see leaked -# stubs. -# --------------------------------------------------------------------------- - - -@pytest.fixture -def backlog_module(monkeypatch): - """Load services.backlog_service in isolation with stubbed dependencies. - - Stubs are installed via monkeypatch (auto-restored after each test) and - the module itself is loaded via importlib so a real backend isn't needed. - """ - # Stub `database` for the late `from database import db` calls in - # backlog_service. The fake_db is configured per-test if needed. - fake_db = MagicMock() - fake_database = types.SimpleNamespace(db=fake_db) - monkeypatch.setitem(sys.modules, "database", fake_database) - - # Stub `utils.helpers` — only `utc_now_iso` is consumed at module load. - if "utils.helpers" not in sys.modules: - helpers = types.ModuleType("utils.helpers") - helpers.utc_now_iso = lambda: datetime.utcnow().isoformat() + "Z" - monkeypatch.setitem(sys.modules, "utils.helpers", helpers) - if "utils" not in sys.modules: - monkeypatch.setitem(sys.modules, "utils", types.ModuleType("utils")) - - # Stub `models` with the symbols backlog_service imports. Use setdefault - # so a richer stub installed by another test (e.g. with `User`) is - # preserved. - if "models" not in sys.modules: - models_mod = types.ModuleType("models") - - class _StubParallelTaskRequest: - def __init__(self, **kwargs): - self.__dict__.update(kwargs) - - models_mod.ParallelTaskRequest = _StubParallelTaskRequest - models_mod.TaskExecutionStatus = MagicMock() - models_mod.User = type("User", (), {}) # forward-compat for other unit tests - monkeypatch.setitem(sys.modules, "models", models_mod) - else: - # Ensure the existing module satisfies what backlog_service imports. - existing = sys.modules["models"] - if not hasattr(existing, "ParallelTaskRequest"): - class _StubParallelTaskRequest: - def __init__(self, **kwargs): - self.__dict__.update(kwargs) - - existing.ParallelTaskRequest = _StubParallelTaskRequest - if not hasattr(existing, "TaskExecutionStatus"): - existing.TaskExecutionStatus = MagicMock() - - # Stub services.slot_service to avoid importing redis. - services_pkg = sys.modules.get("services") or types.ModuleType("services") - monkeypatch.setitem(sys.modules, "services", services_pkg) - - fake_slot = types.ModuleType("services.slot_service") - fake_slot.get_slot_service = lambda: MagicMock() - monkeypatch.setitem(sys.modules, "services.slot_service", fake_slot) - - # Load services.backlog_service via importlib (bypasses services/__init__.py). - bs_path = os.path.join(_BACKEND_PATH, "services", "backlog_service.py") - spec = importlib.util.spec_from_file_location( - "_test_backlog_service", bs_path - ) - module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) # type: ignore[union-attr] - return module - - -def _install_chat_module_stub(monkeypatch, spy: AsyncMock) -> None: - """Install a fake `routers.chat` exposing `_run_async_task_with_persistence`.""" - if "routers" not in sys.modules: - monkeypatch.setitem(sys.modules, "routers", types.ModuleType("routers")) - fake_chat = types.ModuleType("routers.chat") - fake_chat._run_async_task_with_persistence = spy - monkeypatch.setitem(sys.modules, "routers.chat", fake_chat) - - -def _make_metadata(x_source_agent: Any = None) -> Dict[str, Any]: - return { - "message": "hello", - "model": "claude-sonnet-4-6", - "allowed_tools": ["Read"], - "system_prompt": None, - "timeout_seconds": 600, - "max_turns": None, - "save_to_session": False, - "user_message": None, - "create_new_session": False, - "chat_session_id": None, - "resume_session_id": None, - "user_id": 42, - "user_email": "user@example.com", - "subscription_id": "sub-1", - "x_source_agent": x_source_agent, - "x_mcp_key_id": None, - "x_mcp_key_name": None, - "triggered_by": "agent", - "collaboration_activity_id": "collab-1", - "task_activity_id": "task-1", # ignored by the new helper - } - - -def _run_drain(backlog_module, monkeypatch, metadata, agent_name, execution_id): - """Execute `_spawn_drain` and the spawned coroutine, returning captured kwargs.""" - captured_kwargs: List[Dict[str, Any]] = [] - - async def _spy(**kwargs): - captured_kwargs.append(kwargs) - - spy = AsyncMock(side_effect=_spy) - _install_chat_module_stub(monkeypatch, spy) - - captured_coros: List[Any] = [] - - def _fake_create_task(coro): - captured_coros.append(coro) - - class _T: - def add_done_callback(self, *_a, **_kw): - pass - - def cancel(self): - pass - - return _T() - - monkeypatch.setattr( - backlog_module.asyncio, "create_task", _fake_create_task - ) - - service = backlog_module.BacklogService() - asyncio.run( - service._spawn_drain( - agent_name=agent_name, - execution_id=execution_id, - metadata=metadata, - ) - ) - assert len(captured_coros) == 1, ( - "Expected exactly one asyncio.create_task() call from _spawn_drain" - ) - asyncio.run(captured_coros[0]) - assert len(captured_kwargs) == 1, ( - "Expected exactly one call to _run_async_task_with_persistence" - ) - return captured_kwargs[0] - - -def test_spawn_drain_calls_helper_with_expected_kwargs(backlog_module, monkeypatch): - """Runtime contract: _spawn_drain forwards the right kwargs.""" - kwargs = _run_drain( - backlog_module, - monkeypatch, - _make_metadata(x_source_agent="caller-agent"), - agent_name="target-agent", - execution_id="exec-1", - ) - - assert kwargs["agent_name"] == "target-agent" - assert kwargs["execution_id"] == "exec-1" - assert kwargs["collaboration_activity_id"] == "collab-1" - assert kwargs["x_source_agent"] == "caller-agent" - assert kwargs["user_id"] == 42 - assert kwargs["user_email"] == "user@example.com" - assert kwargs["subscription_id"] == "sub-1" - assert kwargs["is_self_task"] is False # caller != target - assert kwargs["self_task_activity_id"] is None - - request = kwargs["request"] - assert request.message == "hello" - assert request.async_mode is True - assert request.timeout_seconds == 600 - - # Ensure no stale params from the deleted helper leak through. - assert "task_activity_id" not in kwargs - assert "release_slot" not in kwargs - - -def test_spawn_drain_passes_through_self_task_from_metadata( - backlog_module, monkeypatch -): - """Post-#500: `is_self_task` and `self_task_activity_id` are captured by - the chat router at enqueue time and replayed verbatim by the drain - (rather than re-derived). Pin that pass-through contract.""" - metadata = _make_metadata(x_source_agent="self-agent") - metadata["is_self_task"] = True - metadata["self_task_activity_id"] = "selftask-1" - kwargs = _run_drain( - backlog_module, - monkeypatch, - metadata, - agent_name="self-agent", - execution_id="exec-2", - ) - assert kwargs["is_self_task"] is True - assert kwargs["self_task_activity_id"] == "selftask-1" - - -def test_spawn_drain_marks_non_self_when_source_missing( - backlog_module, monkeypatch -): - """No x_source_agent → not a self-task.""" - kwargs = _run_drain( - backlog_module, - monkeypatch, - _make_metadata(x_source_agent=None), - agent_name="target-agent", - execution_id="exec-3", - ) - assert kwargs["is_self_task"] is False - assert kwargs["x_source_agent"] is None