Abilityai · vybe · Apr 25, 2026 · Apr 25, 2026
diff --git 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
@@ -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
@@ -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
@@ -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
@@ -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(