From 750078e7ab8182b870755d0fc6a21826e513afb1 Mon Sep 17 00:00:00 2001 From: vybe Date: Sat, 25 Apr 2026 17:46:13 +0100 Subject: [PATCH] =?UTF-8?q?fix(backlog):=20repair=20drain=20spawn=20?= =?UTF-8?q?=E2=80=94=20lazy-import=20target=20after=20#95=20(#496)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit services/backlog_service.py:240 lazy-imported _execute_task_background from routers.chat, but #95 deleted that function. Every backlog drain attempt failed with ImportError; the exception was swallowed at backlog_service.py:218-228, so BACKLOG-001 (#260) was silently dead. Live observation: 23 drain failures / 24h on a fan-out workload, only surface signal was the per-execution `error` column. Why it shipped silently: the unit happy-path test patched sys.modules["routers.chat"] with a SimpleNamespace stub of whatever attribute name it expected, masking the production breakage. Changes: - Lazy-import _run_async_task_with_persistence (the post-#95 replacement) and adjust the call shape (drop release_slot, drop orphaned task_activity_id; the unified executor handles both). - Capture self-task fields (is_self_task, self_task_activity_id, inject_result) at enqueue time and rehydrate on drain so SELF-EXEC-001 (#264) survives backlog overflow. - Emit a stable log token `backlog_drain_spawn_failed` so log-based detection (Vector / dashboards) can catch import drift or similar spawn-time regressions at fleet scale rather than per-row. - AST-based regression guard in tests/unit/test_backlog.py: TestLazyImportTarget parses routers/chat.py and asserts the import target exists; paired test asserts the lazy-import string matches the validated allow-list. Catches both directions of drift without booting the backend. - Update happy-path test to use the new symbol and kwarg surface; add self-task enqueue+drain round-trip tests. Closes #496 Co-Authored-By: Claude Opus 4.7 (1M context) --- .../parallel-headless-execution.md | 51 +++-- .../feature-flows/persistent-task-backlog.md | 57 ++++-- .../cso-2026-04-25-496-diff.json | 40 ++++ .../cso-2026-04-25-496-diff.md | 74 +++++++ src/backend/routers/chat.py | 5 +- src/backend/services/backlog_service.py | 36 +++- tests/registry.json | 3 +- tests/unit/test_backlog.py | 190 +++++++++++++++++- 8 files changed, 397 insertions(+), 59 deletions(-) create mode 100644 docs/security-reports/cso-2026-04-25-496-diff.json create mode 100644 docs/security-reports/cso-2026-04-25-496-diff.md diff --git a/docs/memory/feature-flows/parallel-headless-execution.md b/docs/memory/feature-flows/parallel-headless-execution.md index 8aa129e73..a9927ee11 100644 --- a/docs/memory/feature-flows/parallel-headless-execution.md +++ b/docs/memory/feature-flows/parallel-headless-execution.md @@ -565,41 +565,36 @@ async def _run_async_task_with_persistence( agent_name: str, request: ParallelTaskRequest, execution_id: str, - task_activity_id: str, collaboration_activity_id: Optional[str], x_source_agent: Optional[str], - release_slot: bool = False, user_id: Optional[int] = None, - user_email: Optional[str] = None + user_email: Optional[str] = None, + subscription_id: Optional[str] = None, + is_self_task: bool = False, + self_task_activity_id: Optional[str] = None, ): """ - Background task execution for async mode. - Runs the task and updates execution record/activities when complete. - Note: This still uses inline logic (not TaskExecutionService) because - async mode needs to manage its own activity IDs and collaboration tracking. + Async /task background wrapper (issue #95). + + Delegates the full execution lifecycle to TaskExecutionService (single + path for slot / activity / sanitization / retry / release with + `slot_already_held=True`) and layers on chat-endpoint-specific post-task + side effects: chat session persistence (THINK-001), `chat_response_ready` + WebSocket broadcast, collaboration activity completion, and SELF-EXEC-001 + `inject_result` handling. """ - try: - # Call agent container (agent_post_with_retry imported from task_execution_service) - response = await agent_post_with_retry(agent_name, "/api/task", payload, ...) - - # Sanitize + update execution record with success - db.update_execution_status(execution_id=execution_id, status="success", ...) - - # Persist to chat session if requested (THINK-001) - # Complete activities - await activity_service.complete_activity(task_activity_id, ...) - - except Exception as e: - # Update execution record with failure - db.update_execution_status(execution_id=execution_id, status="failed", error=str(e)) - - # Complete activities with failure - await activity_service.complete_activity(task_activity_id, status="failed", ...) + # Delegate slot+activity+execution lifecycle to the service + result = await task_service.execute_task( + agent_name=agent_name, + message=request.message, + execution_id=execution_id, + slot_already_held=True, # caller (router or backlog drain) pre-acquired + parent_activity_id=collaboration_activity_id, + # ... other passthrough fields + ) - finally: - # Release slot when task completes (CAPACITY-001) - if slot_service and release_slot: - await slot_service.release_slot(agent_name, execution_id) + # Post-task: chat session persistence, WS broadcast, collab completion, + # self-task activity completion + result injection (if is_self_task). ``` **Endpoint Logic — Async branch** (`src/backend/routers/chat.py:735-808`): diff --git a/docs/memory/feature-flows/persistent-task-backlog.md b/docs/memory/feature-flows/persistent-task-backlog.md index f368aab18..4be2ede05 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, + # is_self_task / self_task_activity_id / inject_result + # threaded through so SELF-EXEC-001 (#264) survives queueing ) ) ``` @@ -149,6 +150,7 @@ identity and request parameters: "create_new_session": false, "chat_session_id": null, "resume_session_id": null, + "inject_result": false, "user_id": 42, "user_email": "user@example.com", "subscription_id": "sub-xyz", @@ -157,7 +159,8 @@ identity and request parameters: "x_mcp_key_name": "my-key", "triggered_by": "manual", "collaboration_activity_id": null, - "task_activity_id": null + "is_self_task": false, + "self_task_activity_id": null } ``` @@ -198,7 +201,8 @@ if not slot_acquired: x_mcp_key_name=x_mcp_key_name, triggered_by=triggered_by, collaboration_activity_id=collaboration_activity_id, - task_activity_id=None, + is_self_task=is_self_task, + self_task_activity_id=self_task_activity_id, ) if enqueued: return {"status": "queued", "execution_id": execution_id, ...} @@ -222,7 +226,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` (including `inject_result`) → spawn `_run_async_task_with_persistence` (#496: was `_execute_task_background`, deleted by #95). | | `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 +235,20 @@ 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. **#496 regression + guard**: `tests/unit/test_backlog.py::TestLazyImportTarget` parses + `routers/chat.py` via AST and asserts the import target exists; a paired test + asserts the lazy-import string in `services/backlog_service.py` matches the + validated allow-list. Catches both directions of drift without booting the + backend (the symptom that allowed #496 to ship: a `SimpleNamespace` mock + injected the missing symbol back in, masking the production `ImportError`). +- Drain spawn failures emit a stable log token `backlog_drain_spawn_failed` + so log-based detection (Vector / dashboards) can catch import drift or + similar spawn-time regressions at fleet scale rather than per-row. +- Self-task fields (`is_self_task`, `self_task_activity_id`) are captured at + enqueue and threaded through drain so SELF-EXEC-001 (#264) `inject_result` + semantics survive backlog overflow. - Identity replayed from `backlog_metadata`; no re-auth at drain time. ### Slot Service — `src/backend/services/slot_service.py` @@ -316,13 +332,13 @@ 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 via `TaskExecutionService`, 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. | ## Testing -Unit tests: `tests/unit/test_backlog.py` (29 tests, no backend/Docker required). +Unit tests: `tests/unit/test_backlog.py` (33 tests, no backend/Docker required). Coverage: - TaskExecutionStatus.QUEUED enum value @@ -331,12 +347,18 @@ Coverage: - ScheduleOperations backlog queries: transition to queued, atomic FIFO claim, agent isolation, release-claim-back, single cancel, bulk cancel for agent, stale expiry (normal + tiny-threshold), list agents with queued -- `BacklogService.enqueue`: under cap succeeds, at cap rejected +- `BacklogService.enqueue`: under cap succeeds, at cap rejected, + self-task fields captured for SELF-EXEC-001 round-trip (#496) - `BacklogService.drain_next`: empty noop, failed-claim releases slot, corrupt metadata marks failed, slot-acquire-failure noop, happy path - spawns background task with reconstructed request + spawns background task with reconstructed request, + self-task fields threaded through to `_run_async_task_with_persistence` (#496) - `SlotService.register_on_release` + `release_slot` fan-out, per-callback exception isolation +- **#496 regression guards**: AST-based check that + `_run_async_task_with_persistence` is defined in `routers/chat.py`, and + inverse check that the lazy-import string in + `services/backlog_service.py` matches the validated allow-list ### Prerequisites @@ -389,8 +411,15 @@ CANCELLED state; task never runs. gets one drain per maintenance tick) - [ ] Backlog entries older than 24h — expired to FAILED on next tick -**Last Tested**: 2026-04-13 (unit tests only; manual scenarios pending) -**Status**: ✅ Unit tests passing; integration testing recommended post-merge +**Last Tested**: 2026-04-25 (unit tests; manual scenarios pending) +**Status**: ✅ 33 unit tests passing; integration testing recommended post-merge + +## Changelog + +| Date | Change | +|---|---| +| 2026-04-13 | Initial implementation (PR #316). | +| 2026-04-25 | **#496 fix**: lazy-import target updated from `_execute_task_background` (deleted by #95) to `_run_async_task_with_persistence`. Drain had been silently failing with `ImportError` since #95 because the existing happy-path test injected a `SimpleNamespace` stub into `sys.modules["routers.chat"]` with whatever attribute name it expected. AST-based regression guards added. Self-task fields (`is_self_task`, `self_task_activity_id`, `inject_result`) now captured at enqueue and rehydrated on drain so SELF-EXEC-001 (#264) survives backlog overflow. Drain spawn failures emit stable log token `backlog_drain_spawn_failed`. Stale `task_activity_id` field dropped from metadata (the post-#95 service tracks CHAT_START itself). | ## Acceptance Criteria Coverage diff --git a/docs/security-reports/cso-2026-04-25-496-diff.json b/docs/security-reports/cso-2026-04-25-496-diff.json new file mode 100644 index 000000000..32bcbf0e4 --- /dev/null +++ b/docs/security-reports/cso-2026-04-25-496-diff.json @@ -0,0 +1,40 @@ +{ + "version": 1, + "date": "2026-04-25", + "issue": 496, + "mode": "daily", + "scope": "diff", + "base": "dev", + "branch": "feature/496-backlog-drain-fix", + "phases_run": [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14], + "files_changed": [ + "src/backend/routers/chat.py", + "src/backend/services/backlog_service.py", + "tests/unit/test_backlog.py" + ], + "lines": {"added": 215, "removed": 16}, + "attack_surface_delta": { + "new_endpoints": 0, + "new_websocket_channels": 0, + "new_mcp_tools": 0, + "new_public_surfaces": 0, + "new_background_jobs": 0, + "new_integrations": 0 + }, + "findings": [], + "supply_chain_summary": { + "python_deps_changed": false, + "node_deps_changed": false, + "lockfiles_changed": false + }, + "filter_stats": {"raw": 0, "after_fp_filter": 0, "verified": 0, "tentative": 0}, + "totals": {"critical": 0, "high": 0, "medium": 0, "low": 0}, + "posture_delta": { + "improvements": [ + "A09: stable log token 'backlog_drain_spawn_failed' added — closes fleet-level silent-failure detection gap", + "Regression guard: AST-based lazy-import target validation prevents future test-drift silently breaking BACKLOG-001" + ], + "regressions": [] + }, + "trend": {"persistent": 0, "resolved": 0, "new": 0, "direction": "neutral_to_positive"} +} diff --git a/docs/security-reports/cso-2026-04-25-496-diff.md b/docs/security-reports/cso-2026-04-25-496-diff.md new file mode 100644 index 000000000..63a2680d0 --- /dev/null +++ b/docs/security-reports/cso-2026-04-25-496-diff.md @@ -0,0 +1,74 @@ +# CSO Diff Audit — 2026-04-25 — Issue #496 + +**Mode**: daily (8/10 confidence gate) +**Scope**: branch diff (`feature/496-backlog-drain-fix` vs `dev`) +**Diff**: 3 files, +215/-16 + +## Files Changed + +- `src/backend/routers/chat.py` (+5/-2) +- `src/backend/services/backlog_service.py` (+27/-9) +- `tests/unit/test_backlog.py` (+183/-5) + +## Attack Surface Delta + +| Category | Count | +|---|---| +| New endpoints | 0 | +| New WebSocket channels | 0 | +| New MCP tools | 0 | +| New public surfaces | 0 | +| New background jobs | 0 | +| New external integrations | 0 | + +Service-layer code-correctness fix; zero new attack surface. + +## Findings + +**None.** + +The diff fixes a dead lazy-import in the persistent task backlog (BACKLOG-001), threads SELF-EXEC-001 self-task fields through queueing, and adds AST-based regression tests for the test-drift class that allowed #496 to ship silently. + +## Posture Delta — net positive + +- **A09 (Logging & Monitoring)** improved: stable log token `backlog_drain_spawn_failed` added at `services/backlog_service.py` drain spawn catch site. Closes the fleet-level silent-failure detection gap that allowed `ImportError` on every drain to go unnoticed for weeks (23 failures/24h on a live fan-out workload, only surface signal was the per-row `error` column). +- **Test-drift class regression guard** added: `tests/unit/test_backlog.py::TestLazyImportTarget` parses `routers/chat.py` via AST and asserts the lazy-import target is defined; complementary test asserts `services/backlog_service.py`'s lazy import string matches the validated allow-list. Catches both directions of drift without booting the backend. + +## OWASP Top 10 (diff-scoped) + +| Category | Result | +|---|---| +| A01 Broken Access Control | No new endpoints. Enqueue site inside authenticated handler. ✓ | +| A02 Cryptographic Failures | No crypto changes. ✓ | +| A03 Injection | Zero subprocess/eval/v-html in diff; metadata fields are typed primitives, parameterized through existing DB op. ✓ | +| A04 Insecure Design | Existing `max_backlog_depth` rate-limit unchanged. ✓ | +| A05 Security Misconfiguration | No CORS/CSP/headers changes. ✓ | +| A07 Authentication Failures | No auth changes. ✓ | +| A08 Software/Data Integrity | Metadata schema delta consumed via defensive `.get(..., default)`; JSON, no pickle. ✓ | +| A09 Logging & Monitoring | **IMPROVED** — stable log token added. ✓ | +| A10 SSRF | No URL construction from user input. ✓ | + +## STRIDE — BacklogService drain path + +| Threat | Result | +|---|---| +| Spoofing | Server-side metadata snapshots from authenticated enqueue. ✓ | +| Tampering | DB-write restricted; corrupt-metadata path marks FAILED safely. ✓ | +| Repudiation | Drain logged with stable token. ✓ | +| Information Disclosure | Log token contains agent_name, execution_id, exception `repr()`. None sensitive. ✓ | +| DoS | Hard-excluded by FP filter. | +| Elevation of Privilege | No privilege change. ✓ | + +## Data Classification + +| Class | Fields in `backlog_metadata` | +|---|---| +| INTERNAL | agent_name, execution_id, message preview, model, allowed_tools, system_prompt, is_self_task, self_task_activity_id, inject_result, x_source_agent, x_mcp_key_id, x_mcp_key_name | +| CONFIDENTIAL | user_id, user_email, subscription_id | +| RESTRICTED | **none** — credentials-in-metadata invariant preserved (only opaque IDs, never values) | + +## Trend + +- Persistent: no findings carried over from `cso-2026-04-23-476-diff.md`. +- New: 0 +- Resolved: N/A (the #496 fix itself is corrective; the underlying class — silent test drift on a swallowed exception path — is closed by the new AST guard tests). diff --git a/src/backend/routers/chat.py b/src/backend/routers/chat.py index 8a807e60a..f3f89df4f 100644 --- a/src/backend/routers/chat.py +++ b/src/backend/routers/chat.py @@ -923,7 +923,10 @@ async def execute_parallel_task( x_mcp_key_name=x_mcp_key_name, triggered_by=triggered_by, collaboration_activity_id=collaboration_activity_id, - task_activity_id=None, # chat_start tracked on drain to keep stream clean + # #496: thread self-task fields so SELF-EXEC-001 (#264) + # inject_result still works when a self-task overflows to backlog. + is_self_task=is_self_task, + self_task_activity_id=self_task_activity_id, ) if enqueued: logger.info( diff --git a/src/backend/services/backlog_service.py b/src/backend/services/backlog_service.py index 082bcf984..80d41c6af 100644 --- a/src/backend/services/backlog_service.py +++ b/src/backend/services/backlog_service.py @@ -16,8 +16,9 @@ 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. (#496: was `_execute_task_background`, + deleted by #95; lazy-import target updated.) - Credentials are never stored in backlog_metadata — only opaque references (subscription_id, user_id, mcp key id). """ @@ -68,7 +69,8 @@ async def enqueue( x_mcp_key_name: Optional[str], triggered_by: str, collaboration_activity_id: Optional[str], - task_activity_id: Optional[str], + is_self_task: bool = False, + self_task_activity_id: Optional[str] = None, ) -> bool: """Persist an async task request as a QUEUED backlog item. @@ -99,6 +101,7 @@ async def enqueue( "create_new_session": request.create_new_session, "chat_session_id": request.chat_session_id, "resume_session_id": request.resume_session_id, + "inject_result": request.inject_result, "user_id": user_id, "user_email": user_email, "subscription_id": subscription_id, @@ -107,7 +110,8 @@ async def enqueue( "x_mcp_key_name": x_mcp_key_name, "triggered_by": triggered_by, "collaboration_activity_id": collaboration_activity_id, - "task_activity_id": task_activity_id, + "is_self_task": is_self_task, + "self_task_activity_id": self_task_activity_id, } queued_at = utc_now_iso() ok = db.update_execution_to_queued( @@ -138,7 +142,8 @@ 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. + (#496: was `_execute_task_background`, deleted by #95.) Returns True if a row was drained, False otherwise. """ @@ -216,8 +221,12 @@ async def drain_next(self, agent_name: str) -> bool: try: await self._spawn_drain(agent_name, execution_id, metadata) except Exception as e: # pragma: no cover - defensive + # #496: stable log token "backlog_drain_spawn_failed" so log-based + # detection (Vector / dashboards) can spot import drift or other + # spawn-time regressions at fleet scale rather than per-row. logger.error( - f"[Backlog] Failed to spawn drain for {execution_id}: {e}", + f"[Backlog] backlog_drain_spawn_failed agent='{agent_name}' " + f"execution_id={execution_id} error={e!r}", exc_info=True, ) db.update_execution_status( @@ -236,8 +245,14 @@ async def _spawn_drain( """Reconstruct a ParallelTaskRequest from metadata and spawn the existing background execution helper. Late-imported to avoid the chat.py <-> backlog_service.py cycle. + + #496: lazy-imports `_run_async_task_with_persistence` (post-#95 + replacement). The drain pre-acquires the slot under the real + execution_id (drain_next), so the helper passes + `slot_already_held=True` to TaskExecutionService, which releases the + slot in its `finally` block. """ - from routers.chat import _execute_task_background + from routers.chat import _run_async_task_with_persistence request = ParallelTaskRequest( message=metadata.get("message") or "", @@ -252,20 +267,21 @@ async def _spawn_drain( create_new_session=metadata.get("create_new_session") or False, chat_session_id=metadata.get("chat_session_id"), resume_session_id=metadata.get("resume_session_id"), + inject_result=metadata.get("inject_result") or False, ) 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, user_id=metadata.get("user_id"), user_email=metadata.get("user_email"), subscription_id=metadata.get("subscription_id"), + is_self_task=metadata.get("is_self_task") or False, + self_task_activity_id=metadata.get("self_task_activity_id"), ) ) diff --git a/tests/registry.json b/tests/registry.json index b67be9c94..bc53ba922 100644 --- a/tests/registry.json +++ b/tests/registry.json @@ -136,8 +136,9 @@ "file": "unit/test_backlog.py", "feature": "BACKLOG-001", "added": "2026-04-13", + "updated": "2026-04-25", "categories": ["backend", "unit", "backlog", "async", "executions"], - "description": "Unit tests for persistent async task backlog (#260): TaskExecutionStatus.QUEUED enum, migration (queued_at/backlog_metadata/max_backlog_depth/partial index), ScheduleOperations claim FIFO atomicity, cancel paths, stale expiry, BacklogService enqueue depth-cap, drain slot-first acquire + corrupt metadata failure, SlotService release callback fan-out." + "description": "Unit tests for persistent async task backlog (#260): TaskExecutionStatus.QUEUED enum, migration (queued_at/backlog_metadata/max_backlog_depth/partial index), ScheduleOperations claim FIFO atomicity, cancel paths, stale expiry, BacklogService enqueue depth-cap, drain slot-first acquire + corrupt metadata failure, SlotService release callback fan-out. #496: AST-based regression guards for the lazy-import target in services/backlog_service.py (catches the test-drift class that masked the broken drain), self-task field capture and round-trip through queueing for SELF-EXEC-001 (#264)." }, { "file": "test_validation.py", diff --git a/tests/unit/test_backlog.py b/tests/unit/test_backlog.py index 5308d03fa..4733e9a4a 100644 --- a/tests/unit/test_backlog.py +++ b/tests/unit/test_backlog.py @@ -23,6 +23,9 @@ - BacklogService.drain_next: slot-first acquire, failed-claim releases slot, corrupt metadata marks FAILED, happy path spawns background task - on_slot_released callback hook fires without blocking +- #496: lazy-import target `_run_async_task_with_persistence` exists in the + real `routers/chat.py` source (AST-based — catches test-drift where a + SimpleNamespace mock would inject the missing symbol back in). """ from __future__ import annotations @@ -562,7 +565,6 @@ async def test_enqueue_under_cap_succeeds(self, fake_db, fake_slots): x_mcp_key_name=None, triggered_by="manual", collaboration_activity_id=None, - task_activity_id=None, ) assert ok is True stored = json.loads(fake_db.queued["exec-1"]) @@ -595,11 +597,51 @@ async def test_enqueue_rejected_when_at_cap(self, fake_db, fake_slots): x_mcp_key_name=None, triggered_by="manual", collaboration_activity_id=None, - task_activity_id=None, ) assert ok is False assert "exec-1" not in fake_db.queued + async def test_enqueue_captures_self_task_fields(self, fake_db, fake_slots): + """#496: SELF-EXEC-001 (#264) inject_result must survive backlog + spillover. enqueue stores is_self_task + self_task_activity_id + + request.inject_result so _spawn_drain can rehydrate the request + and pass them to _run_async_task_with_persistence. + """ + from services.backlog_service import BacklogService + from models import ParallelTaskRequest + + svc = BacklogService() + fake_db.queued_count_value = 0 + fake_db.backlog_depth = 50 + + request = ParallelTaskRequest( + message="self-task at capacity", + async_mode=True, + inject_result=True, + ) + ok = await svc.enqueue( + agent_name="alpha", + execution_id="exec-self-1", + request=request, + effective_timeout=300, + user_id=7, + user_email="u@example.com", + subscription_id="sub-1", + x_source_agent="alpha", # source == target = self-task + x_mcp_key_id=None, + x_mcp_key_name=None, + triggered_by="self_task", + collaboration_activity_id=None, + is_self_task=True, + self_task_activity_id="act-self-9", + ) + assert ok is True + stored = json.loads(fake_db.queued["exec-self-1"]) + assert stored["is_self_task"] is True + assert stored["self_task_activity_id"] == "act-self-9" + assert stored["inject_result"] is True + assert stored["x_source_agent"] == "alpha" + # --------------------------------------------------------------------------- # BacklogService.drain_next @@ -665,6 +707,11 @@ async def test_drain_slot_acquire_failure_is_noop(self, fake_db, monkeypatch): async def test_drain_happy_path_spawns_background( self, fake_db, fake_slots, monkeypatch ): + """#496: drain spawns _run_async_task_with_persistence (post-#95 + replacement for the deleted _execute_task_background). Asserts the + new kwarg surface and that self-task metadata round-trips through + the drain so SELF-EXEC-001 (#264) survives backlog spillover. + """ from services.backlog_service import BacklogService spawned = {} @@ -673,8 +720,11 @@ 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. The + # AST-based regression test (TestLazyImportTarget) separately + # asserts the real routers/chat.py defines this symbol — together + # they catch the test-drift class that produced #496. + fake_chat = types.SimpleNamespace(_run_async_task_with_persistence=_fake_bg) monkeypatch.setitem(sys.modules, "routers.chat", fake_chat) metadata = { @@ -684,6 +734,11 @@ async def _fake_bg(**kwargs): "user_id": 5, "user_email": "u@example.com", "subscription_id": "sub-x", + "collaboration_activity_id": "collab-1", + "x_source_agent": "beta", + "is_self_task": False, + "self_task_activity_id": None, + "inject_result": False, } fake_db.queued_count_value = 1 fake_db.claim_next_return = { @@ -700,8 +755,133 @@ 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 + assert spawned["user_email"] == "u@example.com" + assert spawned["subscription_id"] == "sub-x" + assert spawned["collaboration_activity_id"] == "collab-1" + assert spawned["x_source_agent"] == "beta" + assert spawned["is_self_task"] is False + assert spawned["self_task_activity_id"] is None + + async def test_drain_threads_self_task_fields( + self, fake_db, fake_slots, monkeypatch + ): + """#496: when a queued row was a self-task at enqueue time, drain + rehydrates is_self_task + self_task_activity_id + inject_result + on the request, so _run_async_task_with_persistence completes the + SELF-EXEC-001 activity and injects the result. + """ + from services.backlog_service import BacklogService + + spawned = {} + + async def _fake_bg(**kwargs): + spawned.update(kwargs) + spawned["request_inject_result"] = kwargs["request"].inject_result + + fake_chat = types.SimpleNamespace(_run_async_task_with_persistence=_fake_bg) + monkeypatch.setitem(sys.modules, "routers.chat", fake_chat) + + metadata = { + "message": "self-task at capacity", + "timeout_seconds": 300, + "user_id": 5, + "user_email": "u@example.com", + "subscription_id": "sub-x", + "x_source_agent": "alpha", # source == target + "is_self_task": True, + "self_task_activity_id": "act-self-9", + "inject_result": True, + } + fake_db.queued_count_value = 1 + fake_db.claim_next_return = { + "id": "exec-self-7", + "agent_name": "alpha", + "message": "self-task at capacity", + "backlog_metadata": json.dumps(metadata), + } + + svc = BacklogService() + assert await svc.drain_next("alpha") is True + await asyncio.sleep(0) + assert spawned["is_self_task"] is True + assert spawned["self_task_activity_id"] == "act-self-9" + assert spawned["request_inject_result"] is True + + +# --------------------------------------------------------------------------- +# Lazy-import target validation (#496 — test-drift regression guard) +# --------------------------------------------------------------------------- + + +class TestLazyImportTarget: + """Static (AST-based) check that BacklogService._spawn_drain's lazy-import + target exists in the real routers/chat.py source. + + History: #95 deleted `_execute_task_background` from routers/chat.py and + replaced it with `_run_async_task_with_persistence`, but the lazy import + in services/backlog_service.py was missed. The exception was swallowed + at the call site, so backlog drain silently failed for weeks. The + pre-existing happy-path test injected a SimpleNamespace stub of + routers.chat into sys.modules with whatever attribute name the test + expected, masking the bug. + + This test parses the real routers/chat.py without importing it (the + module pulls in FastAPI, the database singleton, etc., which is too + heavy for unit tests) and asserts the symbol is still defined. + Same idea works for any future rename — keep this in sync with the + lazy import in services/backlog_service.py._spawn_drain. + """ + + LAZY_IMPORT_TARGETS = ("_run_async_task_with_persistence",) + + def test_routers_chat_defines_lazy_import_targets(self): + import ast + + chat_src = _BACKEND / "routers" / "chat.py" + assert chat_src.exists(), f"routers/chat.py not found at {chat_src}" + + tree = ast.parse(chat_src.read_text(), filename=str(chat_src)) + defined = { + node.name + for node in ast.walk(tree) + if isinstance(node, (ast.AsyncFunctionDef, ast.FunctionDef)) + } + for target in self.LAZY_IMPORT_TARGETS: + assert target in defined, ( + f"BacklogService._spawn_drain lazy-imports `{target}` from " + f"routers.chat, but the symbol is not defined there. This is " + f"the failure mode that produced #496 — update either the " + f"lazy import or this allow-list." + ) + + def test_backlog_service_lazy_import_matches_target_list(self): + """Belt-and-suspenders: the lazy import string in + services/backlog_service.py must reference one of the targets + validated above. Catches the inverse drift (someone renames the + symbol in routers/chat.py and forgets to update backlog_service.py + — production breaks; this test catches it). + """ + import re + + backlog_src = _BACKEND / "services" / "backlog_service.py" + text = backlog_src.read_text() + # Match: from routers.chat import + matches = re.findall( + r"from\s+routers\.chat\s+import\s+([A-Za-z_][A-Za-z0-9_]*)", + text, + ) + assert matches, ( + "Expected at least one `from routers.chat import ...` in " + "services/backlog_service.py — has the lazy-import scheme changed?" + ) + for imported in matches: + assert imported in self.LAZY_IMPORT_TARGETS, ( + f"services/backlog_service.py lazy-imports `{imported}` from " + f"routers.chat, but it's not in the validated allow-list " + f"{self.LAZY_IMPORT_TARGETS}. Update the allow-list or fix " + f"the lazy import." + ) # ---------------------------------------------------------------------------