From bfbec1bb249a154f85a0d9e44174a8750cc78460 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kim=20Harjam=C3=A4ki?= Date: Tue, 7 Jul 2026 18:37:44 +0300 Subject: [PATCH 1/2] 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. --- maf_starter/provider_fallback.py | 61 +++++++- maf_starter/telemetry.py | 19 +++ tests/test_provider_fallback_telemetry.py | 179 ++++++++++++++++++++++ 3 files changed, 258 insertions(+), 1 deletion(-) create mode 100644 maf_starter/telemetry.py create mode 100644 tests/test_provider_fallback_telemetry.py diff --git a/maf_starter/provider_fallback.py b/maf_starter/provider_fallback.py index 758b1ec..7d31929 100644 --- a/maf_starter/provider_fallback.py +++ b/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: diff --git a/maf_starter/telemetry.py b/maf_starter/telemetry.py new file mode 100644 index 0000000..32fb740 --- /dev/null +++ b/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 diff --git a/tests/test_provider_fallback_telemetry.py b/tests/test_provider_fallback_telemetry.py new file mode 100644 index 0000000..e38f88d --- /dev/null +++ b/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) From 6208345c01721c58a6644007cd276c81adbbbd9f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kim=20Harjam=C3=A4ki?= Date: Tue, 7 Jul 2026 18:37:58 +0300 Subject: [PATCH 2/2] 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. --- maf_starter/worker_boundary.py | 3 +++ tests/test_worker_boundary.py | 15 ++++++++++++--- 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/maf_starter/worker_boundary.py b/maf_starter/worker_boundary.py index 778bb91..1717618 100644 --- a/maf_starter/worker_boundary.py +++ b/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.""" diff --git a/tests/test_worker_boundary.py b/tests/test_worker_boundary.py index d0abe6d..5b251da 100644 --- a/tests/test_worker_boundary.py +++ b/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()