# SPDX-License-Identifier: MIT """Deterministic agent workflow used before the Langfuse adapter is attached.""" from __future__ import annotations from dataclasses import dataclass, field import re from typing import Dict, List, Literal, Mapping, Optional, Protocol, Tuple EventPhase = Literal["start", "end"] ObservationKind = Literal["span", "generation", "tool"] EventStatus = Literal["unset", "ok", "warning", "error"] EMAIL_PATTERN = re.compile(r"[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}", re.I) ACCOUNT_PATTERN = re.compile(r"\bacct_[A-Z0-9]+\b", re.I) @dataclass(frozen=True) class TraceEvent: """One deterministic start or end event for a logical observation.""" phase: EventPhase observation_id: str name: str parent_id: str kind: ObservationKind status: EventStatus attempt: int = 1 attributes: Tuple[Tuple[str, str], ...] = () def to_dict(self) -> Dict[str, object]: """Return a JSON-serializable representation with stable key ordering.""" return { "phase": self.phase, "observation_id": self.observation_id, "name": self.name, "parent_id": self.parent_id, "kind": self.kind, "status": self.status, "attempt": self.attempt, "attributes": dict(self.attributes), } class TraceSink(Protocol): """Receive trace events without controlling application behavior.""" def record(self, event: TraceEvent) -> None: """Record one event or raise an exporter-specific exception.""" @dataclass class MemoryTraceSink: """Collect deterministic events for assertions and adapter development.""" events: List[TraceEvent] = field(default_factory=list) def record(self, event: TraceEvent) -> None: self.events.append(event) class FailingTraceSink: """Simulate an unavailable telemetry exporter.""" def record(self, event: TraceEvent) -> None: del event raise ConnectionError("fixture exporter unavailable") @dataclass class SafeTraceSink: """Keep telemetry failures outside the application result path.""" delegate: TraceSink dropped_events: int = 0 def record(self, event: TraceEvent) -> None: try: self.delegate.record(event) except Exception: # Telemetry is deliberately isolated from the application result path. self.dropped_events += 1 @dataclass(frozen=True) class SupportRequest: """Synthetic support input; never use real customer identifiers in fixtures.""" question: str email: str account_id: str @dataclass(frozen=True) class WorkflowResult: """Application result returned independently of telemetry availability.""" answer: str needs_human_review: bool class TerminalToolError(RuntimeError): """A fixture tool failure that requires a human decision.""" @dataclass class FixtureTools: """Deterministic tool responses for happy, retry, and terminal cases.""" subscription_failures_remaining: int = 0 refund_terminal_error: bool = False def get_subscription(self, account_id: str) -> str: del account_id if self.subscription_failures_remaining > 0: self.subscription_failures_remaining -= 1 raise TimeoutError("subscription fixture timed out") return "pro" def get_refund_window(self, account_id: str) -> str: del account_id if self.refund_terminal_error: raise TerminalToolError("refund fixture requires manual review") return "14 days" def redact_text(value: str) -> str: """Redact the two synthetic identifier shapes used by this fixture.""" value = EMAIL_PATTERN.sub("[EMAIL_REDACTED]", value) return ACCOUNT_PATTERN.sub("[ACCOUNT_REDACTED]", value) def _attributes(values: Mapping[str, object]) -> Tuple[Tuple[str, str], ...]: return tuple(sorted((key, redact_text(str(value))) for key, value in values.items())) def _record( sink: SafeTraceSink, phase: EventPhase, observation_id: str, name: str, parent_id: str, kind: ObservationKind, status: EventStatus, attempt: int = 1, attributes: Optional[Mapping[str, object]] = None, ) -> None: sink.record( TraceEvent( phase=phase, observation_id=observation_id, name=name, parent_id=parent_id, kind=kind, status=status, attempt=attempt, attributes=_attributes(attributes or {}), ) ) def run_support_workflow( request: SupportRequest, tools: FixtureTools, trace_sink: TraceSink, ) -> WorkflowResult: """Run the fixed workflow while treating tracing as a noncritical side effect.""" sink = SafeTraceSink(trace_sink) root_id = "task-1" _record( sink, "start", root_id, "answer-support-question", "", "span", "unset", attributes={ "question": request.question, "email": request.email, "account_id": request.account_id, "environment": "fixture", }, ) _record(sink, "start", "retrieval-1", "retrieve-account-policy", root_id, "span", "unset") _record( sink, "end", "retrieval-1", "retrieve-account-policy", root_id, "span", "ok", attributes={"document": "refund-policy-v1"}, ) _record(sink, "start", "plan-1", "draft-plan", root_id, "generation", "unset") _record( sink, "end", "plan-1", "draft-plan", root_id, "generation", "ok", attributes={"fixture_response": "check subscription and refund window"}, ) subscription = "unknown" for attempt in (1, 2): observation_id = f"subscription-{attempt}" _record( sink, "start", observation_id, "tool:get-subscription", root_id, "tool", "unset", attempt=attempt, attributes={"account_id": request.account_id}, ) try: subscription = tools.get_subscription(request.account_id) except TimeoutError as error: _record( sink, "end", observation_id, "tool:get-subscription", root_id, "tool", "error", attempt=attempt, attributes={"error_type": type(error).__name__}, ) if attempt == 2: raise continue _record( sink, "end", observation_id, "tool:get-subscription", root_id, "tool", "ok", attempt=attempt, attributes={"subscription": subscription}, ) break needs_human_review = False refund_window = "unknown" _record(sink, "start", "refund-1", "tool:get-refund-window", root_id, "tool", "unset") try: refund_window = tools.get_refund_window(request.account_id) except TerminalToolError as error: needs_human_review = True _record( sink, "end", "refund-1", "tool:get-refund-window", root_id, "tool", "error", attributes={"error_type": type(error).__name__}, ) else: _record( sink, "end", "refund-1", "tool:get-refund-window", root_id, "tool", "ok", attributes={"refund_window": refund_window}, ) _record(sink, "start", "answer-1", "compose-answer", root_id, "generation", "unset") if needs_human_review: answer = f"Subscription: {subscription}. A human must confirm refund eligibility." else: answer = f"Subscription: {subscription}. Refund window: {refund_window}." _record( sink, "end", "answer-1", "compose-answer", root_id, "generation", "ok", attributes={"needs_human_review": needs_human_review}, ) _record( sink, "end", root_id, "answer-support-question", "", "span", "warning" if needs_human_review else "ok", attributes={"needs_human_review": needs_human_review}, ) return WorkflowResult(answer=answer, needs_human_review=needs_human_review)