Benchmark harnesses that survive model churn

Introduction to Incident Replay Design Patterns

Overview of Incident Replay

Incident replay re‑executes a recorded sequence of events—logs, metric snapshots, packet captures, or API calls—to reproduce a failure condition in a controlled environment. It validates root‑cause hypotheses, tests remediation scripts, and trains operators or ML models without impacting production traffic.

A typical replay pipeline consists of:

1. Capture – immutable record (e.g., tcpdump, fluentd, OpenTelemetry exporter).
2. Storage – version‑controlled artifact repository (object store, Git‑LFS).
3. Orchestrator – workflow engine that reads the record, replays actions against a testbed, and verifies outcomes.
4. Verification – assertions or diff‑based checks that the reproduced state matches the original incident.

When the orchestrator relies on a large language model (LLM) or other AI‑assisted tool to interpret logs, generate commands, or reason about next steps, the interface contract between orchestrator and model becomes a moving target: tool‑call syntax, expected JSON schema, and prompting style evolve faster than the underlying network cases.

Importance of Adapting to Changing Model Tool Syntax

Model providers frequently release new versions that:

• Rename or reorder tool parameters (e.g., run_command → execute_shell).
• Change the shape of the tool‑call payload (e.g., from flat arguments to nested {"args": {...}}).
• Adjust the reasoning style (e.g., from chain‑of‑thought to self‑consistency prompting).

Hard‑coding these contracts forces a code change on every model upgrade, creates version‑skew risk, and can silently break replay fidelity. Design patterns that isolate the volatile interface from the stable replay logic let teams upgrade models with minimal disruption while preserving correctness guarantees.

Design Patterns for Handling Syntax Changes

Adapter Pattern for Syntax Conversion

The Adapter pattern wraps a third‑party tool‑calling interface in a stable façade. The orchestrator always talks to the façade using a version‑agnostic contract; the adapter translates to the concrete model’s current syntax.

When to use

• The model’s tool‑call API changes frequently but the semantic meaning (e.g., “run a CLI command”) stays the same.
• You need to support multiple model versions concurrently (e.g., canary rollout).

Structure

+----------------+          +----------------+          +----------------+
| Replay Orchestrator | --> | ToolCallAdapter (IF) | --> | ConcreteAdapterV1 |
+----------------+          +----------------+          +----------------+
                                 ^                         |
                                 |                         v
                                 |                +----------------+
                                 +--------------> | ConcreteAdapterV2 |
                                                +----------------+

ToolCallAdapter defines methods like run_command(cmd: str) -> Result. Each concrete adapter knows the exact JSON shape expected by its model version.

Bridge Pattern for Decoupling Syntax from Logic

While Adapter focuses on translation, Bridge separates abstraction (what the orchestrator wants to do) from implementation (how a particular model does it). This lets you vary both hierarchies independently.

When to use

• You have multiple orthogonal variations: e.g., different tool categories (command execution, file transfer, config push) and multiple model families (OpenAI, Anthropic, local LLMs).
• You want to avoid a combinatorial explosion of subclasses.

Structure

Abstraction (ReplayAction)          Implementor (ToolExecutor)
+------------------+                +----------------------+
| +execute()       |<>------------->| +run_command()       |
| +transfer_file() |<>------------->| +transfer_file()     |
+------------------+                +----------------------+
        ^                                   ^
        |                                   |
+------------------+                +----------------------+
| ConcreteActionA  |                | OpenAIExecutor       |
+------------------+                +----------------------+
| ConcreteActionB  |                | AnthropicExecutor    |
+------------------+                +----------------------+

ReplayAction holds a reference to a ToolExecutor. Swapping executors (e.g., for a new model) does not require changing the action classes.

Code Examples: Adapter and Bridge Patterns

# -------------------------------------------------
# 1. Adapter – stable façade for tool calls
# -------------------------------------------------
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any, Dict

@dataclass
class ToolResult:
    stdout: str
    stderr: str
    exit_code: int

class ToolCallAdapter(ABC):
    """Version‑agnostic interface used by the replay orchestrator."""
    @abstractmethod
    def run_command(self, cmd: str) -> ToolResult: ...
    @abstractmethod
    def transfer_file(self, src: str, dst: str) -> ToolResult: ...

