feat(28-02): structured JSON failure telemetry + CLI fallback size guards

maf_starter/provider_fallback.py

@@ -68,9 +68,12 @@ async def get_final_response(self):
 from maf_starter.execution_profile import LOCAL_PROFILE, ExecutionProfile
 from maf_starter.routing_policy import RoutingPlan, build_routing_plan
 from maf_starter.routing_types import CapabilityChange, ChainStep, RouteAttempt
+from maf_starter.telemetry import emit_failure_telemetry
 
 
 FALLBACK_NOTICE = "[Fallback provider used because the earlier model/provider in the chain failed.]\n"
+MAX_CLI_OUTPUT_BYTES = 1_000_000
+MAX_CLI_PROMPT_BYTES = 1_000_000
 FALLBACK_ERROR_MARKERS = (
     "resource_exhausted",
     "quota",
@@ -132,6 +135,12 @@ async def fallback_middleware(context: ChatContext, call_next):
                         error=last_error,
                     )
                 )
+                emit_failure_telemetry(
+                    "provider_failed",
+                    provider=route.primary_provider,
+                    model=route.primary_model,
+                    error=str(exc),
+                )
                 for step in route.fallback_steps:
                     try:
                         context.result = await _execute_chain_step(
@@ -143,6 +152,12 @@ async def fallback_middleware(context: ChatContext, call_next):
                             attempt_log=attempt_log,
                             fallback_index=len(attempt_log),
                         )
+                        emit_failure_telemetry(
+                            "fallback_succeeded",
+                            provider=step.provider,
+                            model=step.model or step.label,
+                            primary_error=str(last_error),
+                        )
                         return
                     except Exception as fallback_exc:
                         last_error = fallback_exc
@@ -156,8 +171,21 @@ async def fallback_middleware(context: ChatContext, call_next):
                                 error=fallback_exc,
                             )
                         )
+                        emit_failure_telemetry(
+                            "fallback_step_failed",
+                            provider=step.provider,
+                            model=step.model or step.label,
+                            error=str(fallback_exc),
+                            fallback_index=len(attempt_log) - 1,
+                        )
                         continue
 
+                emit_failure_telemetry(
+                    "fallback_exhausted",
+                    primary_provider=route.primary_provider,
+                    attempted_providers=[attempt.provider for attempt in attempt_log],
+                    final_error=str(last_error),
+                )
                 raise last_error
         finally:
             reset_run_scope(scope_tokens)
@@ -233,6 +261,12 @@ async def _stream():
                     error=last_error,
                 )
             )
+            emit_failure_telemetry(
+                "provider_failed",
+                provider=route.primary_provider,
+                model=route.primary_model,
+                error=str(exc),
+            )
             for step in route.fallback_steps:
                 try:
                     fallback_result = await _execute_chain_step(
@@ -244,6 +278,12 @@ async def _stream():
                         attempt_log=attempt_log,
                         fallback_index=len(attempt_log),
                     )
+                    emit_failure_telemetry(
+                        "fallback_succeeded",
+                        provider=step.provider,
+                        model=step.model or step.label,
+                        primary_error=str(last_error),
+                    )
                     if isinstance(fallback_result, ResponseStream):
                         async for update in fallback_result:
                             yield update
@@ -274,6 +314,19 @@ async def _stream():
                             error=fallback_exc,
                         )
                     )
+                    emit_failure_telemetry(
+                        "fallback_step_failed",
+                        provider=step.provider,
+                        model=step.model or step.label,
+                        error=str(fallback_exc),
+                        fallback_index=len(attempt_log) - 1,
+                    )
+            emit_failure_telemetry(
+                "fallback_exhausted",
+                primary_provider=route.primary_provider,
+                attempted_providers=[attempt.provider for attempt in attempt_log],
+                final_error=str(last_error),
+            )
             raise last_error
 
         if isinstance(original_stream, ResponseStream):
