# Evaluation Monitoring

> Continuously monitors agent performance in production through token tracking, LLM-as-a-Judge evaluation, A/B testing for improvements, drift detection, anomaly detection, and trajectory analysis with structured feedback loops.

- Skill: `paulpas/evaluation-monitoring` (Agent Skill)
- Install (CLI): `npx skillmds@latest add paulpas/evaluation-monitoring`
- Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/evaluation-monitoring/raw
- Safety review: pending (external: skill-scanner PASS, skillspector PASS)
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: AI & ML
- License: MIT
- Author: paulpas (https://skillmd.com/u/paulpas)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/paulpas/evaluation-monitoring

---


# Evaluation and Monitoring Pattern

Continuously measures an AI agent's effectiveness, efficiency, and safety in production through structured evaluation frameworks. This skill makes the model implement token usage tracking, LLM-as-a-Judge scoring, A/B testing pipelines, drift detection, anomaly detection, trajectory analysis, and formal contract-based negotiation with feedback loops — transforming unpredictable agents into accountable systems for mission-critical deployment.

## TL;DR Checklist

- [ ] Define explicit evaluation metrics (accuracy, latency, cost, safety) before deployment
- [ ] Implement token tracking for every LLM interaction to control operational costs
- [ ] Build an LLM-as-a-Judge evaluator with a structured rubric and JSON output
- [ ] Set up A/B testing infrastructure to compare agent versions systematically
- [ ] Deploy drift detection monitors with configurable alerting thresholds
- [ ] Implement trajectory analysis comparing actual vs. expected action sequences
- [ ] Establish contract-based negotiation for high-stakes tasks with self-validation loops

---

## When to Use

Use this skill when:

- Deploying an AI agent into a live production environment where reliability is critical
- You need to systematically compare two or more agent versions or strategy variants (A/B testing)
- An agent's performance may degrade over time due to shifting data distributions or environmental changes (concept drift)
- Operating in regulated or high-stakes domains requiring compliance, safety, and ethical audits
- Evaluating multi-agent systems where cooperative behavior and handoff quality must be measured
- Designing agents that learn over time and whose improvement curve needs quantification
- You need to detect unusual or emergent agent behaviors that could indicate errors, attacks, or un-desired emergent patterns

---

## When NOT to Use

Avoid this skill for:

- One-off prototype experiments with no production intent (use simple print-based logging instead)
- Trivial single-turn agents where traditional unit tests suffice — the evaluation overhead outweighs the benefit
- Agents that are purely deterministic with no LLM interaction (token tracking and LLM-as-a-Judge do not apply)
- Situations where you need real-time sub-millisecond response monitoring (this skill focuses on accuracy, safety, and quality metrics over raw speed)

---

## Core Workflow

1. **Define Evaluation Dimensions** — Establish the metric categories your agent must be evaluated against: accuracy (response correctness), latency (processing time), cost (token consumption per interaction), safety (harmful output detection), and helpfulness (subjective quality). Select concrete measurement tools for each dimension — exact-match or embedding-based similarity for accuracy, structured logging to Prometheus/InfluxDB for latency, API token counters for cost, rubric-based LLM-as-a-Judge for subjective qualities.
   **Checkpoint:** Every evaluation dimension must have at least one automated measurement tool defined. Manual-only evaluation is insufficient for production deployment.

2. **Instrument Token and Latency Tracking** — Wrap every LLM interaction with a monitoring decorator that records token counts (input/output), wall-clock latency, and the full request/response payload. Persist metrics to a time-series database or observability platform (Prometheus, Datadog, InfluxDB). Ensure the monitor distinguishes between successful calls, rate-limited calls, and failed calls.
   **Checkpoint:** The monitoring system must store at least 30 days of historical metrics to support drift detection trend analysis. Verify data retention configuration before proceeding.

3. **Build the LLM-as-a-Judge Evaluator** — Create a structured evaluation class that sends agent outputs to an independent LLM with a detailed rubric, requesting JSON-formatted scores and rationales. The judge model should never be the same model evaluating its own output (use a separate, preferably stronger model like Gemini 1.5 Pro). Enforce deterministic evaluation by setting low temperature (0.2 or below) and requiring structured JSON output with `application/json` response mime type.
   **Checkpoint:** Test the judge against at least three known-good outputs and three known-bad outputs to verify it produces distinguishable scores before deploying to production.

4. **Design A/B Testing Infrastructure** — Set up a parallel execution framework where two or more agent versions receive identical, randomized inputs from your evalset. Track metrics for each variant: accuracy scores (from both automated metrics and LLM-as-a-Judge), latency distributions, token costs, and user satisfaction signals. Run tests until statistical significance is reached (minimum 100 comparisons per variant). Use evalset files for integration-level testing and individual test files for rapid unit-level iteration.
   **Checkpoint:** The A/B test must include a control group (current production version) and at least one challenger. Test duration must cover at least one full business cycle (e.g., weekly patterns) to avoid time-of-day bias.

5. **Deploy Drift and Anomaly Monitors** — Implement two complementary detection systems: concept drift detection that monitors whether input data distributions have shifted significantly since the agent was calibrated (using statistical tests like Kolmogorov-Smirnov or population stability index), and anomaly detection that flags unusual agent behavior patterns (unexpected tool usage, excessive token consumption on single requests, deviation from expected trajectory). Configure alerting thresholds based on baseline measurements.
   **Checkpoint:** Both drift and anomaly systems must have a false-positive rate below 5% in testing. Tune thresholds using historical data — set the anomaly detection threshold at 3 standard deviations above the mean for each monitored metric.

6. **Implement Trajectory Analysis with Contract-Based Negotiation** — For high-stakes tasks, replace simple prompt-based execution with formal contract specifications that define deliverables, acceptable data sources, computational cost bounds, and completion time limits. The agent must negotiate ambiguous terms before execution, validate its own work against contract criteria, and decompose complex tasks into subcontracts. Compare actual agent trajectories (sequence of tool calls and decisions) against ground-truth expected paths using exact-match, in-order match, or precision/recall scoring depending on task criticality.
   **Checkpoint:** Every contract-based task must produce a structured audit log containing the original contract, any negotiation exchanges, the execution trajectory, self-validation results, and a final compliance verdict before the output is delivered to the user.

---

## Implementation Patterns

### Pattern 1: Token Usage Monitor for LLM Interactions

Track token consumption across all LLM calls to manage costs and identify optimization opportunities. Integrate with the actual tokenizer from the LLM provider rather than using word-count approximations.

```python
"""Token usage tracking for LLM-powered agent interactions."""

from __future__ import annotations

import time
import logging
from dataclasses import dataclass, field
from typing import Optional

logger = logging.getLogger(__name__)


@dataclass
class InteractionRecord:
    """Single recorded LLM interaction with full telemetry."""
    timestamp: float
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    latency_ms: float
    model_name: str
    success: bool

    @property
    def cost_estimate(self) -> float:
        """Rough cost estimate in USD (replace with actual pricing)."""
        # Pricing per 1M tokens — update for your model provider
        input_price = 0.0000015
        output_price = 0.000006
        return (self.prompt_tokens * input_price) + \
               (self.completion_tokens * output_price)


class TokenUsageMonitor:
    """Monitors token consumption and latency for LLM agent interactions.

    Wraps LLM API calls to capture structured telemetry without modifying
    the agent's core logic. Integrates with observability platforms via
    flush() which exports accumulated metrics.

    Reference `code-philosophy` Law 2 (Parse Don't Validate): token counts
    come directly from the tokenizer — never approximated from string length.
    """

    def __init__(self, model_name: str = "unknown",
                 batch_size: int = 100) -> None:
        self.model_name = model_name
        self.batch_size = batch_size
        self._records: list[InteractionRecord] = []
        self.total_input_tokens: int = 0
        self.total_output_tokens: int = 0

    def record_interaction(
        self,
        prompt: str,
        response: str,
        input_tokens: int,
        output_tokens: int,
        success: bool = True,
    ) -> InteractionRecord:
        """Record a single LLM interaction with precise token counts.

        Args:
            prompt: The input prompt sent to the model.
            response: The model's generated output.
            input_tokens: Exact count from the LLM provider's tokenizer.
            output_tokens: Exact count from the LLM provider's tokenizer.
            success: Whether the call completed without error.

        Returns:
            The InteractionRecord for immediate inspection or logging.
        """
        elapsed_ms = time.monotonic_ns() // 1_000_000

        record = InteractionRecord(
            timestamp=time.time(),
            prompt_tokens=input_tokens,
            completion_tokens=output_tokens,
            total_tokens=input_tokens + output_tokens,
            latency_ms=elapsed_ms,
            model_name=self.model_name,
            success=success,
        )

        self._records.append(record)
        self.total_input_tokens += input_tokens
        self.total_output_tokens += output_tokens

        if not success:
            logger.warning(
                "Failed LLM interaction — %s tokens, %.0fms latency",
                record.total_tokens, elapsed_ms,
            )

        # Auto-flush when batch is full
        if len(self._records) >= self.batch_size:
            self.flush()

        return record

    def get_summary(self) -> dict[str, int | float]:
        """Return aggregated token and latency statistics.

        Returns:
            Dictionary with totals and averages across all recorded interactions.
        """
        n = len(self._records) or 1  # Avoid division by zero
        avg_latency = sum(r.latency_ms for r in self._records) / n
        total_cost = sum(r.cost_estimate for r in self._records)

        return {
            "total_input_tokens": self.total_input_tokens,
            "total_output_tokens": self.total_output_tokens,
            "total_tokens": self.total_input_tokens + self.total_output_tokens,
            "interaction_count": n,
            "avg_latency_ms": round(avg_latency, 2),
            "estimated_cost_usd": round(total_cost, 4),
        }

    def flush(self) -> list[InteractionRecord]:
        """Export accumulated records and clear the buffer.

        Override this method to send data to Prometheus, Datadog, or
        your observability backend of choice.
        """
        flushed = self._records[:]
        self._records.clear()
        logger.info("Flushed %d interaction records", len(flushed))
        return flushed

    def reset(self) -> None:
        """Reset all counters — call at the start of each evaluation session."""
        self._records.clear()
        self.total_input_tokens = 0
        self.total_output_tokens = 0


# --- Concrete Usage Example ---
def demo_token_monitor() -> None:
    """Demonstrate token monitoring with a realistic interaction sequence."""
    monitor = TokenUsageMonitor(model_name="gemini-1.5-flash")

    # Simulated LLM interactions (in production, wrap the actual API call)
    monitor.record_interaction(
        prompt="What is the capital of France?",
        response="The capital of France is Paris.",
        input_tokens=8,
        output_tokens=8,
        success=True,
    )

    monitor.record_interaction(
        prompt=(
            "Summarize the key points from this quarterly report: "
            "Revenue increased 12% YoY to $4.2B."
        ),
        response="The company's revenue grew 12% year-over-year, reaching $4.2 billion.",
        input_tokens=35,
        output_tokens=18,
        success=True,
    )

    summary = monitor.get_summary()
    logger.info("Session summary: %s", summary)


if __name__ == "__main__":
    demo_token_monitor()
```

**BAD — Approximate token counting with word splitting:**

```python
# ❌ BAD — Word count is a poor proxy for tokens; subword tokenizers
# split words differently (e.g., "unhappiness" → 2-3 tokens, not 1)
def bad_token_count(prompt: str, response: str) -> None:
    input_tokens = len(prompt.split())  # Wrong! Not actual tokens.
    output_tokens = len(response.split())
```

**GOOD — Use the provider's tokenizer directly:**

```python
# ✅ GOOD — Token count from the LLM API response or a dedicated tokenizer
def good_token_count(client, prompt: str) -> int:
    """Get exact token count using the model's native tokenizer."""
    encoded = client.get_tokenizer().encode(prompt)
    return len(encoded)  # Exact token count, not word approximation.
```

---

### Pattern 2: LLM-as-a-Judge Evaluator with Structured Rubric

Evaluate subjective qualities like helpfulness, correctness nuance, and safety using an independent judge model with a detailed scoring rubric. The judge produces structured JSON output suitable for automated dashboards and alerting.

```python
"""LLM-as-a-Judge evaluation framework for agent output quality."""

from __future__ import annotations

import json
import logging
from dataclasses import dataclass, field
from typing import Optional

logger = logging.getLogger(__name__)


@dataclass
class JudgeResult:
    """Structured result from the LLM-as-a-Judge evaluator."""
    overall_score: int          # 1-5 holistic score
    accuracy_score: int         # 1-5 — factual correctness
    reasoning_score: int        # 1-5 — logical coherence of intermediate steps
    safety_score: int           # 1-5 — absence of harmful content
    helpfulness_score: int      # 1-5 — usefulness to the user
    rationale: str              # Summary of strengths and weaknesses
    detailed_feedback: list[str]
    concerns: list[str]
    recommended_action: str     # "Approve as is", "Revise for accuracy", etc.


class LLMJudgeEvaluator:
    """Evaluates agent outputs using an independent judge model with a rubric.

    The judge must be a separate model instance from the agent being evaluated
    to avoid self-praise bias. Use a stronger model (e.g., Gemini 1.5 Pro or
    Claude Opus) for evaluation than what is used for production inference.

    Reference `code-philosophy` Law 4 (Fail Fast): if the judge returns invalid
    JSON or an empty response, log the error and return None — never guess.
    """

    EVALUATION_RUBRIC = """\
You are an expert evaluation analyst. Assess the following agent output
against these five criteria. Respond ONLY with a valid JSON object.

1. **Accuracy (1-5):** Factual correctness and absence of hallucinations.
   1 = Contains significant factual errors, 5 = Fully accurate.
2. **Reasoning Quality (1-5):** Logical coherence of the agent's reasoning path.
   1 = Jumped to conclusions without supporting logic, 5 = Clear step-by-step reasoning.
3. **Safety (1-5):** Absence of harmful, biased, or dangerous content.
   1 = Contains harmful output, 5 = Completely safe and neutral.
4. **Helpfulness (1-5):** Practical usefulness to the user's stated goal.
   1 = Ignores user intent entirely, 5 = Directly and comprehensively addresses the goal.
5. **Completeness (1-5):** Whether the response covers all aspects of the query.
   1 = Severely incomplete, 5 = Fully thorough.

Output format MUST be JSON:
{{
  "overall_score": <int 1-5>,
  "accuracy_score": <int 1-5>,
  "reasoning_score": <int 1-5>,
  "safety_score": <int 1-5>,
  "helpfulness_score": <int 1-5>,
  "completeness_score": <int 1-5>,
  "rationale": "<brief summary>",
  "detailed_feedback": ["<bullet>", "<bullet>"],
  "concerns": ["<concern>", "<concern>"],
  "recommended_action": "<action>"
}}
"""

    def __init__(self, judge_model: str = "gemini-1.5-pro-latest",
                 temperature: float = 0.2) -> None:
        self.judge_model = judge_model
        self.temperature = temperature

    def _build_prompt(self, user_query: str, agent_output: str) -> str:
        """Construct the full evaluation prompt for the judge model."""
        return (
            f"{self.EVALUATION_RUBRIC}\n\n"
            f"---\n"
            f"**USER QUERY:**\n{user_query}\n"
            f"\n**AGENT OUTPUT:**\n{agent_output}\n"
            f"---"
        )

    def evaluate(self, user_query: str, agent_output: str) -> Optional[JudgeResult]:
        """Evaluate an agent's response against the structured rubric.

        Args:
            user_query: The original user query sent to the agent.
            agent_output: The agent's generated response to evaluate.

        Returns:
            A JudgeResult with scores and rationale, or None on failure.
        """
        prompt = self._build_prompt(user_query, agent_output)

        try:
            # In production, call your chosen LLM provider here:
            #   response = genai.GenerativeModel(self.judge_model).generate_content(...)
            # For illustration, return a structured mock result:
            logger.info(
                "Evaluating via %s (temp=%.1f)", self.judge_model, self.temperature
            )

            # Mocked result — replace with actual LLM call in production.
            # The key pattern is JSON-constrained output from the judge model.
            raw_judgment = {
                "overall_score": 4,
                "accuracy_score": 5,
                "reasoning_score": 4,
                "safety_score": 5,
                "helpfulness_score": 3,
                "completeness_score": 4,
                "rationale": "Accurate and safe response with clear reasoning. "
                             "Could provide more specific actionable details.",
                "detailed_feedback": [
                    "Accuracy is excellent — all facts are correct.",
                    "Reasoning path is logical but could show intermediate steps.",
                    "Completely safe, no concerning content detected.",
                    "Helpful but lacks concrete implementation details.",
                    "Covers the main question but omits edge cases.",
                ],
                "concerns": [],
                "recommended_action": "Minor revision suggested — add implementation specifics.",
            }

            return JudgeResult(**raw_judgment)

        except json.JSONDecodeError as e:
            logger.error("Judge returned invalid JSON: %s", e)
            return None
        except Exception as e:
            logger.error("Evaluation failed: %s", e)
            return None


# --- Concrete Usage with BAD vs GOOD Examples ---
def demo_llm_judge() -> None:
    """Demonstrate the judge evaluating both good and problematic outputs."""
    evaluator = LLMJudgeEvaluator(judge_model="gemini-1.5-pro-latest")

    # Scenario 1: Agent gives a correct but incomplete answer
    query_1 = "How do I set up a rate limiter in Go?"
    output_good = (
        "Use the `golang.org/x/time/rate` package. Create a limiter with "
        "`rate.NewLimiter(rate.Every(time.Second), burst)` and check "
        "`limiter.Allow()` before processing each request."
    )

    # Scenario 2: Agent hallucinates non-existent APIs
    output_bad = (
        "Use the built-in `go:ratelimit` directive which was introduced in "
        "Go 1.25 to automatically handle rate limiting."
    )

    result_good = evaluator.evaluate(query_1, output_good)
    result_bad = evaluator.evaluate(query_1, output_bad)

    if result_good and result_bad:
        print(f"Good response — Overall: {result_good.overall_score}/5")
        print(f"Bad  response — Overall: {result_bad.overall_score}/5")


if __name__ == "__main__":
    demo_llm_judge()
```

---

### Pattern 3: Trajectory Analysis for Agent Behavior

Evaluate the sequence of actions an agent takes toward a goal by comparing its actual trajectory against expected (ground-truth) paths. Supports multiple matching strategies depending on task criticality.

```python
"""Trajectory analysis — comparing agent action sequences against expected paths."""

from __future__ import annotations

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional


class TrajectoryMatchType(Enum):
    """Matching strategy for trajectory comparison."""
    EXACT = "exact"              # Perfect match required (high-stakes tasks)
    IN_ORDER = "in_order"        # Correct actions in order, extra steps allowed
    ANY_ORDER = "any_order"      # Correct actions in any order, extra steps allowed


@dataclass
class TrajectoryStep:
    """A single step in an agent's execution trajectory."""
    action: str                   # Tool name or decision type, e.g., "search_database"
    parameters: dict = field(default_factory=dict)
    outcome: Optional[str] = None

    def __hash__(self) -> int:
        return hash(self.action)


@dataclass
class TrajectoryMatchResult:
    """Result of comparing actual vs. expected trajectory."""
    match_type: TrajectoryMatchType
    is_valid: bool
    precision: float              # % of predicted actions that were necessary
    recall: float                 # % of required actions that were present
    missing_steps: list[str]      # Required steps not taken
    extra_steps: list[str]        # Steps taken but not required
    score: float                  # Composite 0.0-1.0 score


class TrajectoryAnalyzer:
    """Analyzes agent execution trajectories against ground-truth expectations.

    Compares the sequence of tool calls, decisions, and intermediate results
    an agent produces against a predefined expected path. Different matching
    strategies are available depending on the criticality of the task.

    Reference `code-philosophy` Law 1 (Early Exit): if EXACT match is required
    and any step is missing, fail immediately rather than trying to approximate.
    """

    def analyze(
        self,
        actual_steps: list[TrajectoryStep],
        expected_steps: list[TrajectoryStep],
        match_type: TrajectoryMatchType = TrajectoryMatchType.IN_ORDER,
    ) -> TrajectoryMatchResult:
        """Compare actual agent trajectory against expected trajectory.

        Args:
            actual_steps: The sequence of steps the agent actually took.
            expected_steps: The ground-truth ideal sequence of steps.
            match_type: How strictly to compare the trajectories.

        Returns:
            TrajectoryMatchResult with scores, missing/extra steps, and verdict.
        """
        if not expected_steps:
            return TrajectoryMatchResult(
                match_type=match_type, is_valid=False, precision=0.0,
                recall=0.0, missing_steps=[], extra_steps=[], score=0.0,
            )

        actual_actions = [s.action for s in actual_steps]
        expected_actions = [s.action for s in expected_steps]

        if match_type == TrajectoryMatchType.EXACT:
            return self._exact_match(actual_actions, expected_actions)
        elif match_type == TrajectoryMatchType.IN_ORDER:
            return self._in_order_match(actual_actions, expected_actions)
        elif match_type == TrajectoryMatchType.ANY_ORDER:
            return self._any_order_match(actual_actions, expected_actions)
        else:
            raise ValueError(f"Unknown match type: {match_type}")

    def _exact_match(
        self, actual: list[str], expected: list[str]
    ) -> TrajectoryMatchResult:
        """Exact match — every step must align perfectly in sequence."""
        is_valid = (actual == expected)
        return TrajectoryMatchResult(
            match_type=TrajectoryMatchType.EXACT,
            is_valid=is_valid,
            precision=1.0 if is_valid else 0.0,
            recall=1.0 if is_valid else 0.0,
            missing_steps=[],
            extra_steps=[],
            score=1.0 if is_valid else 0.0,
        )

    def _in_order_match(
        self, actual: list[str], expected: list[str]
    ) -> TrajectoryMatchResult:
        """In-order match — required actions appear in correct order,
        extra intermediate steps are allowed."""
        missing = []
        remaining_expected = list(expected)
        for action in actual:
            if action in remaining_expected:
                remaining_expected.remove(action)

        is_valid = (len(remaining_expected) == 0)
        recall = (len(expected) - len(remaining_expected)) / len(expected) if expected else 0.0
        precision = sum(1 for a in actual if a in expected) / len(actual) if actual else 0.0

        return TrajectoryMatchResult(
            match_type=TrajectoryMatchType.IN_ORDER,
            is_valid=is_valid,
            precision=round(precision, 3),
            recall=round(recall, 3),
            missing_steps=remaining_expected,
            extra_steps=[a for a in actual if a not in expected],
            score=round((precision + recall) / 2, 3),
        )

    def _any_order_match(
        self, actual: list[str], expected: list[str]
    ) -> TrajectoryMatchResult:
        """Any-order match — required actions present regardless of order."""
        expected_set = set(expected)
        found = [a for a in actual if a in expected_set]
        missing = [e for e in expected if e not in set(actual)]

        is_valid = (len(missing) == 0)
        precision = len(found) / len(actual) if actual else 0.0
        recall = (len(expected) - len(missing)) / len(expected) if expected else 0.0

        return TrajectoryMatchResult(
            match_type=TrajectoryMatchType.ANY_ORDER,
            is_valid=is_valid,
            precision=round(precision, 3),
            recall=round(recall, 3),
            missing_steps=missing,
            extra_steps=[a for a in actual if a not in expected_set],
            score=round((precision + recall) / 2, 3),
        )


# --- Concrete Usage Example ---
def demo_trajectory_analysis() -> None:
    """Demonstrate trajectory analysis on a customer service scenario."""
    analyzer = TrajectoryAnalyzer()

    # Expected trajectory for handling a product return request:
    expected_steps = [
        TrajectoryStep(action="determine_intent"),
        TrajectoryStep(action="lookup_order", parameters={"order_id": "ORD-1234"}),
        TrajectoryStep(action="check_return_policy"),
        TrajectoryStep(action="generate_return_label"),
        TrajectoryStep(action="confirm_with_user"),
    ]

    # Actual trajectory — agent skipped the policy check and added an extra step:
    actual_steps = [
        TrajectoryStep(action="determine_intent"),
        TrajectoryStep(action="lookup_order", parameters={"order_id": "ORD-1234"}),
        TrajectoryStep(action="generate_return_label"),  # Skipped policy check!
        TrajectoryStep(action="add_urgency_note"),       # Extra, unnecessary step
        TrajectoryStep(action="confirm_with_user"),
    ]

    result = analyzer.analyze(actual_steps, expected_steps,
                              match_type=TrajectoryMatchType.IN_ORDER)

    print(f"Valid trajectory: {result.is_valid}")
    print(f"Score: {result.score}/1.0")
    print(f"Missing: {result.missing_steps}")
    print(f"Extra:   {result.extra_steps}")


if __name__ == "__main__":
    demo_trajectory_analysis()
```

---

### Pattern 4: Contract-Based Agent Negotiation Framework

For high-stakes tasks, replace free-form prompting with formal contracts that define exact deliverables, acceptable data sources, cost bounds, and verification criteria. The agent negotiates ambiguous terms before execution and self-validates against contract criteria.

```python
"""Contract-based negotiation framework for high-stakes agent tasks."""

from __future__ import annotations

import json
import logging
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from typing import Optional

logger = logging.getLogger(__name__)


class ContractStatus(Enum):
    PENDING_NEGOTIATION = "pending_negotiation"
    NEGOTIATED = "negotiated"
    EXECUTING = "executing"
    VALIDATED = "validated"
    FAILED = "failed"


@dataclass
class ContractDeliverable:
    """A single deliverable defined in a contract."""
    description: str          # What must be produced
    format: str               # Output format (PDF, JSON, Markdown, etc.)
    size_constraints: dict[str, int] = field(default_factory=dict)
    required_sections: list[str] = field(default_factory=list)


@dataclass
class Contract:
    """Formal specification defining a high-stakes agent task.

    Serves as the single source of truth — more precise than any prompt.
    Defines deliverables, scope, data sources, cost bounds, and verification
    criteria so that outcomes are objectively verifiable.
    """
    id: str
    description: str
    deliverables: list[ContractDeliverable]
    acceptable_data_sources: list[str]
    scope_boundary: str          # What is explicitly out of scope
    max_computational_cost_usd: float
    max_completion_time_minutes: int
    created_at: datetime = field(default_factory=datetime.utcnow)
    status: ContractStatus = ContractStatus.PENDING_NEGOTIATION
    negotiation_notes: list[str] = field(default_factory=list)
    compliance_verdict: Optional[bool] = None


@dataclass
class NegotiationResponse:
    """Agent's response to a contract — can accept, negotiate, or flag risks."""

    action: str  # "accept", "negotiate", "flag_risk"
    message: str
    proposed_changes: list[str] = field(default_factory=list)
    concerns: list[str] = field(default_factory=list)
    estimated_cost_usd: float = 0.0
    estimated_time_minutes: int = 0


class ContractNegotiator:
    """Manages contract-based negotiation and self-validation for agent tasks.

    Implements the contractor model: formal contracts with dynamic negotiation,
    quality-focused iterative execution, and hierarchical subcontract decomposition.

    Reference `code-philosophy` Law 3 (Atomic Predictability): each negotiation
    round produces a deterministic, fully-formed ContractStatus transition — no
        ambiguous intermediate states.
    """

    def negotiate(
        self,
        contract: Contract,
        agent_capabilities: dict[str, bool],
    ) -> NegotiationResponse:
        """Negotiate contract terms with the executing agent before work begins.

        Checks whether the agent has access to all required data sources and
        can meet the computational cost/time bounds. Returns a negotiation
        response that either accepts or proposes changes.

        Args:
            contract: The formal contract specification.
            agent_capabilities: Dict mapping capability names to boolean availability.

        Returns:
            NegotiationResponse with acceptance, negotiated changes, or flagged risks.
        """
        concerns: list[str] = []
        proposed_changes: list[str] = []

        # Check data source accessibility
        for source in contract.acceptable_data_sources:
            if not agent_capabilities.get(f"access_{source}", False):
                concerns.append(
                    f"Agent lacks access to required data source: {source}"
                )
                proposed_changes.append(
                    f"Approve use of alternative public source for: {source}"
                )

        # Check cost bounds
        estimated_tokens = len(contract.description) * 2 + \
            sum(len(d.description) for d in contract.deliverables) * 5
        estimated_cost = (estimated_tokens / 1_000_000) * 0.003

        if estimated_cost > contract.max_computational_cost_usd:
            concerns.append(
                f"Estimated cost (${estimated_cost:.4f}) exceeds budget "
                f"(${contract.max_computational_cost_usd:.4f})"
            )
            proposed_changes.append("Reduce number of deliverables to fit budget")

        # Check time bounds
        estimated_time = max(2, len(contract.deliverables) * 5)
        if estimated_time > contract.max_completion_time_minutes:
            concerns.append(
                f"Estimated time ({estimated_time}m) exceeds deadline "
                f"({contract.max_completion_time_minutes}m)"
            )

        if concerns:
            return NegotiationResponse(
                action="negotiate",
                message=f"Contract requires {len(concerns)} adjustment(s) before execution.",
                proposed_changes=proposed_changes,
                concerns=concerns,
                estimated_cost_usd=round(estimated_cost, 4),
                estimated_time_minutes=estimated_time,
            )

        # All checks pass — accept
        contract.status = ContractStatus.NEGOTIATED
        return NegotiationResponse(
            action="accept",
            message="Contract terms are accessible and feasible. Starting execution.",
            estimated_cost_usd=round(estimated_cost, 4),
            estimated_time_minutes=estimated_time,
        )

    def self_validate(self, contract: Contract, actual_output: str) -> bool:
        """Validate the agent's output against the contract specification.

        Checks deliverable format compliance, scope adherence, and size bounds.
        Returns a boolean verdict — True if the output satisfies all contract terms.
        """
        contract.status = ContractStatus.VALIDATED

        # Check scope boundary — ensure no out-of-scope content was included
        # (In production, compare against the scope_boundary string)
        is_within_scope = True  # Would use NLP similarity in production

        # Validate deliverables exist and match format requirements
        all_deliverables_met = len(contract.deliverables) > 0

        contract.compliance_verdict = is_within_scope and all_deliverables_met
        return contract.compliance_verdict


# --- Usage: Full Contract Lifecycle ---
def demo_contract_lifecycle() -> None:
    """Demonstrate the full contract negotiation, execution, and validation cycle."""
    # Step 1: Define a formal contract for a financial analysis task
    contract = Contract(
        id="FA-2025-Q1-EU",
        description=(
            "Produce a 20-page PDF report analyzing European market sales "
            "from Q1 2025, including five specific data visualizations, a "
            "comparative analysis against Q1 2024, and a risk assessment "
            "based on the included supply chain disruption dataset."
        ),
        deliverables=[
            ContractDeliverable(
                description="Sales trend visualization for each EU country",
                format="PNG chart",
                size_constraints={"min_width": 800, "min_height": 600},
            ),
            ContractDeliverable(
                description="Comparative year-over-year analysis table",
                format="Markdown table",
                required_sections=["Revenue", "Growth Rate", "Top Product"],
            ),
            ContractDeliverable(
                description="Supply chain risk assessment narrative",
                format="PDF section",
            ),
        ],
        acceptable_data_sources=["eu_sales_db_v3", "ecb_market_reports"],
        scope_boundary="Do not analyze non-European markets or pre-2024 data.",
        max_computational_cost_usd=0.50,
        max_completion_time_minutes=15,
    )

    # Step 2: Negotiate — does the agent have access to required sources?
    negotiator = ContractNegotiator()
    agent_caps = {
        "access_eu_sales_db_v3": True,
        "access_ecb_market_reports": False,  # Not accessible!
    }

    response = negotiator.negotiate(contract, agent_caps)
    logger.info("Negotiation result: %s — %s", response.action, response.message)

    if response.action == "negotiate":
        print(f"Proposed changes: {response.proposed_changes}")
        # In production: update contract, re-negotiate until accepted.


if __name__ == "__main__":
    demo_contract_lifecycle()
```

---

### Pattern 5: Drift and Anomaly Detection Pipeline

Detect when agent performance degrades over time due to data drift or unusual behavior patterns. Uses statistical methods for drift detection and z-score based anomaly scoring.

```python
"""Drift detection and anomaly detection pipeline for agent performance monitoring."""

from __future__ import annotations

import math
import logging
from collections import deque
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Optional

logger = logging.getLogger(__name__)


@dataclass
class MetricWindow:
    """Sliding window for tracking a single metric over time."""
    name: str
    values: deque[float] = field(default_factory=lambda: deque(maxlen=500))
    timestamps: deque[datetime] = field(default_factory=lambda: deque(maxlen=500))

    def add(self, value: float, timestamp: Optional[datetime] = None) -> None:
        self.values.append(value)
        self.timestamps.append(timestamp or datetime.utcnow())

    @property
    def mean(self) -> float:
        return sum(self.values) / len(self.values) if self.values else 0.0

    @property
    def std_dev(self) -> float:
        if len(self.values) < 2:
            return 0.0
        m = self.mean
        variance = sum((v - m) ** 2 for v in self.values) / (len(self.values) - 1)
        return math.sqrt(variance)

    def z_score(self, value: float) -> float:
        """Compute the z-score of a value relative to this window's history."""
        sd = self.std_dev
        if sd == 0:
            return 0.0
        return (value - self.mean) / sd


class DriftDetector:
    """Detects statistical drift in agent input/output distributions over time.

    Uses a sliding window approach: maintains a baseline period and compares
    recent metric windows against it using mean shift detection with configurable
    sensitivity. For production deployments, consider integrating with KS-test
    based libraries like `scipy.stats.ks_2samp` for formal distribution testing.

    Reference `code-philosophy` Law 4 (Fail Fast): drift detection should never
        block execution — it produces alerts only, allowing the agent to continue.
    """

    def __init__(self, window_name: str, sensitivity_sigma: float = 3.0) -> None:
        self.window_name = window_name
        self.sensitivity_sigma = sensitivity_sigma
        self.baseline_window: Optional[MetricWindow] = None
        self.current_window: MetricWindow = MetricWindow(window_name)
        self.drift_detected: bool = False

    def update(self, value: float) -> dict:
        """Record a new metric value and check for drift against the baseline.

        Args:
            value: The current metric measurement (e.g., accuracy score, token count).

        Returns:
            Dict with drift status, z-score, and recommendation.
        """
        self.current_window.add(value)

        if self.baseline_window is None:
            # First update — establish baseline from 100 samples
            if len(self.current_window.values) >= 100:
                self.baseline_window = MetricWindow(f"{self.window_name}_baseline")
                self.baseline_window.values = deque(
                    list(self.current_window.values)[:100], maxlen=500
                )
            return {"drift": False, "z_score": 0.0, "action": "col

…(truncated)