# Concrete adapter for Model‑V1 (expects flat args)
class V1Adapter(ToolCallAdapter):
    def __init__(self, client):
        self._client = client   # hypothetical low‑level model client

    def run_command(self, cmd: str) -> ToolResult:
        payload = {"tool": "run_command", "arguments": {"command": cmd}}
        resp = self._client.invoke(payload)   # returns dict per V1 spec
        return ToolResult(
            stdout=resp.get("stdout", ""),
            stderr=resp.get("stderr", ""),
            exit_code=resp.get("exit_code", 1),
        )

    def transfer_file(self, src: str, dst: str) -> ToolResult:
        payload = {"tool": "transfer_file", "arguments": {"src": src, "dst": dst}}
        resp = self._client.invoke(payload)
        return ToolResult(
            stdout=resp.get("stdout", ""),
            stderr=resp.get("stderr", ""),
            exit_code=resp.get("exit_code", 1),
        )

# Concrete adapter for Model‑V2 (expects nested args under "args")
class V2Adapter(ToolCallAdapter):
    def __init__(self, client):
        self._client = client

    def run_command(self, cmd: str) -> ToolResult:
        payload = {
            "name": "run_command",
            "args": {"command": cmd},   # V2 shape
        }
        resp = self._client.invoke(payload)
        return ToolResult(
            stdout=resp.get("output", {}).get("stdout", ""),
            stderr=resp.get("output", {}).get("stderr", ""),
            exit_code=resp.get("output", {}).get("exit_code", 1),
        )

    def transfer_file(self, src: str, dst: str) -> ToolResult:
        payload = {
            "name": "transfer_file",
            "args": {"src": src, "dst": dst},
        }
        resp = self._client.invoke(payload)
        return ToolResult(
            stdout=resp.get("output", {}).get("stdout", ""),
            stderr=resp.get("output", {}).get("stderr", ""),
            exit_code=resp.get("output", {}).get("exit_code", 1),
        )

# -------------------------------------------------
# 2. Bridge – abstraction (ReplayAction) vs implementor (ToolExecutor)
# -------------------------------------------------
class ToolExecutor(ABC):
    """Implementor hierarchy – knows how to talk to a specific model."""
    @abstractmethod
    def execute(self, action: str, **kwargs) -> ToolResult: ...

class OpenAIExecutor(ToolExecutor):
    def __init__(self, adapter: ToolCallAdapter):
        self._adapter = adapter

    def execute(self, action: str, **kwargs) -> ToolResult:
        if action == "run_command":
            return self._adapter.run_command(kwargs["command"])
        if action == "transfer_file":
            return self._adapter.transfer_file(kwargs["src"], kwargs["dst"])
        raise ValueError(f"Unsupported action {action}")

class AnthropicExecutor(ToolExecutor):
    def __init__(self, adapter: ToolCallAdapter):
        self._adapter = adapter

    def execute(self, action: str, **kwargs) -> ToolResult:
        # Anthropic may have slightly different action names
        if action == "run_command":
            return self._adapter.run_command(kwargs["command"])
        if action == "transfer_file":
            return self._adapter.transfer_file(kwargs["src"], kwargs["dst"])
        raise ValueError(f"Unsupported action {action}")

# Abstraction – what the orchestrator wants to do
class ReplayAction:
    def __init__(self, executor: ToolExecutor):
        self._executor = executor

    def run_command(self, command: str) -> ToolResult:
        return self._executor.execute("run_command", command=command)

    def transfer_file(self, src: str, dst: str) -> ToolResult:
        return self._executor.execute("transfer_file", src=src, dst=dst)

# -------------------------------------------------
# Usage example – swapping models without touching ReplayAction
# -------------------------------------------------
if __name__ == "__main__":
    # Suppose we have a low‑level client shim for each provider
    openai_client = ...   # placeholder
    anthropic_client = ...

    v1_adapter = V1Adapter(openai_client)
    v2_adapter = V2Adapter(anthropic_client)

    # Bridge composition
    openai_action = ReplayAction(OpenAIExecutor(v1_adapter))
    anthropic_action = ReplayAction(AnthropicExecutor(v2_adapter))

    # Both calls use the same ReplayAction API
    result1 = openai_action.run_command("show interface status")
    result2 = anthropic_action.transfer_file("/tmp/pcap.pcap", "/replay/pcap.pcap")

Key points

• The orchestrator only sees ReplayAction.
• Adding a new model version only requires a new ToolCallAdapter implementation; the ReplayAction and ToolExecutor hierarchies stay untouched.
• If a new action type appears (e.g., apply_config), you extend ToolExecutor and optionally add a convenience method on ReplayAction—again without touching existing adapters.

Design Patterns for Context Handling Evolution

Strategy Pattern for Context Handling