@@ -791,6 +844,9 @@ def _run_subprocess(
         raise RuntimeError(f"{provider_name} failed: {detail}")
 
     output = completed.stdout.strip()
+    output_bytes = len(output.encode("utf-8", errors="replace"))
+    if output_bytes > MAX_CLI_OUTPUT_BYTES:
+        raise RuntimeError(f"{provider_name} output exceeds {MAX_CLI_OUTPUT_BYTES} bytes")
     if not output:
         raise RuntimeError(f"{provider_name} returned empty output")
 
@@ -818,7 +874,10 @@ def _messages_to_prompt(messages: list[Message] | tuple[Message, ...]) -> str:
     for message in messages:
         rendered.append(f"{str(message.role).upper()}: {_message_text(message)}")
     rendered.append("ASSISTANT:")
-    return "\n\n".join(rendered)
+    prompt = "\n\n".join(rendered)
+    if len(prompt.encode("utf-8", errors="replace")) > MAX_CLI_PROMPT_BYTES:
+        raise ValueError(f"Rendered CLI prompt exceeds {MAX_CLI_PROMPT_BYTES} bytes")
+    return prompt
 
 
 def _message_text(message: Message) -> str:

maf_starter/telemetry.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+import json
+import sys
+from datetime import datetime, timezone
+from typing import Any
+
+
+def emit_failure_telemetry(event: str, **fields: Any) -> None:
+    payload = {
+        "event": event,
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+        **fields,
+    }
+    try:
+        sys.stderr.write(json.dumps(payload, separators=(",", ":")) + "\n")
+        sys.stderr.flush()
+    except Exception:  # pragma: no cover - telemetry must never mask the original failure
+        return

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_provider_fallback_telemetry.py

@@ -0,0 +1,179 @@
+from __future__ import annotations
+
+import contextlib
+import io
+import json
+import subprocess
+import unittest
+import uuid
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from agent_framework import ChatResponse, Message
+
+from maf_starter.config import Settings, load_settings
+from maf_starter.provider_fallback import (
+    _messages_to_prompt,
+    _run_subprocess,
+    build_fallback_middleware,
+)
+from maf_starter.routing_policy import RoutingPlan
+from maf_starter.routing_types import ChainStep
+from maf_starter.telemetry import emit_failure_telemetry
+
+
+SCRATCH_ROOT = Path(__file__).resolve().parents[1] / ".tmp-tests"
+
+
+class RepoScratchTestCase(unittest.TestCase):
+    def make_scratch_dir(self) -> Path:
+        path = SCRATCH_ROOT / uuid.uuid4().hex
+        path.mkdir(parents=True, exist_ok=False)
+        self.addCleanup(lambda: path.exists() and __import__("shutil").rmtree(path, ignore_errors=True))
+        return path
+
+
+class ProviderFallbackTelemetryTests(RepoScratchTestCase):
+    def _make_settings(self, root: Path) -> Settings:
+        entities = root / "entities"
+        repo = root / "repo"
+        entities.mkdir()
+        repo.mkdir()
+        env = {
+            "MAF_API_KEY": "test-key",
+            "MAF_REPO_ROOT": str(repo),
+            "MAF_ENTITIES_DIR": str(entities),
+            "MAF_FALLBACK_CHAIN": "gemini-cli:gemini-2.5-pro,claude-cli:claude-sonnet-4-6",
+        }
+        with patch.dict("os.environ", env, clear=False):
+            return load_settings(project_root=root, env_path=root / ".missing-env")
+
+    @staticmethod
+    def _make_route_plan() -> RoutingPlan:
+        return RoutingPlan(
+            mode="auto",
+            route_lane="auto",
+            tier="standard",
+            rationale="test route",
+            primary_provider="gemini",
+            primary_model="gemini-2.5-pro",
+            requested_provider="gemini",
+            requested_model="gemini-2.5-pro",
+            fallback_steps=(
+                ChainStep("gemini-cli", "gemini-2.5-pro"),
+                ChainStep("claude-cli", "claude-sonnet-4-6"),
+            ),
+        )
+
+    @staticmethod
+    def _make_context() -> SimpleNamespace:
+        return SimpleNamespace(
+            messages=[Message(role="user", text="hello fallback")],
+            options={},
+            kwargs={},
+            metadata={},
+            result=None,
+        )
+
+    def test_emit_failure_telemetry_writes_single_parseable_json_line(self) -> None:
+        stream = io.StringIO()
+        with contextlib.redirect_stderr(stream):
+            emit_failure_telemetry("provider_failed", provider="gemini", error="quota exceeded")
+
+        lines = stream.getvalue().strip().splitlines()
+        self.assertEqual(len(lines), 1)
+        payload = json.loads(lines[0])
+        self.assertEqual(payload["event"], "provider_failed")
+        self.assertEqual(payload["provider"], "gemini")
+        self.assertEqual(payload["error"], "quota exceeded")
+        self.assertIn("timestamp", payload)
+
+    def test_emit_failure_telemetry_swallows_stderr_write_failures(self) -> None:
+        fake_stderr = SimpleNamespace(
+            write=lambda _: (_ for _ in ()).throw(RuntimeError("stderr failed")),
+            flush=lambda: None,
+        )
+        with patch("sys.stderr", fake_stderr):
+            self.assertIsNone(emit_failure_telemetry("provider_failed", provider="gemini", error="boom"))
+
+    def test_fallback_middleware_emits_primary_step_and_exhausted_events(self) -> None:
+        root = self.make_scratch_dir()
+        settings = self._make_settings(root)
+        middleware = build_fallback_middleware(
+            settings,
+            primary_provider="gemini",
+            primary_model="gemini-2.5-pro",
+            routing_mode="auto",
+        )
+        context = self._make_context()
+        stderr = io.StringIO()
+
+        async def call_next():
+            raise RuntimeError("rate limit from primary")
+
+        attempts = [
+            RuntimeError("rate limit in gemini-cli fallback"),
+            RuntimeError("quota exhausted in claude fallback"),
+        ]
+
+        async def fake_execute_chain_step(**_kwargs):
+            raise attempts.pop(0)
+
+        with (
+            patch("maf_starter.provider_fallback.build_routing_plan", return_value=self._make_route_plan()),
+            patch("maf_starter.provider_fallback._execute_chain_step", side_effect=fake_execute_chain_step),
+            contextlib.redirect_stderr(stderr),
+        ):
+            with self.assertRaisesRegex(RuntimeError, "quota exhausted in claude fallback"):
+                __import__("asyncio").run(middleware(context, call_next))
+
+        events = [json.loads(line)["event"] for line in stderr.getvalue().strip().splitlines()]
+        self.assertEqual(events, ["provider_failed", "fallback_step_failed", "fallback_step_failed", "fallback_exhausted"])
+
+    def test_fallback_middleware_emits_recovery_event_when_fallback_succeeds(self) -> None:
+        root = self.make_scratch_dir()
+        settings = self._make_settings(root)
+        middleware = build_fallback_middleware(
+            settings,
+            primary_provider="gemini",
+            primary_model="gemini-2.5-pro",
+            routing_mode="auto",
+        )
+        context = self._make_context()
+        stderr = io.StringIO()
+
+        async def call_next():
+            raise RuntimeError("rate limit from primary")
+
+        async def fake_execute_chain_step(**_kwargs):
+            return ChatResponse(messages=[Message(role="assistant", text="fallback ok")])
+
+        with (
+            patch("maf_starter.provider_fallback.build_routing_plan", return_value=self._make_route_plan()),
+            patch("maf_starter.provider_fallback._execute_chain_step", side_effect=fake_execute_chain_step),
+            contextlib.redirect_stderr(stderr),
+        ):
+            __import__("asyncio").run(middleware(context, call_next))
+
+        payloads = [json.loads(line) for line in stderr.getvalue().strip().splitlines()]
+        self.assertEqual(payloads[0]["event"], "provider_failed")
+        self.assertEqual(payloads[1]["event"], "fallback_succeeded")
+        self.assertNotIn("fallback_exhausted", [payload["event"] for payload in payloads])
+        self.assertIsInstance(context.result, ChatResponse)
+
+    def test_run_subprocess_rejects_oversized_output(self) -> None:
+        completed = subprocess.CompletedProcess(args=["codex.cmd"], returncode=0, stdout="x" * 32, stderr="")
+        with (
+            patch("maf_starter.provider_fallback.subprocess.run", return_value=completed),
+            patch("maf_starter.provider_fallback.shutil.which", return_value="codex.cmd"),
+            patch("maf_starter.provider_fallback.MAX_CLI_OUTPUT_BYTES", 8),
+        ):
+            with self.assertRaisesRegex(RuntimeError, "output exceeds 8 bytes"):
+                _run_subprocess("codex-cli", ["codex.cmd", "exec"], Path.cwd(), None)
+
+    def test_messages_to_prompt_rejects_oversized_prompt(self) -> None:
+        messages = [Message(role="user", text="x" * 32)]
+        with patch("maf_starter.provider_fallback.MAX_CLI_PROMPT_BYTES", 8):
+            with self.assertRaisesRegex(ValueError, "Rendered CLI prompt exceeds 8 bytes"):
+                _messages_to_prompt(messages)

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()