"""Offline teaching fixture: contracts, bounded retries and structured logging. Python 3.12 / Pydantic 2.13.5. No LLM, database or network credentials required. This is a companion exercise, not the author's historical Agent application. """ import asyncio import json import logging import math import random import sys import time import uuid from dataclasses import dataclass from logging.handlers import RotatingFileHandler from pathlib import Path from typing import Awaitable, Callable, Literal, Optional, Union from pydantic import BaseModel, ConfigDict, Field, ValidationError class MultiplyInput(BaseModel): model_config = ConfigDict(strict=True, extra="forbid") a: int = Field(ge=-10000, le=10000) b: int = Field(ge=-10000, le=10000) class MultiplyOutput(BaseModel): model_config = ConfigDict(strict=True, extra="forbid") value: int class Success(BaseModel): model_config = ConfigDict(extra="forbid") success: Literal[True] = True data: MultiplyOutput error: None = None class Failure(BaseModel): model_config = ConfigDict(extra="forbid") success: Literal[False] = False data: None = None error: Literal[ "unknown_tool", "input_validation", "output_validation", "business", "timeout", "connection", "internal" ] class BusinessError(Exception): pass @dataclass(frozen=True) class Tool: call: Callable[[MultiplyInput], Awaitable[dict]] retry_safe: bool = False async def multiply(args: MultiplyInput) -> dict: return {"value": args.a * args.b} TOOLS = {"multiply": Tool(multiply, retry_safe=True)} def tool_schema() -> dict: """A description for a future model adapter, not an execution permission.""" return { "name": "multiply", "description": "Multiply two bounded integers", "parameters": MultiplyInput.model_json_schema(), } class EventFormatter(logging.Formatter): def format(self, record: logging.LogRecord) -> str: # Allowlist only. No raw arguments, messages, exception text or tokens. payload = {"event": "tool_attempt", "level": record.levelname} for name in ( "run_id", "tool_call_id", "step", "tool_name", "attempt", "success", "error_type", "duration_ms" ): payload[name] = getattr(record, name, None) return json.dumps(payload, ensure_ascii=False) def make_logger(log_path: Optional[Path] = None, stream=None, max_bytes: int = 1000000, backup_count: int = 2) -> logging.Logger: # Independent logger: importing or invoking twice does not accumulate handlers. logger = logging.Logger("agent-exercise", level=logging.DEBUG) logger.propagate = False console = logging.StreamHandler(stream if stream is not None else sys.stdout) console.setLevel(logging.DEBUG) console.setFormatter(EventFormatter()) logger.addHandler(console) if log_path is not None: if max_bytes < 1 or backup_count < 1: raise ValueError("rotation limits must be positive") file_handler = RotatingFileHandler( log_path, maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8" ) file_handler.setLevel(logging.INFO) file_handler.setFormatter(EventFormatter()) logger.addHandler(file_handler) return logger def close_logger(logger: logging.Logger) -> None: for handler in logger.handlers[:]: handler.close() logger.removeHandler(handler) async def execute( name: str, payload: dict, *, logger: logging.Logger, run_id: str, tool_call_id: str, step: int = 1, registry=None, timeout_s: float = 0.5, max_attempts: int = 3, base_delay: float = 0.05, cap_delay: float = 0.5, sleep=asyncio.sleep, rng=random.random, ) -> Union[Success, Failure]: """Bounded *per-attempt* timeout; max_attempts includes the initial call. The tool must be cooperative async code. wait_for is not a hard kill switch. Caller cancellation propagates. No authorization or global run deadline here. """ if type(max_attempts) is not int or not 1 <= max_attempts <= 5: raise ValueError("max_attempts must be an integer in 1..5") if not math.isfinite(timeout_s) or timeout_s <= 0: raise ValueError("timeout_s must be finite and positive") if not all(math.isfinite(x) and x >= 0 for x in (base_delay, cap_delay)): raise ValueError("delays must be finite and nonnegative") registry = TOOLS if registry is None else registry tool = registry.get(name) def emit(attempt, started, error): logger.info("tool_attempt", extra={ "run_id": run_id, "tool_call_id": tool_call_id, "step": step, "tool_name": name if tool is not None else "unknown", "attempt": attempt, "success": error is None, "error_type": error, "duration_ms": round((time.perf_counter() - started) * 1000, 3), }) started = time.perf_counter() if tool is None: emit(0, started, "unknown_tool") return Failure(error="unknown_tool") try: args = MultiplyInput.model_validate(payload) except ValidationError: emit(0, started, "input_validation") return Failure(error="input_validation") for attempt in range(1, max_attempts + 1): started = time.perf_counter() try: raw = await asyncio.wait_for(tool.call(args), timeout=timeout_s) except asyncio.CancelledError: emit(attempt, started, "cancelled") raise except (TimeoutError, asyncio.TimeoutError): error = "timeout" except ConnectionError: error = "connection" except BusinessError: error = "business" except Exception: error = "internal" else: try: data = MultiplyOutput.model_validate(raw) except ValidationError: emit(attempt, started, "output_validation") return Failure(error="output_validation") emit(attempt, started, None) return Success(data=data) emit(attempt, started, error) if error not in ("timeout", "connection") or not tool.retry_safe or attempt == max_attempts: return Failure(error=error) # Full jitter, bounded attempts and delay cap. The injected RNG aids testing. delay = min(cap_delay, base_delay * (2 ** (attempt - 1))) * rng() await sleep(delay) raise AssertionError("unreachable") async def demo() -> None: logger = make_logger() run_id = uuid.uuid4().hex try: for step, payload in enumerate([{"a": 6, "b": 7}, {"a": "6", "b": 7}], 1): result = await execute( "multiply", payload, logger=logger, run_id=run_id, tool_call_id=uuid.uuid4().hex, step=step ) print(result.model_dump_json()) finally: close_logger(logger) if __name__ == "__main__": asyncio.run(demo())