Context handling refers to how the orchestrator builds the prompt or tool‑call payload from the incident record (e.g., selecting relevant log lines, enriching with topology, applying time windows). As models improve, the optimal context strategy may shift from simple concatenation to retrieval‑augmented generation (RAG), hierarchical summarization, or dynamic windowing.

The Strategy pattern encapsulates each context‑building algorithm behind a common interface, allowing the orchestrator to swap strategies at runtime or via configuration.

Interface

from abc import ABC, abstractmethod
from typing import Any, Dict

class ContextStrategy(ABC):
    @abstractmethod
    def build(self, incident: Dict[str, Any]) -> str: ...

Concrete strategies

• RawLogStrategy – returns the raw log slice unchanged.
• SlidingWindowStrategy – returns a window of N lines around each error timestamp.
• TopologyEnrichedStrategy – augments logs with interface descriptions from a CMDB.
• RagStrategy – queries a vector store for semantically similar past incidents and concatenates the top‑k results.

The orchestrator holds a reference to a ContextStrategy and calls build() before invoking the model.

Observer Pattern for Notifying Context Changes

During a long replay, the incident record may be updated incrementally (e.g., new log chunks streamed from a live capture). Observers let interested components—such as a context‑strategy selector, a metrics exporter, or a UI—react to these changes without tight coupling.

Core components

• Subject – the replay engine that emits events when a new chunk arrives.
• Observer – defines an update(chunk) method.
• Concrete observers: ContextReevaluator (may switch strategy), LatencyTracker, AlertDispatcher.

Pseudo‑code

class ReplaySubject:
    def __init__(self):
        self._observers: list[Observer] = []

    def attach(self, obs: Observer):
        self._observers.append(obs)

    def detach(self, obs: Observer):
        self._observers.remove(obs)

    def _notify(self, chunk: Dict):
        for obs in self._observers:
            obs.update(chunk)

    def ingest_chunk(self, chunk: Dict):
        # ... store chunk ...
        self._notify(chunk)

class ContextReevaluator(Observer):
    def __init__(self, strategy_factory):
        self._factory = strategy_factory
        self.current_strategy: ContextStrategy | None = None

    def update(self, chunk):
        # Example heuristic: if chunk size > threshold, switch to RagStrategy
        if len(chunk.get("lines", [])) > 5000:
            self.current_strategy = self._factory.create("rag")
        else:
            self.current_strategy = self._factory.create("sliding_window")

CLI Examples: Strategy and Observer Patterns

Directory layout

replay/
├─ replay_ctl.py          # orchestrator CLI
├─ strategies/
│   ├─ __init__.py
│   ├─ raw.py
│   ├─ sliding.py
│   ├─ topology.py
│   └─ rag.py
└─ observers/
    ├─ __init__.py
    └─ reevaluator.py

replay_ctl.py

#!/usr/bin/env python3
import argparse, json, sys
from strategies import raw, sliding, topology, rag  # noqa: F401
from observers.reevaluator import ContextReevaluator

STRATEGY_MAP = {
    "raw": raw.RawLogStrategy,
    "sliding": sliding.SlidingWindowStrategy,
    "topology": topology.TopologyEnrichedStrategy,
    "rag": rag.RagStrategy,
}

def load_strategy(name: str) -> "ContextStrategy":
    cls = STRATEGY_MAP.get(name)
    if not cls:
        sys.exit(f"Unknown strategy: {name}")
    return cls()

def main():
    parser = argparse.ArgumentParser(description="Replay incident with selectable context strategy")
    parser.add_argument("--incident", required=True, help="Path to JSONL incident file")
    parser.add_argument("--strategy", default="sliding", help="Context strategy name")
    parser.add_argument(
        "--observe",
        action="store_true",
        help="Enable observer for dynamic strategy switch during ingest",
    )
    args = parser.parse_args()

    strategy = load_strategy(args.strategy)
    if args.observe:
        reevaluator = ContextReevaluator(load_strategy)
        print("Observer attached – strategy may change during ingest")

    with open(args.incident, "r") as f:
        for line in f:
            record = json.loads(line)
            prompt = strategy.build(record)
            # In a real engine you would send `prompt` to the model here.
            print(prompt)  # demo output

if __name__ == "__main__":
    main()

Usage

# Static strategy
python replay_ctl.py --incident incident.jsonl --strategy sliding

# With observer for dynamic switching
python replay_ctl.py --incident incident.jsonl --strategy sliding --observe

These patterns keep the replay orchestrator resilient to rapid changes in model tool syntax, context‑handling techniques, and reasoning styles while preserving the fidelity of incident replays.