feat(28-02): emit structured telemetry from WorkerBoundary background task failures

- Call emit_failure_telemetry("worker_task_failed", run_id=run_id,
  error=str(exc)) inside WorkerBoundary._run's existing except block,
  alongside the existing self._status[run_id] = f"error:{exc}" assignment,
  so background workflow failures are visible to log-based monitoring
  instead of only an in-memory status dict polled via get_status.
- Existing status-string contract (get_status/is_done semantics) is
  unchanged; only the additional telemetry call was added.
- Extend tests/test_worker_boundary.py's error-path test to capture stderr
  and assert the emitted worker_task_failed telemetry line is parseable
  JSON containing run_id and error.

Author: Aitomates

maf_starter/worker_boundary.py

@@ -4,6 +4,8 @@
 from enum import Enum
 from typing import Any, Awaitable, Callable
 
+from maf_starter.telemetry import emit_failure_telemetry
+
 
 class WorkerProfile(str, Enum):
     LOCAL = "local"
@@ -58,6 +60,7 @@ async def _run(self, run_id: str, workflow: Callable[[], Awaitable[Any]]) -> Non
             self._status[run_id] = "done"
         except Exception as exc:  # noqa: BLE001
             self._status[run_id] = f"error:{exc}"
+            emit_failure_telemetry("worker_task_failed", run_id=run_id, error=str(exc))
 
     def get_status(self, run_id: str) -> _RunStatus | None:
         """Return the current status string for *run_id*, or None if unknown."""

tests/test_worker_boundary.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
+import io
+import json
 import unittest
 
 from maf_starter.worker_boundary import WorkerBoundary, WorkerProfile
@@ -71,19 +74,25 @@ async def controlled_workflow():
 
     async def test_status_records_error_on_exception(self) -> None:
         boundary = WorkerBoundary()
+        stderr = io.StringIO()
 
         async def failing_workflow():
             raise ValueError("something went wrong")
 
-        boundary.submit_async("run-3", failing_workflow)
-        # Yield so the task can run and fail
-        await asyncio.sleep(0)
+        with contextlib.redirect_stderr(stderr):
+            boundary.submit_async("run-3", failing_workflow)
+            # Yield so the task can run and fail
+            await asyncio.sleep(0)
 
         status = boundary.get_status("run-3")
         assert status is not None
         self.assertTrue(status.startswith("error:"), f"Expected error: prefix, got: {status!r}")
         self.assertIn("something went wrong", status)
         self.assertTrue(boundary.is_done("run-3"))
+        payload = json.loads(stderr.getvalue().strip())
+        self.assertEqual(payload["event"], "worker_task_failed")
+        self.assertEqual(payload["run_id"], "run-3")
+        self.assertEqual(payload["error"], "something went wrong")
 
     def test_get_status_returns_none_for_unknown_run(self) -> None:
         boundary = WorkerBoundary()