feat(28-02): emit structured JSON failure telemetry in provider fallback and cap CLI subprocess size

- Add maf_starter/telemetry.py with emit_failure_telemetry(event, **fields),
  a stdlib-only, non-raising helper that writes one line of parseable JSON to
  stderr so failures are observable to log-based monitoring, not just
  in-memory response metadata.
- Wire emit_failure_telemetry into provider_fallback.py's fallback_middleware
  and _wrap_stream_with_fallback at four points: primary-provider failure,
  each fallback step failure, fallback success/recovery, and full-chain
  exhaustion, in both the non-streaming and streaming code paths.
- Add MAX_CLI_OUTPUT_BYTES and MAX_CLI_PROMPT_BYTES (1MB, mirroring
  loop_worker_cli.py's existing MAX_REQUEST_BYTES convention) and enforce
  them in _run_subprocess and _messages_to_prompt, closing the remaining C6
  gap on the CLI-fallback subprocess path.
- Add tests/test_provider_fallback_telemetry.py: parseable single-line JSON
  contract, non-raising stderr-write-failure behavior, ordered telemetry
  events for chain exhaustion vs. fallback recovery, and size-limit
  rejection tests for both new constants.

Author: Aitomates

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

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)