diff --git a/benchmarks/reliability/ARCHITECTURE.md b/benchmarks/reliability/ARCHITECTURE.md
new file mode 100644
index 000000000..b4f581bae
--- /dev/null
+++ b/benchmarks/reliability/ARCHITECTURE.md
@@ -0,0 +1,230 @@
+# Fault-injection reliability benchmark architecture
+
+Status: verified architecture for the implemented benchmark core and its
+infrastructure adapter boundary.
+
+## Evidence base
+
+The architecture was checked against:
+
+- `OpenHands/benchmarks` at `4e5469e0caaf54d1ad827d18b524bdfb79d58430`
+- `OpenHands/software-agent-sdk` at
+  `6d597ff7d5d3c89ef8ba0c8e3b3c6a09169da07c`
+- `OpenHands/benchmarks#488`
+- `OpenHands/OpenHands#14260` and `OpenHands/OpenHands#13578`
+
+The source was first inspected through the connected GitHub app at the pinned
+revisions, then checked against local clones before this implementation was published.
+
+## Existing execution path
+
+For SWE-bench, the verified path is:
+
+1. `benchmarks/utils/evaluation.py:Evaluation.run`
+2. `Evaluation._run_iterative_mode_async`
+3. `Evaluation._run_attempt_async`
+4. `Evaluation._process_one_sync`
+5. `Evaluation._execute_single_attempt`
+6. `benchmarks/swebench/run_infer.py:SWEBenchEvaluation.prepare_workspace`
+7. `SWEBenchEvaluation.evaluate_instance`
+8. `benchmarks/utils/fake_user_response.py:
+   run_conversation_with_fake_user_response`
+9. SDK `Conversation.run` / `RemoteConversation.run`
+10. agent response dispatch, persisted action event, tool execution, and
+    persisted observation/error event
+
+`Evaluation._execute_single_attempt` is the runner-level lifecycle boundary. It
+creates the workspace, calls the benchmark-specific evaluator, captures failure
+artifacts, classifies retries, and cleans up. It is appropriate for scenario
+ownership, coarse runtime faults, and fault receipts. It is not precise enough
+to model an ambiguous tool outcome.
+
+`SWEBenchEvaluation.prepare_workspace` is the concrete workspace selection seam.
+It constructs `DockerWorkspace`, `ApptainerWorkspace`, or
+`APIRemoteWorkspace`, using an eval agent-server image. This is where a
+scenario-scoped lifecycle adapter can be attached without changing task
+grading.
+
+`SWEBenchEvaluation.evaluate_instance` creates the SDK `Conversation` with
+`build_event_persistence_callback(...)`, sends the instruction, and enters
+`run_conversation_with_fake_user_response`. The callback and returned event
+history are the primary inspectable trigger/evidence stream.
+
+## Verified SDK state and dispatch model
+
+`openhands-sdk/openhands/sdk/conversation/state.py:
+ConversationState.append_event` is the event storage chokepoint. It appends to
+`EventLog` and advances the active event leaf.
+
+`conversation/event_store.py:EventLog.append` writes one serialized event under
+a file-store lock and rejects duplicate event IDs. `ConversationState.create`
+opens or creates the store, reloads `base_state.json`, attaches the event log,
+and rebuilds derived state.
+
+The relevant ambiguous-outcome window is verified in:
+
+- `openhands-sdk/openhands/sdk/agent/response_dispatch.py:
+  _handle_tool_calls` / `_ahandle_tool_calls`
+- `openhands-sdk/openhands/sdk/agent/agent.py:_ActionBatch.prepare`
+- `agent.py:_ActionBatch.emit`
+- `agent.py:_execute_action_event`
+
+The dispatcher emits an `ActionEvent` through `on_event` before executing the
+tool. Observations or errors are emitted only after the tool runner returns.
+Therefore a process can die after an irreversible external effect commits but
+before its observation is persisted.
+
+`Agent.step` checks `ConversationState.get_unmatched_actions` before sampling a
+new response and can execute unmatched actions. This is a useful replay seam
+and also the source of duplicate-effect risk in ambiguous outcomes.
+
+Agent-server cold load behaves differently. In
+`openhands-agent-server/openhands/agent_server/event_service.py:
+EventService.start`, a persisted conversation left `RUNNING` is changed to
+`ERROR`; the server emits an `AgentErrorEvent` for the first unmatched action.
+That prevents blind replay of that action on this path, but does not establish
+whether the external effect committed. Parallel batches may leave additional
+unmatched actions, so the benchmark must record actual behavior rather than
+assume all orphans are reconciled.
+
+## Fault hook map
+
+### Sandbox restart
+
+Runner ownership:
+
+- `Evaluation._execute_single_attempt`
+- `SWEBenchEvaluation.prepare_workspace`
+
+Local runtime seams:
+
+- `openhands-workspace/openhands/workspace/docker/workspace.py:
+  DockerWorkspace._start_container`
+- `DockerWorkspace.cleanup`, `pause`, and `resume`
+
+Remote runtime seams:
+
+- `openhands-workspace/openhands/workspace/remote_api/workspace.py:
+  _start_or_attach_to_runtime`
+- `_start_runtime`, `_resume_runtime`, `pause`, and `resume`
+
+A restart injector must stop/replace the agent-server container or remote
+runtime and then exercise the real attach/load path. Docker pause/unpause is
+not a restart and must be reported as a separate fault if used.
+
+### Lost dispatch response
+
+The benchmark wrapper around `Conversation.run` can drop a run-level result,
+but that does not prove a tool dispatch committed. The precise in-process seam
+is after `_execute_action_event` returns and before `_ActionBatch.emit`
+persists the resulting observation.
+
+For remote runs, the production request/response seam still needs confirmation
+in the remote conversation or workspace client before implementation. The
+`APIRemoteWorkspace._send_api_request` seam controls runtime lifecycle calls,
+not every tool call to the agent server. The scenario must state whether it
+models:
+
+- a lost tool result before observation persistence, or
+- a lost client/agent-server transport response.
+
+They are not interchangeable.
+
+### SIGKILL mid-tool-call
+
+Trigger on the persisted `ActionEvent`, then kill the process/container while
+`_execute_action_event` is active and before an observation/error exists.
+
+`LocalConversation.interrupt()` is not a substitute for SIGKILL. It sets a
+cancellation token and cancels the tracked async task; worker threads may
+continue until their tools return. A real SIGKILL bypasses cleanup and must be
+implemented at the process/container layer.
+
+### Network partition
+
+Runtime-control partitions can wrap
+`APIRemoteWorkspace._send_api_request`. Agent-server tool/run traffic requires
+the corresponding remote client seam to be verified before implementation.
+Partitions need an explicit direction, endpoint class, start trigger, and
+duration or release trigger.
+
+`benchmarks/utils/acp.py:workspace_keepalive` is an existing observation point.
+It executes `true` every 60 seconds and suppresses failures. Issue #488 shows it
+is not a sufficient recovery mechanism.
+
+## Outcome judgment
+
+Task correctness stays with the existing benchmark grader. For SWE-bench,
+`benchmarks/swebench/eval_infer.py:run_swebench_evaluation` invokes
+`python -m swebench.harness.run_evaluation`. SDK critics only gate retries or
+rank candidates; they are not the official correctness oracle.
+
+Reliability grading consumes inspectable artifacts:
+
+- terminal `ConversationExecutionStatus`
+- ordinary task-grader result
+- persisted event log and callback event stream
+- scheduled and observed fault receipts
+- runtime/conversation identity before and after recovery
+- scenario-owned external effect ledger
+- baseline and faulted timing, iteration, event, LLM, and tool-call metrics
+
+Completion/resume passes only when the ordinary task passes and the run shows a
+real reattachment/recovery path. Starting the instance over in a fresh
+workspace is a retry, not resume.
+
+No-duplicate-effect grading cannot be inferred from the SDK event log alone.
+Each irreversible-action scenario must expose an inspectable ledger or
+query-by-idempotency-key oracle independent of the agent. The grader compares
+committed effects against intended logical operations.
+
+Recovery overhead compares a faulted run with a matched no-fault baseline for
+the same scenario, seed, instance, agent, model, and resource configuration.
+Raw measures include wall time, recovery-window time, iterations, event count,
+tool calls, token/cost metrics, and replay/detection time.
+
+## Persistence and restart receipts
+
+Issue #14260 demonstrates that durable OpenHands events do not imply provider
+session recovery. Acceptance evidence should distinguish:
+
+- `conversation_history_restored`
+- `provider_session_rebound`
+
+The second requires the same provider session ID/cwd and preserved provider
+storage, not merely a bootstrap prompt containing prior messages.
+
+Issue #13578 demonstrates that successful in-memory agent activity is not
+evidence of durable recovery. The event store must be visible after app/server
+restart. That issue narrows the startup timeout only to app-server/agent-server
+connectivity; it does not identify a more specific root cause, so the benchmark
+must not encode one as fact.
+
+## Existing recovery measurement to reuse
+
+`scripts/event_sourcing_benchmarks/bench_replay_and_recovery.py` already
+measures event deserialization, replay, and unmatched-action scan cost on real
+SWE-bench traces. Reuse its concepts and data where possible. It does not inject
+live faults, judge end-to-end task completion, or detect duplicate external
+effects.
+
+## Placement
+
+The first milestone should live as a benchmark package under
+`benchmarks/reliability`. Scenario adapters remain package-local until at least
+one other benchmark needs them. Only then should generic components move to
+`benchmarks/utils`, consistent with the repository's contribution guidance.
+
+SDK changes should be limited to narrow, reusable injection hooks or observable
+receipts that cannot be implemented at the benchmark layer. The current
+`build_reliability_event_callback` composes with SDK callbacks without changing
+the vendored SDK.
+
+## Adapter-specific follow-up
+
+- The exact remote client seam for lost tool responses and agent-server network
+  partitions.
+- Whether cold-start recovery should reconcile all unmatched parallel actions
+  or intentionally fail closed after the first.
+- Which irreversible test action is acceptable for the initial milestone.
+- Whether provider-session rebound is in scope for V1 or reported separately.
diff --git a/benchmarks/reliability/DESIGN.md b/benchmarks/reliability/DESIGN.md
new file mode 100644
index 000000000..e7c0dd48b
--- /dev/null
+++ b/benchmarks/reliability/DESIGN.md
@@ -0,0 +1,283 @@
+# Fault-injection reliability benchmark design
+
+Status: implemented benchmark core; infrastructure-specific adapters supply the
+real runtime operations.
+
+## Goals
+
+The benchmark runs a real agent task under a deterministic, declarative fault
+schedule and produces an inspectable reliability scorecard. It answers:
+
+1. Did the run recover and complete the ordinary task?
+2. Did replay avoid duplicating an already committed irreversible effect?
+3. What recovery overhead did the fault add?
+
+It does not replace the task's existing grader, infer external commits from
+language-model output, or treat a fresh retry as successful resume.
+
+## Scenario schema
+
+Every scenario is a versioned JSON document. A complete example lives at
+`benchmarks/reliability/scenarios/example.json`.
+
+Required identity fields are `schema_version`, `scenario_id`, task identity,
+agent/config identity, and seed. A run manifest records resolved defaults,
+source revisions, images, resource factors, and generated fault timestamps.
+
+## Determinism
+
+The schedule takes an explicit integer seed. Random choices use a dedicated
+scenario RNG and never module-global randomness. A schedule is compiled before
+the run into ordered fault specifications. Trigger matching is deterministic
+over the persisted event stream and explicit lifecycle phases.
+
+Preferred trigger selectors:
+
+- lifecycle phase, such as `workspace_ready` or `conversation_attached`
+- event type plus ordinal
+- action/tool-call ID
+- logical operation ID from a scenario-owned effect
+
+Wall-clock-only triggers are allowed for transport duration but are not the
+primary placement mechanism because scheduler variance makes them hard to
+reproduce.
+
+Every injected fault produces a receipt containing the scenario/run/fault IDs,
+seed, resolved trigger, observed event/tool-call ID, monotonic timestamp,
+runtime and conversation identities, injector result, and release result.
+
+## Fault interface
+
+The implementation defines a typed `FaultInjector` protocol:
+
+- `arm(context, fault) -> FaultHandle`
+- `inject(context, fault, handle) -> FaultReceipt`
+- `release(context, fault, handle) -> FaultReceipt | None`
+
+`FaultContext` exposes only explicit adapters: event observation, runtime
+lifecycle control, transport control, process control, and receipt recording.
+It must not grant graders access to hidden agent reasoning.
+
+### Sandbox restart
+
+Parameters:
+
+- target: agent-server container or remote runtime
+- restart mode: graceful, hard stop, or replacement
+- persistence policy: preserve workspace/events, preserve events only, or
+  scenario-defined
+- reattach deadline
+
+The injector records pre/post runtime ID, conversation ID, event-store
+location, provider session ID when available, and first post-recovery event.
+
+### Lost dispatch response
+
+Parameters:
+
+- boundary: tool-result-before-observation or remote transport response
+- target tool/action selector
+- drop count
+
+The first milestone should implement only a boundary verified in the actual
+execution path. A synthetic exception before dispatch is not a lost response.
+The effect ledger independently records whether the external operation
+committed.
+
+### SIGKILL mid-tool-call
+
+Parameters:
+
+- target process/container
+- target action selector
+- signal (`SIGKILL` for the named scenario)
+- optional replacement/start policy
+
+The trigger requires a persisted action receipt and no persisted
+observation/error at kill time. Cooperative `interrupt()` is a different fault
+kind and must not be reported as SIGKILL.
+
+### Network partition
+
+Parameters:
+
+- endpoint class: runtime control, conversation transport, tool transport, or
+  scenario external service
+- direction: ingress, egress, or both
+- failure mode: drop, timeout, reset, or bounded latency
+- release: duration or deterministic receipt trigger
+
+The report names the actual wrapped client/endpoint. “Network partition”
+without a boundary is invalid.
+
+## Inspectable scenario effects
+
+Duplicate-effect scenarios use a deterministic local service or ledger-backed
+tool. Each logical action has:
+
+- `operation_id`
+- optional idempotency key
+- request payload digest
+- commit sequence and timestamp
+- query endpoint or append-only ledger
+
+The grader reads this evidence directly. The agent cannot mark its own effect
+as successful. Initial scenarios should avoid real destructive external
+systems; they can model an irreversible append, charge, or publish operation in
+a local deterministic service.
+
+## Run phases
+
+1. Resolve scenario and seed.
+2. Record source/image/config manifest.
+3. Execute a matched no-fault baseline if one is not cached.
+4. Prepare the real benchmark workspace and conversation.
+5. Arm the compiled schedule.
+6. Stream persisted events and lifecycle receipts to the scheduler.
+7. Inject and release faults at resolved triggers.
+8. Exercise the product's real resume/reattach path.
+9. Run the ordinary benchmark grader.
+10. Run inspectable reliability graders.
+11. Emit per-run artifacts and aggregate scorecard.
+
+Failed injection is not silently converted to an agent failure. It is a
+separate `invalid_injection` outcome so reliability scores cannot improve when
+the requested fault never happened.
+
+## Scoring model
+
+Report raw evidence and sub-scores. Do not collapse failures into a single
+opaque model judgment.
+
+### Completion/resume
+
+Per run:
+
+- `task_completed`: ordinary grader pass/fail
+- `fault_injected`: required receipt present
+- `history_restored`: persisted event continuity observed
+- `runtime_reattached`: recovery path observed
+- `provider_session_rebound`: true, false, or not applicable
+- `fresh_retry_used`: whether a new attempt/workspace replaced recovery
+
+`completion_resume_score` is 1 only when the fault was injected, the ordinary
+task passed, required recovery receipts are present, and the scenario did not
+fall back to a forbidden fresh retry. Otherwise it is 0. Aggregate as a rate
+with numerator and denominator shown.
+
+### No duplicate irreversible effect
+
+For each logical operation:
+
+```text
+duplicate_count = max(0, committed_effect_count - intended_effect_count)
+```
+
+Per-run score is 1 only when all intended effects have exactly the expected
+commit count and payload. Missing effects and duplicate effects are reported
+separately. Aggregate:
+
+- clean-effect run rate
+- duplicated logical operations / attempted logical operations
+- total excess commits
+
+### Recovery overhead
+
+Compare a faulted run with its matched baseline:
+
+```text
+wall_time_ratio = faulted_wall_time / baseline_wall_time
+wall_time_delta = faulted_wall_time - baseline_wall_time
+```
+
+Also report recovery-window seconds, iteration delta, event delta, tool-call
+delta, token/cost delta, and replay/detection time. Ratios with a zero or
+missing baseline are null and excluded from aggregates, never coerced.
+
+The aggregate scorecard presents medians and tail percentiles rather than a
+hidden weighted scalar. If maintainers later request one headline number, its
+formula and component weights must be published alongside the raw components.
+
+## Grader interface
+
+Each grader receives:
+
+- resolved scenario and manifest
+- baseline evidence when applicable
+- persisted event evidence
+- fault receipts
+- runtime/recovery receipts
+- ordinary benchmark result
+- external effect ledger
+- metrics and timestamps
+
+It returns a typed result with `passed`, `value`, `reason_codes`, and
+`evidence_refs`. Reason codes are stable and machine-readable; evidence
+references point to concrete JSONL records, event IDs, or grader outputs.
+
+No grader calls an LLM.
+
+## Artifacts
+
+Per run:
+
+- `manifest.json`
+- `resolved_schedule.json`
+- `fault_receipts.jsonl`
+- `recovery_receipts.jsonl`
+- `events.jsonl` or an immutable reference to the product event store
+- `effects.jsonl`
+- `native_grader.json`
+- `reliability_result.json`
+- `metrics.json`
+
+Aggregate:
+
+- `scorecard.json`
+- `scorecard.md`
+- optional CSV rows for analysis
+
+Secrets, provider transcripts not needed for grading, and raw environment
+values are excluded or redacted.
+
+## Scorecard
+
+The Markdown scorecard groups by agent/config, scenario, fault kind, and seed
+set. Columns:
+
+- runs / valid injections
+- ordinary task pass rate
+- completion/resume rate
+- clean-effect run rate
+- duplicate operations and excess commits
+- median and p95 recovery seconds
+- median wall-time ratio
+- median tool-call, token, and cost deltas
+- invalid-injection and infrastructure-failure counts
+
+Links from each aggregate row lead to inspectable per-run artifacts and reason
+codes.
+
+## Test strategy after scope approval
+
+- schema validation and deterministic schedule compilation
+- trigger matching over fixed event fixtures
+- injector contract tests with fake adapters
+- grader tests over hand-authored event/effect ledgers
+- negative tests for fault-not-injected, fresh retry, missing baseline, missing
+  effect, and duplicate effect
+- one live local agent-server smoke scenario
+- one real SWE-bench subset scenario only after the local smoke path is stable
+
+The focused tests exercise schema rejection, deterministic schedules, SDK
+callback ordering, effect evidence, all four injectors, every grader, and both
+scorecard formats.
+
+## Phasing
+
+The generic harness, evidence model, graders, reporting, and SDK event callback
+are implemented in this package. Infrastructure adapters should land
+incrementally because a Docker restart, remote-runtime replacement, and network
+partition require different privileges and recovery operations. Each adapter
+must identify its concrete boundary in receipts and add a live smoke test before
+being used for published scores.
diff --git a/benchmarks/reliability/README.md b/benchmarks/reliability/README.md
new file mode 100644
index 000000000..215cc919d
--- /dev/null
+++ b/benchmarks/reliability/README.md
@@ -0,0 +1,121 @@
+# Fault-injection reliability benchmark
+
+This package implements the benchmark core proposed in
+[RFC #764](https://github.com/OpenHands/benchmarks/issues/764).
+
+It runs a matched no-fault baseline and faulted run, places faults from a
+seeded declarative schedule after persisted SDK events, records external-effect
+evidence independently of the agent, and emits inspectable JSON and Markdown
+scorecards.
+
+## What is implemented
+
+- strict, versioned JSON scenario loading
+- deterministic event-trigger matching with duplicate-event suppression
+- explicit adapters for sandbox restart, lost dispatch response, `SIGKILL`
+  mid-tool-call, and network partition/release
+- durable event, fault, recovery, effect, task, and metric artifacts
+- a scenario-owned append-only effect ledger with canonical payload digests
+- deterministic completion/resume, duplicate-effect, and recovery-overhead
+  graders
+- JSON and Markdown scorecards with raw component results
+- an OpenHands SDK callback adapter that preserves persist-before-inject order
+- a CLI entrypoint and end-to-end tests covering all four fault categories
+
+The core does not pretend that pausing a process is a sandbox restart or that
+raising an exception before dispatch is a lost response. A benchmark-specific
+adapter must implement the real infrastructure operation at a named, verified
+boundary and return details for the fault receipt.
+
+## Scenario
+
+See [`scenarios/example.json`](scenarios/example.json). Every trigger names a
+persisted event type and 1-based ordinal. An optional `tool_call_id` narrows the
+match. Fault IDs must be unique.
+
+The seed deterministically orders faults that share a trigger. Replaying the
+same event ID never fires a fault twice.
+
+## Adapter contract
+
+An adapter implements `ReliabilityAdapter.open_run` and returns a `RunSession`.
+The session:
+
+1. executes one independent task run;
+2. publishes persisted events synchronously through `on_event`;
+3. implements the four explicit `FaultContext` operations;
+4. writes intended and committed external operations through
+   `EffectLedger(artifacts.effects_path)`;
+5. returns native task success, recovery receipts, and raw metrics.
+
+Use `build_reliability_event_callback` to compose the reliability callback with
+the existing benchmark event-persistence callback. Existing callbacks run
+first, so a fault cannot fire before its trigger event is durable.
+
+For a recovery to score as resume, the adapter must emit successful
+`conversation_history_restored` and `runtime_reattached` receipts. A run that
+sets `fresh_retry_used=True` fails the resume grader even if the ordinary task
+eventually passes.
+
+## Run
+
+Register a zero-argument adapter factory and run:
+
+```bash
+uv run reliability-eval \
+  --scenario benchmarks/reliability/scenarios/example.json \
+  --adapter your_package.reliability_adapter:create_adapter \
+  --output-dir evaluation_outputs/reliability
+```
+
+The adapter is loaded with ordinary `importlib`; no `sys.path` mutation is
+performed.
+
+## Artifacts
+
+Each baseline and faulted run contains:
+
+- `manifest.json`
+- `events.jsonl`
+- `fault_receipts.jsonl`
+- `recovery_receipts.jsonl`
+- `effects.jsonl`
+- `task_result.json`
+- `metrics.json`
+- `reliability_result.json` for the faulted run
+
+The output root also receives `scorecard.json` and `scorecard.md`.
+
+## Grading
+
+Completion/resume requires:
+
+- the ordinary benchmark task to pass;
+- every scheduled fault to have an `applied` receipt;
+- no injection failure;
+- history-restored and runtime-reattached recovery receipts; and
+- no forbidden fresh retry.
+
+The duplicate-effect grader compares effect `intent` and `commit` records by
+logical operation ID and canonical payload digest. It reports missing,
+unexpected, mismatched, and excess commits separately.
+
+Recovery overhead compares the faulted run against its matched baseline and
+reports wall-time delta/ratio, recovery time, and iteration, event, tool-call,
+token, and cost deltas. Missing or zero baselines remain explicit rather than
+being coerced.
+
+## Validate
+
+```bash
+uv run pre-commit run --files benchmarks/reliability tests/test_reliability.py
+uv run pytest tests/test_reliability.py
+```
+
+## Context
+
+- [Architecture](ARCHITECTURE.md)
+- [Design](DESIGN.md)
+- [`OpenHands/benchmarks#488`](https://github.com/OpenHands/benchmarks/issues/488)
+- [`OpenHands/OpenHands#14260`](https://github.com/OpenHands/OpenHands/issues/14260)
+- [`OpenHands/OpenHands#13578`](https://github.com/OpenHands/OpenHands/issues/13578)
diff --git a/benchmarks/reliability/__init__.py b/benchmarks/reliability/__init__.py
new file mode 100644
index 000000000..8dc31b208
--- /dev/null
+++ b/benchmarks/reliability/__init__.py
@@ -0,0 +1,25 @@
+"""Inspectable, adapter-backed fault-injection reliability benchmark."""
+
+from benchmarks.reliability.models import (
+    EffectRecord,
+    EventRecord,
+    FaultKind,
+    FaultReceipt,
+    FaultSpec,
+    ReliabilityResult,
+    Scenario,
+)
+from benchmarks.reliability.runner import ReliabilityAdapter, run_scenario
+
+
+__all__ = [
+    "EffectRecord",
+    "EventRecord",
+    "FaultKind",
+    "FaultReceipt",
+    "FaultSpec",
+    "ReliabilityAdapter",
+    "ReliabilityResult",
+    "Scenario",
+    "run_scenario",
+]
diff --git a/benchmarks/reliability/artifacts.py b/benchmarks/reliability/artifacts.py
new file mode 100644
index 000000000..5646e4031
--- /dev/null
+++ b/benchmarks/reliability/artifacts.py
@@ -0,0 +1,195 @@
+"""Stable JSON/JSONL artifact storage for reliability runs."""
+
+import json
+import os
+import threading
+from dataclasses import dataclass
+from pathlib import Path
+from typing import cast
+
+from benchmarks.reliability.models import (
+    EffectRecord,
+    EventRecord,
+    FaultReceipt,
+    JsonObject,
+    JsonValue,
+    RecoveryReceipt,
+    ReliabilityResult,
+    RunIdentity,
+    RunMetrics,
+    RunObservation,
+    Scenario,
+    as_json_object,
+)
+
+
+@dataclass(frozen=True, slots=True)
+class RunArtifacts:
+    """Canonical artifact paths and append operations for one run."""
+
+    root: Path
+
+    @property
+    def manifest_path(self) -> Path:
+        return self.root / "manifest.json"
+
+    @property
+    def events_path(self) -> Path:
+        return self.root / "events.jsonl"
+
+    @property
+    def fault_receipts_path(self) -> Path:
+        return self.root / "fault_receipts.jsonl"
+
+    @property
+    def recovery_receipts_path(self) -> Path:
+        return self.root / "recovery_receipts.jsonl"
+
+    @property
+    def effects_path(self) -> Path:
+        return self.root / "effects.jsonl"
+
+    @property
+    def metrics_path(self) -> Path:
+        return self.root / "metrics.json"
+
+    @property
+    def task_result_path(self) -> Path:
+        return self.root / "task_result.json"
+
+    @property
+    def reliability_result_path(self) -> Path:
+        return self.root / "reliability_result.json"
+
+    def initialize(
+        self,
+        *,
+        run: RunIdentity,
+        scenario: Scenario,
+        baseline: bool,
+    ) -> None:
+        """Create an empty, self-identifying artifact directory."""
+        self.root.mkdir(parents=True, exist_ok=False)
+        write_json(
+            self.manifest_path,
+            {
+                "run": as_json_object(run),
+                "scenario": as_json_object(scenario),
+                "baseline": baseline,
+            },
+        )
+        for path in (
+            self.events_path,
+            self.fault_receipts_path,
+            self.recovery_receipts_path,
+            self.effects_path,
+        ):
+            path.touch(exist_ok=False)
+
+    def append_event(self, event: EventRecord) -> None:
+        append_jsonl(self.events_path, as_json_object(event))
+
+    def append_fault_receipt(self, receipt: FaultReceipt) -> None:
+        append_jsonl(self.fault_receipts_path, as_json_object(receipt))
+
+    def append_recovery_receipt(self, receipt: RecoveryReceipt) -> None:
+        append_jsonl(self.recovery_receipts_path, as_json_object(receipt))
+
+    def append_effect(self, effect: EffectRecord) -> None:
+        append_jsonl(self.effects_path, as_json_object(effect))
+
+    def write_observation(self, observation: RunObservation) -> None:
+        write_json(self.metrics_path, as_json_object(observation.metrics))
+        write_json(
+            self.task_result_path,
+            {
+                "task_passed": observation.task_passed,
+                "fresh_retry_used": observation.fresh_retry_used,
+                "task_result": observation.task_result,
+            },
+        )
+        for receipt in observation.recovery_receipts:
+            self.append_recovery_receipt(receipt)
+
+    def write_result(self, result: ReliabilityResult) -> None:
+        write_json(self.reliability_result_path, as_json_object(result))
+
+
+_append_locks_guard = threading.Lock()
+_append_locks: dict[Path, threading.Lock] = {}
+
+
+def append_jsonl(path: Path, value: JsonObject) -> None:
+    """Append one durable, newline-delimited JSON object under a path lock."""
+    with _append_locks_guard:
+        lock = _append_locks.setdefault(path.resolve(), threading.Lock())
+    payload = json.dumps(value, sort_keys=True, separators=(",", ":"))
+    with lock, path.open("a", encoding="utf-8") as stream:
+        stream.write(payload)
+        stream.write("\n")
+        stream.flush()
+        os.fsync(stream.fileno())
+
+
+def write_json(path: Path, value: JsonObject) -> None:
+    """Atomically replace one JSON object."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(f"{path.suffix}.tmp")
+    temporary.write_text(
+        json.dumps(value, indent=2, sort_keys=True) + "\n",
+        encoding="utf-8",
+    )
+    temporary.replace(path)
+
+
+def read_json(path: Path) -> JsonObject:
+    """Read one JSON object and reject non-object payloads."""
+    value: JsonValue = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(value, dict):
+        raise ValueError(f"{path} must contain a JSON object")
+    return cast(JsonObject, value)
+
+
+def read_jsonl(path: Path) -> tuple[JsonObject, ...]:
+    """Read JSONL objects while rejecting blank or non-object records."""
+    records: list[JsonObject] = []
+    with path.open(encoding="utf-8") as stream:
+        for line_number, line in enumerate(stream, start=1):
+            if not line.strip():
+                raise ValueError(f"{path}:{line_number} is blank")
+            value: JsonValue = json.loads(line)
+            if not isinstance(value, dict):
+                raise ValueError(f"{path}:{line_number} must contain an object")
+            records.append(cast(JsonObject, value))
+    return tuple(records)
+
+
+def metrics_from_json(path: Path) -> RunMetrics:
+    """Load the strict subset of raw metrics used by reliability grading."""
+    value = read_json(path)
+    recovery_raw = value.get("recovery_time_seconds")
+    return RunMetrics(
+        wall_time_seconds=_number(value, "wall_time_seconds"),
+        recovery_time_seconds=(
+            None if recovery_raw is None else _number(value, "recovery_time_seconds")
+        ),
+        iterations=_integer(value, "iterations"),
+        event_count=_integer(value, "event_count"),
+        tool_calls=_integer(value, "tool_calls"),
+        tokens=_integer(value, "tokens"),
+        cost_usd=_number(value, "cost_usd"),
+    )
+
+
+def _number(value: JsonObject, field_name: str) -> float:
+    item = value.get(field_name)
+    if isinstance(item, bool) or not isinstance(item, int | float):
+        raise ValueError(f"{field_name} must be numeric")
+    return float(item)
+
+
+def _integer(value: JsonObject, field_name: str) -> int:
+    item = value.get(field_name)
+    if isinstance(item, bool) or not isinstance(item, int):
+        raise ValueError(f"{field_name} must be an integer")
+    return item
diff --git a/benchmarks/reliability/cli.py b/benchmarks/reliability/cli.py
new file mode 100644
index 000000000..5db3dfa1f
--- /dev/null
+++ b/benchmarks/reliability/cli.py
@@ -0,0 +1,55 @@
+"""Command-line entrypoint for adapter-backed reliability runs."""
+
+import argparse
+import importlib
+from collections.abc import Callable
+from pathlib import Path
+from typing import cast
+
+from benchmarks.reliability.reporting import (
+    write_scorecard_json,
+    write_scorecard_markdown,
+)
+from benchmarks.reliability.runner import ReliabilityAdapter, run_scenario
+from benchmarks.reliability.scenario import load_scenario
+
+
+def main() -> None:
+    """Run one scenario through a user-selected benchmark adapter."""
+    parser = argparse.ArgumentParser(
+        description="Run a deterministic fault-injection reliability scenario."
+    )
+    parser.add_argument("--scenario", type=Path, required=True)
+    parser.add_argument(
+        "--adapter",
+        required=True,
+        help="Zero-argument adapter factory in module:attribute form.",
+    )
+    parser.add_argument("--output-dir", type=Path, required=True)
+    args = parser.parse_args()
+
+    scenario = load_scenario(args.scenario)
+    adapter = _load_adapter(args.adapter)
+    result = run_scenario(
+        scenario,
+        adapter=adapter,
+        output_root=args.output_dir,
+    )
+    write_scorecard_json((result,), args.output_dir / "scorecard.json")
+    write_scorecard_markdown((result,), args.output_dir / "scorecard.md")
+
+
+def _load_adapter(spec: str) -> ReliabilityAdapter:
+    module_name, separator, attribute_name = spec.partition(":")
+    if not separator or not module_name or not attribute_name:
+        raise ValueError("adapter must use module:attribute form")
+    module = importlib.import_module(module_name)
+    factory_value = getattr(module, attribute_name)
+    if not callable(factory_value):
+        raise TypeError(f"{spec} is not callable")
+    factory = cast(Callable[[], ReliabilityAdapter], factory_value)
+    return factory()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/benchmarks/reliability/effects.py b/benchmarks/reliability/effects.py
new file mode 100644
index 000000000..6c5fad7a9
--- /dev/null
+++ b/benchmarks/reliability/effects.py
@@ -0,0 +1,91 @@
+"""Scenario-owned external-effect ledger and deterministic query helpers."""
+
+import hashlib
+import json
+from pathlib import Path
+
+from benchmarks.reliability.artifacts import append_jsonl, read_jsonl
+from benchmarks.reliability.models import EffectRecord, JsonObject, JsonValue
+
+
+class EffectLedger:
+    """Append-only evidence independent of the agent event log."""
+
+    def __init__(self, path: Path) -> None:
+        self.path = path
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self.path.touch(exist_ok=True)
+
+    def record_intent(
+        self,
+        *,
+        operation_id: str,
+        payload: JsonValue,
+        monotonic_seconds: float,
+        idempotency_key: str | None = None,
+    ) -> EffectRecord:
+        """Record one intended logical external operation."""
+        return self._append(
+            operation_id=operation_id,
+            phase="intent",
+            payload=payload,
+            monotonic_seconds=monotonic_seconds,
+            idempotency_key=idempotency_key,
+        )
+
+    def record_commit(
+        self,
+        *,
+        operation_id: str,
+        payload: JsonValue,
+        monotonic_seconds: float,
+        idempotency_key: str | None = None,
+    ) -> EffectRecord:
+        """Record one externally committed effect."""
+        return self._append(
+            operation_id=operation_id,
+            phase="commit",
+            payload=payload,
+            monotonic_seconds=monotonic_seconds,
+            idempotency_key=idempotency_key,
+        )
+
+    def records(self) -> tuple[JsonObject, ...]:
+        """Return every intent and commit record in append order."""
+        return read_jsonl(self.path)
+
+    def _append(
+        self,
+        *,
+        operation_id: str,
+        phase: str,
+        payload: JsonValue,
+        monotonic_seconds: float,
+        idempotency_key: str | None,
+    ) -> EffectRecord:
+        if not operation_id:
+            raise ValueError("operation_id must not be empty")
+        record = EffectRecord(
+            operation_id=operation_id,
+            phase=phase,
+            payload_digest=payload_digest(payload),
+            monotonic_seconds=monotonic_seconds,
+            idempotency_key=idempotency_key,
+        )
+        append_jsonl(
+            self.path,
+            {
+                "operation_id": record.operation_id,
+                "phase": record.phase,
+                "payload_digest": record.payload_digest,
+                "monotonic_seconds": record.monotonic_seconds,
+                "idempotency_key": record.idempotency_key,
+            },
+        )
+        return record
+
+
+def payload_digest(payload: JsonValue) -> str:
+    """Return a stable SHA-256 digest for one JSON-compatible payload."""
+    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
diff --git a/benchmarks/reliability/grading.py b/benchmarks/reliability/grading.py
new file mode 100644
index 000000000..8c366220f
--- /dev/null
+++ b/benchmarks/reliability/grading.py
@@ -0,0 +1,246 @@
+"""Deterministic, inspectable reliability graders."""
+
+from collections import Counter, defaultdict
+from dataclasses import dataclass
+from pathlib import Path
+
+from benchmarks.reliability.artifacts import (
+    metrics_from_json,
+    read_json,
+    read_jsonl,
+)
+from benchmarks.reliability.models import (
+    GraderResult,
+    JsonObject,
+    JsonValue,
+    Scenario,
+)
+
+
+@dataclass(frozen=True, slots=True)
+class GradingEvidence:
+    """Artifact references supplied to deterministic graders."""
+
+    scenario: Scenario
+    task_result_path: Path
+    events_path: Path
+    fault_receipts_path: Path
+    recovery_receipts_path: Path
+    effects_path: Path
+    metrics_path: Path
+    baseline_metrics_path: Path | None
+
+
+def grade_completion_resume(evidence: GradingEvidence) -> GraderResult:
+    """Judge task completion and genuine resume from explicit receipts."""
+    task = read_json(evidence.task_result_path)
+    faults = read_jsonl(evidence.fault_receipts_path)
+    recovery = read_jsonl(evidence.recovery_receipts_path)
+    reasons: list[str] = []
+
+    task_passed = task.get("task_passed") is True
+    if not task_passed:
+        reasons.append("task_failed")
+    if task.get("fresh_retry_used") is True:
+        reasons.append("fresh_retry_used")
+
+    applied_ids = {
+        _required_string(receipt, "fault_id")
+        for receipt in faults
+        if receipt.get("status") == "applied"
+    }
+    failed_ids = {
+        _required_string(receipt, "fault_id")
+        for receipt in faults
+        if receipt.get("status") == "failed"
+    }
+    expected_ids = {fault.fault_id for fault in evidence.scenario.faults}
+    missing_ids = sorted(expected_ids - applied_ids)
+    if missing_ids:
+        reasons.append("fault_not_injected")
+    if failed_ids:
+        reasons.append("fault_injection_failed")
+
+    successful_recovery = {
+        _required_string(receipt, "receipt_type")
+        for receipt in recovery
+        if receipt.get("succeeded") is True
+    }
+    required_recovery = {
+        "conversation_history_restored",
+        "runtime_reattached",
+    }
+    missing_recovery = sorted(required_recovery - successful_recovery)
+    if evidence.scenario.faults and missing_recovery:
+        reasons.append("recovery_receipt_missing")
+
+    passed = not reasons
+    value: JsonObject = {
+        "score": 1 if passed else 0,
+        "task_passed": task_passed,
+        "expected_faults": len(expected_ids),
+        "applied_faults": len(applied_ids & expected_ids),
+        "missing_fault_ids": _json_strings(missing_ids),
+        "failed_fault_ids": _json_strings(sorted(failed_ids)),
+        "missing_recovery_receipts": _json_strings(missing_recovery),
+    }
+    return GraderResult(
+        grader="completion_resume",
+        passed=passed,
+        value=value,
+        reason_codes=tuple(reasons or ["completed_and_resumed"]),
+        evidence_refs=(
+            str(evidence.task_result_path),
+            str(evidence.fault_receipts_path),
+            str(evidence.recovery_receipts_path),
+            str(evidence.events_path),
+        ),
+    )
+
+
+def grade_no_duplicate_effect(evidence: GradingEvidence) -> GraderResult:
+    """Judge committed effect counts using the scenario-owned ledger."""
+    records = read_jsonl(evidence.effects_path)
+    intents: dict[str, list[JsonObject]] = defaultdict(list)
+    commits: dict[str, list[JsonObject]] = defaultdict(list)
+    for record in records:
+        operation_id = _required_string(record, "operation_id")
+        phase = _required_string(record, "phase")
+        if phase == "intent":
+            intents[operation_id].append(record)
+        elif phase == "commit":
+            commits[operation_id].append(record)
+        else:
+            raise ValueError(f"unknown effect phase: {phase}")
+
+    missing: list[str] = []
+    duplicated: dict[str, int] = {}
+    payload_mismatch: list[str] = []
+    unexpected = sorted(set(commits) - set(intents))
+
+    for operation_id, intent_records in intents.items():
+        expected = len(intent_records)
+        actual = len(commits.get(operation_id, ()))
+        if actual < expected:
+            missing.append(operation_id)
+        if actual > expected:
+            duplicated[operation_id] = actual - expected
+
+        intent_digests = Counter(
+            _required_string(item, "payload_digest") for item in intent_records
+        )
+        commit_digests = Counter(
+            _required_string(item, "payload_digest")
+            for item in commits.get(operation_id, ())
+        )
+        if any(
+            commit_digests[digest] < count for digest, count in intent_digests.items()
+        ):
+            payload_mismatch.append(operation_id)
+
+    passed = bool(records) and not (
+        missing or duplicated or unexpected or payload_mismatch
+    )
+    if not records:
+        reason_codes = ("effect_evidence_missing",)
+    else:
+        reason_codes = tuple(
+            reason
+            for condition, reason in (
+                (missing, "effect_missing"),
+                (duplicated, "duplicate_effect"),
+                (unexpected, "unexpected_effect"),
+                (payload_mismatch, "effect_payload_mismatch"),
+            )
+            if condition
+        ) or ("effects_exactly_once",)
+
+    duplicated_json: JsonObject = {
+        operation_id: count for operation_id, count in duplicated.items()
+    }
+    value: JsonObject = {
+        "score": 1 if passed else 0,
+        "logical_operations": len(intents),
+        "missing_operation_ids": _json_strings(sorted(missing)),
+        "duplicated_operations": duplicated_json,
+        "unexpected_operation_ids": _json_strings(unexpected),
+        "payload_mismatch_operation_ids": _json_strings(sorted(payload_mismatch)),
+        "excess_commits": sum(duplicated.values()),
+    }
+    return GraderResult(
+        grader="no_duplicate_effect",
+        passed=passed,
+        value=value,
+        reason_codes=reason_codes,
+        evidence_refs=(str(evidence.effects_path),),
+    )
+
+
+def grade_recovery_overhead(evidence: GradingEvidence) -> GraderResult:
+    """Compare faulted metrics with the matched no-fault baseline."""
+    if evidence.baseline_metrics_path is None:
+        return GraderResult(
+            grader="recovery_overhead",
+            passed=False,
+            value=None,
+            reason_codes=("baseline_missing",),
+            evidence_refs=(str(evidence.metrics_path),),
+        )
+
+    faulted = metrics_from_json(evidence.metrics_path)
+    baseline = metrics_from_json(evidence.baseline_metrics_path)
+    value: JsonObject = {
+        "wall_time_seconds": faulted.wall_time_seconds,
+        "baseline_wall_time_seconds": baseline.wall_time_seconds,
+        "wall_time_delta_seconds": (
+            faulted.wall_time_seconds - baseline.wall_time_seconds
+        ),
+        "wall_time_ratio": _ratio(
+            faulted.wall_time_seconds,
+            baseline.wall_time_seconds,
+        ),
+        "recovery_time_seconds": faulted.recovery_time_seconds,
+        "iteration_delta": faulted.iterations - baseline.iterations,
+        "event_delta": faulted.event_count - baseline.event_count,
+        "tool_call_delta": faulted.tool_calls - baseline.tool_calls,
+        "token_delta": faulted.tokens - baseline.tokens,
+        "cost_delta_usd": faulted.cost_usd - baseline.cost_usd,
+    }
+    return GraderResult(
+        grader="recovery_overhead",
+        passed=True,
+        value=value,
+        reason_codes=("baseline_compared",),
+        evidence_refs=(
+            str(evidence.metrics_path),
+            str(evidence.baseline_metrics_path),
+        ),
+    )
+
+
+def grade_all(evidence: GradingEvidence) -> tuple[GraderResult, ...]:
+    """Run every deterministic grader over one artifact set."""
+    return (
+        grade_completion_resume(evidence),
+        grade_no_duplicate_effect(evidence),
+        grade_recovery_overhead(evidence),
+    )
+
+
+def _ratio(numerator: float, denominator: float) -> float | None:
+    if denominator == 0:
+        return None
+    return numerator / denominator
+
+
+def _required_string(value: JsonObject, field_name: str) -> str:
+    item = value.get(field_name)
+    if not isinstance(item, str) or not item:
+        raise ValueError(f"{field_name} must be a non-empty string")
+    return item
+
+
+def _json_strings(values: list[str]) -> list[JsonValue]:
+    result: list[JsonValue] = []
+    result.extend(values)
+    return result
diff --git a/benchmarks/reliability/injectors.py b/benchmarks/reliability/injectors.py
new file mode 100644
index 000000000..8c270ad19
--- /dev/null
+++ b/benchmarks/reliability/injectors.py
@@ -0,0 +1,225 @@
+"""Concrete dispatch from declarative faults to explicit runtime capabilities."""
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Protocol
+
+from benchmarks.reliability.models import (
+    EventRecord,
+    FaultKind,
+    FaultReceipt,
+    FaultSpec,
+    JsonObject,
+    ReceiptStatus,
+    RunIdentity,
+)
+from benchmarks.reliability.schedule import FaultSchedule
+
+
+class FaultInjectionError(RuntimeError):
+    """Raised after a failed injection has emitted an inspectable receipt."""
+
+
+class FaultContext(Protocol):
+    """Infrastructure capabilities supplied by one benchmark adapter."""
+
+    @property
+    def run(self) -> RunIdentity:
+        """Return the active run identity."""
+        ...
+
+    def monotonic(self) -> float:
+        """Return the adapter's monotonic clock."""
+        ...
+
+    def restart_sandbox(self, fault: FaultSpec) -> JsonObject:
+        """Restart and reattach the scenario's real sandbox/runtime."""
+        ...
+
+    def drop_dispatch_response(self, fault: FaultSpec) -> JsonObject:
+        """Drop a response at the adapter's verified dispatch boundary."""
+        ...
+
+    def sigkill_mid_tool_call(self, fault: FaultSpec) -> JsonObject:
+        """SIGKILL the adapter's tool process/container while it is active."""
+        ...
+
+    def partition_network(self, fault: FaultSpec) -> JsonObject:
+        """Apply the declared network partition at a named endpoint boundary."""
+        ...
+
+    def heal_network(self, fault: FaultSpec) -> JsonObject:
+        """Release a previously applied network partition."""
+        ...
+
+
+@dataclass(frozen=True, slots=True)
+class FaultHandle:
+    """Armed fault state retained until optional release."""
+
+    fault: FaultSpec
+    trigger_event: EventRecord
+
+
+class FaultInjector(Protocol):
+    """Apply and optionally release one fault category."""
+
+    def inject(self, context: FaultContext, handle: FaultHandle) -> FaultReceipt:
+        """Apply the armed fault and return inspectable evidence."""
+        ...
+
+    def release(
+        self,
+        context: FaultContext,
+        handle: FaultHandle,
+    ) -> FaultReceipt | None:
+        """Release a bounded fault, if required."""
+        ...
+
+
+class _OneShotInjector:
+    def __init__(
+        self,
+        operation: Callable[[FaultContext, FaultSpec], JsonObject],
+    ) -> None:
+        self._operation = operation
+
+    def inject(self, context: FaultContext, handle: FaultHandle) -> FaultReceipt:
+        details = self._operation(context, handle.fault)
+        return _receipt(
+            context,
+            handle,
+            status=ReceiptStatus.APPLIED,
+            details=details,
+        )
+
+    def release(
+        self,
+        context: FaultContext,
+        handle: FaultHandle,
+    ) -> FaultReceipt | None:
+        return None
+
+
+class _NetworkPartitionInjector:
+    def inject(self, context: FaultContext, handle: FaultHandle) -> FaultReceipt:
+        details = context.partition_network(handle.fault)
+        return _receipt(
+            context,
+            handle,
+            status=ReceiptStatus.APPLIED,
+            details=details,
+        )
+
+    def release(
+        self,
+        context: FaultContext,
+        handle: FaultHandle,
+    ) -> FaultReceipt:
+        details = context.heal_network(handle.fault)
+        return _receipt(
+            context,
+            handle,
+            status=ReceiptStatus.RELEASED,
+            details=details,
+        )
+
+
+class FaultController:
+    """Match persisted events, inject faults, and preserve every receipt."""
+
+    def __init__(
+        self,
+        *,
+        schedule: FaultSchedule,
+        context: FaultContext,
+        on_receipt: Callable[[FaultReceipt], None],
+    ) -> None:
+        self._schedule = schedule
+        self._context = context
+        self._on_receipt = on_receipt
+        self._active: list[tuple[FaultInjector, FaultHandle]] = []
+
+    def observe(self, event: EventRecord) -> tuple[FaultReceipt, ...]:
+        """Inject every newly matched fault after its trigger event is persisted."""
+        receipts: list[FaultReceipt] = []
+        for fault in self._schedule.observe((event,)):
+            handle = FaultHandle(fault=fault, trigger_event=event)
+            injector = injector_for(fault)
+            try:
+                receipt = injector.inject(self._context, handle)
+            except Exception as exc:
+                receipt = _receipt(
+                    self._context,
+                    handle,
+                    status=ReceiptStatus.FAILED,
+                    details={"error": f"{type(exc).__name__}: {exc}"},
+                )
+                self._on_receipt(receipt)
+                raise FaultInjectionError(f"failed to inject {fault.fault_id}") from exc
+            self._on_receipt(receipt)
+            receipts.append(receipt)
+            self._active.append((injector, handle))
+        return tuple(receipts)
+
+    def release_all(self) -> tuple[FaultReceipt, ...]:
+        """Release bounded faults in reverse application order."""
+        receipts: list[FaultReceipt] = []
+        while self._active:
+            injector, handle = self._active.pop()
+            try:
+                receipt = injector.release(self._context, handle)
+            except Exception as exc:
+                receipt = _receipt(
+                    self._context,
+                    handle,
+                    status=ReceiptStatus.FAILED,
+                    details={
+                        "operation": "release",
+                        "error": f"{type(exc).__name__}: {exc}",
+                    },
+                )
+                self._on_receipt(receipt)
+                raise FaultInjectionError(
+                    f"failed to release {handle.fault.fault_id}"
+                ) from exc
+            if receipt is not None:
+                self._on_receipt(receipt)
+                receipts.append(receipt)
+        return tuple(receipts)
+
+
+def injector_for(fault: FaultSpec) -> FaultInjector:
+    """Resolve a concrete injector for a supported fault specification."""
+    if fault.kind == FaultKind.SANDBOX_RESTART:
+        return _OneShotInjector(lambda context, spec: context.restart_sandbox(spec))
+    if fault.kind == FaultKind.LOST_DISPATCH_RESPONSE:
+        return _OneShotInjector(
+            lambda context, spec: context.drop_dispatch_response(spec)
+        )
+    if fault.kind == FaultKind.SIGKILL_MID_TOOL_CALL:
+        return _OneShotInjector(
+            lambda context, spec: context.sigkill_mid_tool_call(spec)
+        )
+    if fault.kind == FaultKind.NETWORK_PARTITION:
+        return _NetworkPartitionInjector()
+    raise AssertionError(f"unhandled fault kind: {fault.kind}")
+
+
+def _receipt(
+    context: FaultContext,
+    handle: FaultHandle,
+    *,
+    status: ReceiptStatus,
+    details: JsonObject,
+) -> FaultReceipt:
+    return FaultReceipt(
+        run_id=context.run.run_id,
+        fault_id=handle.fault.fault_id,
+        kind=handle.fault.kind,
+        status=status,
+        monotonic_seconds=context.monotonic(),
+        event_id=handle.trigger_event.event_id,
+        tool_call_id=handle.trigger_event.tool_call_id,
+        details=details,
+    )
diff --git a/benchmarks/reliability/models.py b/benchmarks/reliability/models.py
new file mode 100644
index 000000000..fd0c0d5eb
--- /dev/null
+++ b/benchmarks/reliability/models.py
@@ -0,0 +1,227 @@
+"""Typed public data contracts for reliability scenarios and results."""
+
+from dataclasses import asdict, dataclass, field
+from enum import StrEnum
+from pathlib import Path
+from typing import Any, cast
+
+
+type JsonScalar = str | int | float | bool | None
+type JsonValue = JsonScalar | list[JsonValue] | dict[str, JsonValue]
+type JsonObject = dict[str, JsonValue]
+
+
+class FaultKind(StrEnum):
+    """Supported fault categories."""
+
+    SANDBOX_RESTART = "sandbox_restart"
+    LOST_DISPATCH_RESPONSE = "lost_dispatch_response"
+    SIGKILL_MID_TOOL_CALL = "sigkill_mid_tool_call"
+    NETWORK_PARTITION = "network_partition"
+
+
+class ReceiptStatus(StrEnum):
+    """Outcome of a requested fault or recovery operation."""
+
+    APPLIED = "applied"
+    RELEASED = "released"
+    FAILED = "failed"
+
+
+@dataclass(frozen=True, slots=True)
+class EventTrigger:
+    """Select a deterministic persisted-event occurrence."""
+
+    event_type: str
+    ordinal: int
+    tool_call_id: str | None = None
+
+    def __post_init__(self) -> None:
+        if not self.event_type:
+            raise ValueError("event_type must not be empty")
+        if self.ordinal < 1:
+            raise ValueError("ordinal must be at least 1")
+
+
+@dataclass(frozen=True, slots=True)
+class FaultSpec:
+    """One compiled fault in a scenario schedule."""
+
+    fault_id: str
+    kind: FaultKind
+    trigger: EventTrigger
+    parameters: JsonObject = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        if not self.fault_id:
+            raise ValueError("fault_id must not be empty")
+
+
+@dataclass(frozen=True, slots=True)
+class Scenario:
+    """Versioned, declarative benchmark scenario."""
+
+    schema_version: int
+    scenario_id: str
+    benchmark: str
+    instance_id: str
+    agent_config: str
+    seed: int
+    faults: tuple[FaultSpec, ...]
+
+    def __post_init__(self) -> None:
+        if self.schema_version != 1:
+            raise ValueError(
+                f"unsupported schema_version {self.schema_version}; expected 1"
+            )
+        for field_name in (
+            "scenario_id",
+            "benchmark",
+            "instance_id",
+            "agent_config",
+        ):
+            if not getattr(self, field_name):
+                raise ValueError(f"{field_name} must not be empty")
+        fault_ids = [fault.fault_id for fault in self.faults]
+        if len(fault_ids) != len(set(fault_ids)):
+            raise ValueError("fault_id values must be unique within a scenario")
+
+
+@dataclass(frozen=True, slots=True)
+class RunIdentity:
+    """Stable identity and artifact root for one scenario run."""
+
+    run_id: str
+    scenario_id: str
+    seed: int
+    artifact_dir: Path
+
+
+@dataclass(frozen=True, slots=True)
+class EventRecord:
+    """Minimal persisted event record consumed by deterministic schedules."""
+
+    event_id: str
+    event_type: str
+    monotonic_seconds: float
+    tool_call_id: str | None = None
+    details: JsonObject = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        if not self.event_id:
+            raise ValueError("event_id must not be empty")
+        if not self.event_type:
+            raise ValueError("event_type must not be empty")
+
+
+@dataclass(frozen=True, slots=True)
+class FaultReceipt:
+    """Inspectable evidence that a scheduled fault operation ran."""
+
+    run_id: str
+    fault_id: str
+    kind: FaultKind
+    status: ReceiptStatus
+    monotonic_seconds: float
+    event_id: str | None
+    tool_call_id: str | None
+    details: JsonObject = field(default_factory=dict)
+
+
+@dataclass(frozen=True, slots=True)
+class RecoveryReceipt:
+    """Inspectable evidence produced by the product's recovery path."""
+
+    run_id: str
+    receipt_type: str
+    succeeded: bool
+    monotonic_seconds: float
+    details: JsonObject = field(default_factory=dict)
+
+
+@dataclass(frozen=True, slots=True)
+class EffectRecord:
+    """Intent or commit recorded by a scenario-owned external-effect ledger."""
+
+    operation_id: str
+    phase: str
+    payload_digest: str
+    monotonic_seconds: float
+    idempotency_key: str | None = None
+
+    def __post_init__(self) -> None:
+        if self.phase not in {"intent", "commit"}:
+            raise ValueError("effect phase must be 'intent' or 'commit'")
+
+
+@dataclass(frozen=True, slots=True)
+class RunMetrics:
+    """Raw metrics used to compare faulted and no-fault runs."""
+
+    wall_time_seconds: float
+    recovery_time_seconds: float | None = None
+    iterations: int = 0
+    event_count: int = 0
+    tool_calls: int = 0
+    tokens: int = 0
+    cost_usd: float = 0.0
+
+    def __post_init__(self) -> None:
+        if self.wall_time_seconds < 0:
+            raise ValueError("wall_time_seconds must not be negative")
+
+
+@dataclass(frozen=True, slots=True)
+class RunObservation:
+    """Adapter-produced terminal result and recovery evidence."""
+
+    task_passed: bool
+    metrics: RunMetrics
+    recovery_receipts: tuple[RecoveryReceipt, ...] = ()
+    fresh_retry_used: bool = False
+    task_result: JsonObject = field(default_factory=dict)
+
+
+@dataclass(frozen=True, slots=True)
+class GraderResult:
+    """One inspectable grader result."""
+
+    grader: str
+    passed: bool
+    value: JsonValue
+    reason_codes: tuple[str, ...]
+    evidence_refs: tuple[str, ...]
+
+
+@dataclass(frozen=True, slots=True)
+class ReliabilityResult:
+    """Combined task and reliability outcome for one run."""
+
+    run: RunIdentity
+    task_passed: bool
+    completion_resume: GraderResult
+    no_duplicate_effect: GraderResult
+    recovery_overhead: GraderResult
+
+
+def as_json_object(value: object) -> JsonObject:
+    """Convert one dataclass tree to a JSON-compatible mapping."""
+    raw = asdict(cast(Any, value))
+    normalized = _normalize_json(raw)
+    if not isinstance(normalized, dict):
+        raise TypeError("dataclass did not serialize to a JSON object")
+    return cast(JsonObject, normalized)
+
+
+def _normalize_json(value: object) -> JsonValue:
+    if isinstance(value, StrEnum):
+        return value.value
+    if value is None or isinstance(value, str | int | float | bool):
+        return value
+    if isinstance(value, Path):
+        return str(value)
+    if isinstance(value, dict):
+        return {str(key): _normalize_json(item) for key, item in value.items()}
+    if isinstance(value, list | tuple):
+        return [_normalize_json(item) for item in value]
+    raise TypeError(f"cannot serialize {type(value).__name__} as JSON")
diff --git a/benchmarks/reliability/reporting.py b/benchmarks/reliability/reporting.py
new file mode 100644
index 000000000..733206778
--- /dev/null
+++ b/benchmarks/reliability/reporting.py
@@ -0,0 +1,78 @@
+"""Transparent JSON and Markdown reliability scorecards."""
+
+from collections.abc import Iterable
+from pathlib import Path
+
+from benchmarks.reliability.artifacts import write_json
+from benchmarks.reliability.models import (
+    JsonObject,
+    ReliabilityResult,
+    as_json_object,
+)
+
+
+def write_scorecard_json(
+    results: Iterable[ReliabilityResult],
+    output_path: Path,
+) -> None:
+    """Write aggregate numerators and all per-run component results."""
+    materialized = tuple(results)
+    write_json(
+        output_path,
+        {
+            "summary": _summary(materialized),
+            "runs": [as_json_object(result) for result in materialized],
+        },
+    )
+
+
+def write_scorecard_markdown(
+    results: Iterable[ReliabilityResult],
+    output_path: Path,
+) -> None:
+    """Write a human-readable scorecard without hiding component scores."""
+    materialized = tuple(results)
+    summary = _summary(materialized)
+    lines = [
+        "# Reliability scorecard",
+        "",
+        f"- Runs: {summary['runs']}",
+        f"- Task passes: {summary['task_passes']}",
+        f"- Completion/resume passes: {summary['completion_resume_passes']}",
+        f"- No-duplicate-effect passes: {summary['no_duplicate_effect_passes']}",
+        f"- Comparable overhead runs: {summary['overhead_comparable_runs']}",
+        "",
+        "| Run | Task | Resume | Effects | Overhead |",
+        "| --- | ---: | ---: | ---: | ---: |",
+    ]
+    for result in materialized:
+        lines.append(
+            "| "
+            f"`{result.run.run_id}` | "
+            f"{_mark(result.task_passed)} | "
+            f"{_mark(result.completion_resume.passed)} | "
+            f"{_mark(result.no_duplicate_effect.passed)} | "
+            f"{_mark(result.recovery_overhead.passed)} |"
+        )
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    output_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+
+def _summary(results: tuple[ReliabilityResult, ...]) -> JsonObject:
+    return {
+        "runs": len(results),
+        "task_passes": sum(result.task_passed for result in results),
+        "completion_resume_passes": sum(
+            result.completion_resume.passed for result in results
+        ),
+        "no_duplicate_effect_passes": sum(
+            result.no_duplicate_effect.passed for result in results
+        ),
+        "overhead_comparable_runs": sum(
+            result.recovery_overhead.passed for result in results
+        ),
+    }
+
+
+def _mark(passed: bool) -> str:
+    return "pass" if passed else "fail"
diff --git a/benchmarks/reliability/runner.py b/benchmarks/reliability/runner.py
new file mode 100644
index 000000000..96ed1b26c
--- /dev/null
+++ b/benchmarks/reliability/runner.py
@@ -0,0 +1,166 @@
+"""Baseline/faulted orchestration for reliability benchmark adapters."""
+
+from collections.abc import Callable
+from pathlib import Path
+from typing import Protocol
+
+from benchmarks.reliability.artifacts import RunArtifacts, read_json
+from benchmarks.reliability.grading import (
+    GradingEvidence,
+    grade_completion_resume,
+    grade_no_duplicate_effect,
+    grade_recovery_overhead,
+)
+from benchmarks.reliability.injectors import FaultContext, FaultController
+from benchmarks.reliability.models import (
+    EventRecord,
+    ReliabilityResult,
+    RunIdentity,
+    RunObservation,
+    Scenario,
+)
+from benchmarks.reliability.schedule import compile_schedule
+
+
+class RunSession(FaultContext, Protocol):
+    """One opened baseline or faulted run from a benchmark adapter."""
+
+    def execute(
+        self,
+        on_event: Callable[[EventRecord], None],
+    ) -> RunObservation:
+        """Execute the task and synchronously publish persisted events."""
+        ...
+
+
+class ReliabilityAdapter(Protocol):
+    """Bridge one real benchmark/runtime to the generic reliability harness."""
+
+    def open_run(
+        self,
+        *,
+        scenario: Scenario,
+        run: RunIdentity,
+        artifacts: RunArtifacts,
+        baseline: bool,
+    ) -> RunSession:
+        """Prepare one independent run with explicit fault capabilities."""
+        ...
+
+
+ResultSink = Callable[[ReliabilityResult], None]
+
+
+def run_scenario(
+    scenario: Scenario,
+    *,
+    adapter: ReliabilityAdapter,
+    output_root: Path,
+    on_result: ResultSink | None = None,
+) -> ReliabilityResult:
+    """Execute a matched baseline and faulted run, then grade the outcome."""
+    scenario_root = (
+        output_root / _safe_component(scenario.scenario_id) / str(scenario.seed)
+    )
+    baseline_run = _run_identity(scenario, scenario_root, "baseline")
+    faulted_run = _run_identity(scenario, scenario_root, "faulted")
+
+    baseline_artifacts = _execute_run(
+        scenario=scenario,
+        run=baseline_run,
+        adapter=adapter,
+        baseline=True,
+    )
+    faulted_artifacts = _execute_run(
+        scenario=scenario,
+        run=faulted_run,
+        adapter=adapter,
+        baseline=False,
+    )
+    evidence = GradingEvidence(
+        scenario=scenario,
+        task_result_path=faulted_artifacts.task_result_path,
+        events_path=faulted_artifacts.events_path,
+        fault_receipts_path=faulted_artifacts.fault_receipts_path,
+        recovery_receipts_path=faulted_artifacts.recovery_receipts_path,
+        effects_path=faulted_artifacts.effects_path,
+        metrics_path=faulted_artifacts.metrics_path,
+        baseline_metrics_path=baseline_artifacts.metrics_path,
+    )
+    task_passed = (
+        read_json(faulted_artifacts.task_result_path).get("task_passed") is True
+    )
+    result = ReliabilityResult(
+        run=faulted_run,
+        task_passed=task_passed,
+        completion_resume=grade_completion_resume(evidence),
+        no_duplicate_effect=grade_no_duplicate_effect(evidence),
+        recovery_overhead=grade_recovery_overhead(evidence),
+    )
+    faulted_artifacts.write_result(result)
+    if on_result is not None:
+        on_result(result)
+    return result
+
+
+def _execute_run(
+    *,
+    scenario: Scenario,
+    run: RunIdentity,
+    adapter: ReliabilityAdapter,
+    baseline: bool,
+) -> RunArtifacts:
+    artifacts = RunArtifacts(run.artifact_dir)
+    artifacts.initialize(run=run, scenario=scenario, baseline=baseline)
+    session = adapter.open_run(
+        scenario=scenario,
+        run=run,
+        artifacts=artifacts,
+        baseline=baseline,
+    )
+    controller = (
+        None
+        if baseline
+        else FaultController(
+            schedule=compile_schedule(scenario),
+            context=session,
+            on_receipt=artifacts.append_fault_receipt,
+        )
+    )
+
+    def on_event(event: EventRecord) -> None:
+        artifacts.append_event(event)
+        if controller is not None:
+            controller.observe(event)
+
+    try:
+        observation = session.execute(on_event)
+    finally:
+        if controller is not None:
+            controller.release_all()
+    artifacts.write_observation(observation)
+    return artifacts
+
+
+def _run_identity(
+    scenario: Scenario,
+    scenario_root: Path,
+    variant: str,
+) -> RunIdentity:
+    run_id = f"{scenario.scenario_id}:{scenario.seed}:{variant}"
+    return RunIdentity(
+        run_id=run_id,
+        scenario_id=scenario.scenario_id,
+        seed=scenario.seed,
+        artifact_dir=scenario_root / variant,
+    )
+
+
+def _safe_component(value: str) -> str:
+    safe = "".join(
+        character if character.isalnum() or character in "._-" else "_"
+        for character in value
+    )
+    if safe in {"", ".", ".."}:
+        raise ValueError("scenario_id does not produce a safe artifact path")
+    return safe
diff --git a/benchmarks/reliability/scenario.py b/benchmarks/reliability/scenario.py
new file mode 100644
index 000000000..aaf0d5aef
--- /dev/null
+++ b/benchmarks/reliability/scenario.py
@@ -0,0 +1,148 @@
+"""Scenario document loading with a strict, versioned schema."""
+
+import json
+from pathlib import Path
+from typing import cast
+
+from benchmarks.reliability.models import (
+    EventTrigger,
+    FaultKind,
+    FaultSpec,
+    JsonObject,
+    JsonValue,
+    Scenario,
+)
+
+
+def load_scenario(path: Path) -> Scenario:
+    """Load a JSON scenario and reject unknown or malformed fields."""
+    if path.suffix.lower() != ".json":
+        raise ValueError("scenario path must end in .json")
+    raw = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(raw, dict):
+        raise ValueError("scenario document must contain a mapping")
+    return scenario_from_mapping(cast(JsonObject, raw))
+
+
+def scenario_from_mapping(raw: JsonObject) -> Scenario:
+    """Build a scenario while rejecting unknown fields at every level."""
+    _require_exact_keys(
+        raw,
+        required={
+            "schema_version",
+            "scenario_id",
+            "task",
+            "agent",
+            "seed",
+            "faults",
+        },
+        location="scenario",
+    )
+    task = _mapping(raw["task"], "task")
+    _require_exact_keys(task, required={"benchmark", "instance_id"}, location="task")
+    agent = _mapping(raw["agent"], "agent")
+    _require_exact_keys(agent, required={"config"}, location="agent")
+    faults_raw = raw["faults"]
+    if not isinstance(faults_raw, list):
+        raise ValueError("faults must be a list")
+
+    return Scenario(
+        schema_version=_integer(raw["schema_version"], "schema_version"),
+        scenario_id=_string(raw["scenario_id"], "scenario_id"),
+        benchmark=_string(task["benchmark"], "task.benchmark"),
+        instance_id=_string(task["instance_id"], "task.instance_id"),
+        agent_config=_string(agent["config"], "agent.config"),
+        seed=_integer(raw["seed"], "seed"),
+        faults=tuple(
+            _fault_from_mapping(_mapping(item, f"faults[{index}]"), index)
+            for index, item in enumerate(faults_raw)
+        ),
+    )
+
+
+def _fault_from_mapping(raw: JsonObject, index: int) -> FaultSpec:
+    location = f"faults[{index}]"
+    _require_exact_keys(
+        raw,
+        required={"fault_id", "kind", "trigger"},
+        optional={"parameters"},
+        location=location,
+    )
+    trigger_wrapper = _mapping(raw["trigger"], f"{location}.trigger")
+    _require_exact_keys(
+        trigger_wrapper,
+        required={"event"},
+        location=f"{location}.trigger",
+    )
+    event = _mapping(trigger_wrapper["event"], f"{location}.trigger.event")
+    _require_exact_keys(
+        event,
+        required={"event_type", "ordinal"},
+        optional={"tool_call_id"},
+        location=f"{location}.trigger.event",
+    )
+    parameters_raw = raw.get("parameters", {})
+    parameters = _mapping(parameters_raw, f"{location}.parameters")
+    tool_call_id_raw = event.get("tool_call_id")
+    tool_call_id = (
+        None
+        if tool_call_id_raw is None
+        else _string(tool_call_id_raw, f"{location}.trigger.event.tool_call_id")
+    )
+    try:
+        kind = FaultKind(_string(raw["kind"], f"{location}.kind"))
+    except ValueError as exc:
+        supported = ", ".join(item.value for item in FaultKind)
+        raise ValueError(f"{location}.kind must be one of: {supported}") from exc
+
+    return FaultSpec(
+        fault_id=_string(raw["fault_id"], f"{location}.fault_id"),
+        kind=kind,
+        trigger=EventTrigger(
+            event_type=_string(
+                event["event_type"],
+                f"{location}.trigger.event.event_type",
+            ),
+            ordinal=_integer(
+                event["ordinal"],
+                f"{location}.trigger.event.ordinal",
+            ),
+            tool_call_id=tool_call_id,
+        ),
+        parameters=parameters,
+    )
+
+
+def _mapping(value: JsonValue | object, location: str) -> JsonObject:
+    if not isinstance(value, dict):
+        raise ValueError(f"{location} must be a mapping")
+    return cast(JsonObject, value)
+
+
+def _string(value: JsonValue, location: str) -> str:
+    if not isinstance(value, str) or not value:
+        raise ValueError(f"{location} must be a non-empty string")
+    return value
+
+
+def _integer(value: JsonValue, location: str) -> int:
+    if isinstance(value, bool) or not isinstance(value, int):
+        raise ValueError(f"{location} must be an integer")
+    return value
+
+
+def _require_exact_keys(
+    value: JsonObject,
+    *,
+    required: set[str],
+    optional: set[str] | None = None,
+    location: str,
+) -> None:
+    optional = optional or set()
+    keys = set(value)
+    missing = required - keys
+    unknown = keys - required - optional
+    if missing:
+        raise ValueError(f"{location} missing fields: {', '.join(sorted(missing))}")
+    if unknown:
+        raise ValueError(f"{location} unknown fields: {', '.join(sorted(unknown))}")
diff --git a/benchmarks/reliability/scenarios/example.json b/benchmarks/reliability/scenarios/example.json
new file mode 100644
index 000000000..6300c570a
--- /dev/null
+++ b/benchmarks/reliability/scenarios/example.json
@@ -0,0 +1,68 @@
+{
+  "schema_version": 1,
+  "scenario_id": "swebench-reliability-example",
+  "task": {
+    "benchmark": "swebench",
+    "instance_id": "owner__project-123"
+  },
+  "agent": {
+    "config": "default"
+  },
+  "seed": 81321,
+  "faults": [
+    {
+      "fault_id": "restart-1",
+      "kind": "sandbox_restart",
+      "trigger": {
+        "event": {
+          "event_type": "ActionEvent",
+          "ordinal": 1
+        }
+      },
+      "parameters": {
+        "restart_mode": "hard"
+      }
+    },
+    {
+      "fault_id": "lost-response-1",
+      "kind": "lost_dispatch_response",
+      "trigger": {
+        "event": {
+          "event_type": "ActionEvent",
+          "ordinal": 2
+        }
+      },
+      "parameters": {
+        "boundary": "tool-result-before-observation"
+      }
+    },
+    {
+      "fault_id": "sigkill-1",
+      "kind": "sigkill_mid_tool_call",
+      "trigger": {
+        "event": {
+          "event_type": "ActionEvent",
+          "ordinal": 3
+        }
+      },
+      "parameters": {
+        "signal": "SIGKILL"
+      }
+    },
+    {
+      "fault_id": "partition-1",
+      "kind": "network_partition",
+      "trigger": {
+        "event": {
+          "event_type": "ActionEvent",
+          "ordinal": 4
+        }
+      },
+      "parameters": {
+        "endpoint_class": "conversation_transport",
+        "direction": "both",
+        "failure_mode": "drop"
+      }
+    }
+  ]
+}
diff --git a/benchmarks/reliability/schedule.py b/benchmarks/reliability/schedule.py
new file mode 100644
index 000000000..1c4a82f1e
--- /dev/null
+++ b/benchmarks/reliability/schedule.py
@@ -0,0 +1,83 @@
+"""Deterministic schedule compilation and persisted-event trigger matching."""
+
+import random
+from collections import Counter
+from collections.abc import Iterable
+
+from benchmarks.reliability.models import EventRecord, FaultSpec, Scenario
+
+
+class FaultSchedule:
+    """Stateful deterministic schedule for one scenario run."""
+
+    def __init__(self, scenario: Scenario) -> None:
+        self._faults = {fault.fault_id: fault for fault in scenario.faults}
+        fault_ids = sorted(self._faults)
+        random.Random(scenario.seed).shuffle(fault_ids)
+        self._tie_break_order = {
+            fault_id: rank for rank, fault_id in enumerate(fault_ids)
+        }
+        self._event_counts: Counter[str] = Counter()
+        self._seen_event_ids: set[str] = set()
+        self._fired_fault_ids: set[str] = set()
+
+    def pending(self) -> tuple[FaultSpec, ...]:
+        """Return faults that have not fired in deterministic order."""
+        return tuple(
+            fault
+            for fault in self._ordered_faults()
+            if fault.fault_id not in self._fired_fault_ids
+        )
+
+    def fired(self) -> tuple[FaultSpec, ...]:
+        """Return faults that have already fired in deterministic order."""
+        return tuple(
+            fault
+            for fault in self._ordered_faults()
+            if fault.fault_id in self._fired_fault_ids
+        )
+
+    def observe(self, events: Iterable[EventRecord]) -> tuple[FaultSpec, ...]:
+        """Return newly matched faults for previously unseen persisted events."""
+        matched: list[FaultSpec] = []
+        for event in events:
+            if event.event_id in self._seen_event_ids:
+                continue
+            self._seen_event_ids.add(event.event_id)
+            self._event_counts[event.event_type] += 1
+            ordinal = self._event_counts[event.event_type]
+
+            event_matches = [
+                fault for fault in self.pending() if _matches(fault, event, ordinal)
+            ]
+            event_matches.sort(key=lambda item: self._tie_break_order[item.fault_id])
+            for fault in event_matches:
+                self._fired_fault_ids.add(fault.fault_id)
+                matched.append(fault)
+        return tuple(matched)
+
+    def _ordered_faults(self) -> tuple[FaultSpec, ...]:
+        return tuple(
+            sorted(
+                self._faults.values(),
+                key=lambda fault: (
+                    fault.trigger.ordinal,
+                    fault.trigger.event_type,
+                    self._tie_break_order[fault.fault_id],
+                ),
+            )
+        )
+
+
+def compile_schedule(scenario: Scenario) -> FaultSchedule:
+    """Validate and compile a scenario into a deterministic fault schedule."""
+    return FaultSchedule(scenario)
+
+
+def _matches(fault: FaultSpec, event: EventRecord, ordinal: int) -> bool:
+    trigger = fault.trigger
+    return (
+        trigger.event_type == event.event_type
+        and trigger.ordinal == ordinal
+        and (trigger.tool_call_id is None or trigger.tool_call_id == event.tool_call_id)
+    )
diff --git a/benchmarks/reliability/sdk_events.py b/benchmarks/reliability/sdk_events.py
new file mode 100644
index 000000000..d509f59f5
--- /dev/null
+++ b/benchmarks/reliability/sdk_events.py
@@ -0,0 +1,72 @@
+"""Lossless trigger metadata conversion for OpenHands SDK events."""
+
+import time
+from collections.abc import Callable
+
+from benchmarks.reliability.models import EventRecord, JsonObject
+from openhands.sdk.event import (
+    ACPToolCallEvent,
+    ActionEvent,
+    Event,
+    ObservationBaseEvent,
+)
+
+
+SDKEventCallback = Callable[[Event], None]
+ReliabilityEventCallback = Callable[[EventRecord], None]
+
+
+def sdk_event_to_record(
+    event: Event,
+    *,
+    monotonic_seconds: float | None = None,
+) -> EventRecord:
+    """Convert one persisted SDK event without exposing hidden reasoning."""
+    tool_call_id: str | None = None
+    tool_name: str | None = None
+    if isinstance(event, ActionEvent | ObservationBaseEvent | ACPToolCallEvent):
+        tool_call_id = str(event.tool_call_id)
+    if isinstance(event, ActionEvent | ObservationBaseEvent):
+        tool_name = event.tool_name
+
+    details: JsonObject = {
+        "source": str(event.source),
+        "timestamp": event.timestamp,
+    }
+    if tool_name is not None:
+        details["tool_name"] = tool_name
+    return EventRecord(
+        event_id=str(event.id),
+        event_type=type(event).__name__,
+        monotonic_seconds=(
+            time.monotonic() if monotonic_seconds is None else monotonic_seconds
+        ),
+        tool_call_id=tool_call_id,
+        details=details,
+    )
+
+
+def build_reliability_event_callback(
+    on_event: ReliabilityEventCallback,
+    *,
+    persistence_callbacks: tuple[SDKEventCallback, ...] = (),
+    monotonic: Callable[[], float] = time.monotonic,
+) -> SDKEventCallback:
+    """Persist through existing callbacks before triggering fault injection.
+
+    Each persistence callback runs first. Only after all return is the event
+    exposed to the reliability scheduler. This preserves the required
+    persist-before-inject ordering.
+    """
+
+    def callback(event: Event) -> None:
+        for existing_callback in persistence_callbacks:
+            existing_callback(event)
+        on_event(
+            sdk_event_to_record(
+                event,
+                monotonic_seconds=monotonic(),
+            )
+        )
+
+    return callback
diff --git a/pyproject.toml b/pyproject.toml
index ce3d03192..f436510bf 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -93,6 +93,7 @@ swesmith-infer = "benchmarks.swesmith.run_infer:main"
 swesmith-eval = "benchmarks.swesmith.eval_infer:main"
 programbench-infer = "benchmarks.programbench.run_infer:main"
 programbench-eval = "benchmarks.programbench.eval_infer:main"
+reliability-eval = "benchmarks.reliability.cli:main"
 
 [build-system]
 requires = ["setuptools>=61.0", "wheel"]
diff --git a/tests/test_reliability.py b/tests/test_reliability.py
new file mode 100644
index 000000000..31b897f1c
--- /dev/null
+++ b/tests/test_reliability.py
@@ -0,0 +1,481 @@
+"""Focused tests for the deterministic reliability benchmark core."""
+
+from collections.abc import Callable
+from pathlib import Path
+
+import pytest
+
+from benchmarks.reliability.artifacts import RunArtifacts, read_json, read_jsonl
+from benchmarks.reliability.effects import EffectLedger, payload_digest
+from benchmarks.reliability.injectors import FaultInjectionError
+from benchmarks.reliability.models import (
+    EventRecord,
+    EventTrigger,
+    FaultKind,
+    FaultSpec,
+    JsonObject,
+    RecoveryReceipt,
+    RunIdentity,
+    RunMetrics,
+    RunObservation,
+    Scenario,
+)
+from benchmarks.reliability.reporting import (
+    write_scorecard_json,
+    write_scorecard_markdown,
+)
+from benchmarks.reliability.runner import ReliabilityAdapter, RunSession, run_scenario
+from benchmarks.reliability.scenario import load_scenario
+from benchmarks.reliability.schedule import compile_schedule
+from benchmarks.reliability.sdk_events import build_reliability_event_callback
+from openhands.sdk.event import Event, PauseEvent
+
+
+def test_load_json_scenario_and_reject_unknown_fields(tmp_path: Path) -> None:
+    scenario_path = tmp_path / "scenario.json"
+    scenario_path.write_text(
+        """
+{
+  "schema_version": 1,
+  "scenario_id": "restart-on-action",
+  "task": {
+    "benchmark": "swebench",
+    "instance_id": "project__issue-1"
+  },
+  "agent": {"config": "default"},
+  "seed": 17,
+  "faults": [
+    {
+      "fault_id": "restart",
+      "kind": "sandbox_restart",
+      "trigger": {
+        "event": {"event_type": "ActionEvent", "ordinal": 1}
+      },
+      "parameters": {"mode": "hard"}
+    }
+  ]
+}
+""".lstrip(),
+        encoding="utf-8",
+    )
+
+    scenario = load_scenario(scenario_path)
+
+    assert scenario.scenario_id == "restart-on-action"
+    assert scenario.faults[0].kind == FaultKind.SANDBOX_RESTART
+    invalid_path = tmp_path / "invalid.json"
+    invalid_path.write_text(
+        scenario_path.read_text(encoding="utf-8").replace(
+            '"schema_version": 1,',
+            '"schema_version": 1, "unknown": true,',
+        ),
+        encoding="utf-8",
+    )
+    with pytest.raises(ValueError, match="unknown fields: unknown"):
+        load_scenario(invalid_path)
+
+
+def test_schedule_is_seeded_deterministic_and_ignores_duplicate_events() -> None:
+    scenario = _scenario()
+    first = compile_schedule(scenario)
+    second = compile_schedule(scenario)
+    event = _event("action-1", ordinal=1)
+
+    first_match = first.observe((event,))
+    second_match = second.observe((event,))
+
+    assert [fault.fault_id for fault in first_match] == [
+        fault.fault_id for fault in second_match
+    ]
+    assert set(fault.fault_id for fault in first_match) == {
+        "restart",
+        "lost-response",
+        "sigkill",
+        "partition",
+    }
+    assert first.observe((event,)) == ()
+    assert first.pending() == ()
+
+
+def test_effect_ledger_uses_canonical_payload_digests(tmp_path: Path) -> None:
+    ledger = EffectLedger(tmp_path / "effects.jsonl")
+
+    intent = ledger.record_intent(
+        operation_id="publish-1",
+        payload={"b": 2, "a": 1},
+        monotonic_seconds=1.0,
+    )
+    commit = ledger.record_commit(
+        operation_id="publish-1",
+        payload={"a": 1, "b": 2},
+        monotonic_seconds=2.0,
+    )
+
+    assert intent.payload_digest == commit.payload_digest
+    assert intent.payload_digest == payload_digest({"a": 1, "b": 2})
+    assert len(ledger.records()) == 2
+
+
+def test_sdk_callback_persists_before_exposing_trigger() -> None:
+    order: list[str] = []
+    records: list[EventRecord] = []
+
+    def persist(event: Event) -> None:
+        order.append(f"persist:{event.id}")
+
+    def observe(record: EventRecord) -> None:
+        order.append("reliability")
+        records.append(record)
+
+    callback = build_reliability_event_callback(
+        observe,
+        persistence_callbacks=(persist,),
+        monotonic=lambda: 12.5,
+    )
+    event = PauseEvent(id="pause-1")
+
+    callback(event)
+
+    assert order == ["persist:pause-1", "reliability"]
+    assert records == [
+        EventRecord(
+            event_id="pause-1",
+            event_type="PauseEvent",
+            monotonic_seconds=12.5,
+            details={
+                "source": "user",
+                "timestamp": event.timestamp,
+            },
+        )
+    ]
+
+
+def test_end_to_end_run_injects_all_faults_and_scores_artifacts(
+    tmp_path: Path,
+) -> None:
+    adapter = _FakeAdapter(duplicate_effect=False)
+
+    result = run_scenario(
+        _scenario(),
+        adapter=adapter,
+        output_root=tmp_path,
+    )
+
+    assert result.task_passed
+    assert result.completion_resume.passed
+    assert result.no_duplicate_effect.passed
+    assert result.recovery_overhead.passed
+    fault_root = result.run.artifact_dir
+    receipts = read_jsonl(fault_root / "fault_receipts.jsonl")
+    assert sum(item["status"] == "applied" for item in receipts) == 4
+    assert sum(item["status"] == "released" for item in receipts) == 1
+    assert adapter.faulted_session is not None
+    assert set(adapter.faulted_session.calls[:-1]) == {
+        "restart_sandbox",
+        "drop_dispatch_response",
+        "sigkill_mid_tool_call",
+        "partition_network",
+    }
+    assert adapter.faulted_session.calls[-1] == "heal_network"
+    overhead = result.recovery_overhead.value
+    assert isinstance(overhead, dict)
+    assert overhead["wall_time_ratio"] == pytest.approx(1.5)
+    assert read_json(fault_root / "reliability_result.json")["task_passed"] is True
+
+
+def test_end_to_end_duplicate_effect_is_visible_not_inferred(
+    tmp_path: Path,
+) -> None:
+    result = run_scenario(
+        _scenario(),
+        adapter=_FakeAdapter(duplicate_effect=True),
+        output_root=tmp_path,
+    )
+
+    assert not result.no_duplicate_effect.passed
+    assert result.no_duplicate_effect.reason_codes == ("duplicate_effect",)
+    value = result.no_duplicate_effect.value
+    assert isinstance(value, dict)
+    assert value["excess_commits"] == 1
+
+
+def test_scorecards_include_raw_components(
+    tmp_path: Path,
+) -> None:
+    result = run_scenario(
+        _scenario(),
+        adapter=_FakeAdapter(duplicate_effect=False),
+        output_root=tmp_path / "runs",
+    )
+    json_path = tmp_path / "scorecard.json"
+    markdown_path = tmp_path / "scorecard.md"
+
+    write_scorecard_json((result,), json_path)
+    write_scorecard_markdown((result,), markdown_path)
+
+    scorecard = read_json(json_path)
+    summary = scorecard["summary"]
+    assert isinstance(summary, dict)
+    assert summary["completion_resume_passes"] == 1
+    markdown = markdown_path.read_text(encoding="utf-8")
+    assert "Completion/resume passes: 1" in markdown
+    assert result.run.run_id in markdown
+
+
+def test_missing_recovery_receipts_cannot_score_as_resume(tmp_path: Path) -> None:
+    result = run_scenario(
+        _scenario(),
+        adapter=_FakeAdapter(
+            duplicate_effect=False,
+            emit_recovery=False,
+        ),
+        output_root=tmp_path,
+    )
+
+    assert result.task_passed
+    assert not result.completion_resume.passed
+    assert "recovery_receipt_missing" in result.completion_resume.reason_codes
+
+
+def test_failed_injection_is_recorded_before_run_fails(tmp_path: Path) -> None:
+    with pytest.raises(FaultInjectionError, match="failed to inject restart"):
+        run_scenario(
+            _scenario(),
+            adapter=_FakeAdapter(
+                duplicate_effect=False,
+                fail_restart=True,
+            ),
+            output_root=tmp_path,
+        )
+
+    receipts = read_jsonl(
+        tmp_path / "e2e-faults" / "23" / "faulted" / "fault_receipts.jsonl"
+    )
+    restart_receipts = [
+        receipt for receipt in receipts if receipt["fault_id"] == "restart"
+    ]
+    assert restart_receipts[0]["status"] == "failed"
+    assert "RuntimeError: restart failed" in str(restart_receipts[0]["details"])
+
+
+def test_missing_effect_evidence_cannot_score_as_exactly_once(
+    tmp_path: Path,
+) -> None:
+    result = run_scenario(
+        _scenario(),
+        adapter=_FakeAdapter(
+            duplicate_effect=False,
+            emit_effects=False,
+        ),
+        output_root=tmp_path,
+    )
+
+    assert not result.no_duplicate_effect.passed
+    assert result.no_duplicate_effect.reason_codes == ("effect_evidence_missing",)
+
+
+class _FakeAdapter(ReliabilityAdapter):
+    def __init__(
+        self,
+        *,
+        duplicate_effect: bool,
+        emit_recovery: bool = True,
+        fail_restart: bool = False,
+        emit_effects: bool = True,
+    ) -> None:
+        self.duplicate_effect = duplicate_effect
+        self.emit_recovery = emit_recovery
+        self.fail_restart = fail_restart
+        self.emit_effects = emit_effects
+        self.faulted_session: _FakeSession | None = None
+
+    def open_run(
+        self,
+        *,
+        scenario: Scenario,
+        run: RunIdentity,
+        artifacts: RunArtifacts,
+        baseline: bool,
+    ) -> RunSession:
+        session = _FakeSession(
+            run=run,
+            artifacts=artifacts,
+            baseline=baseline,
+            duplicate_effect=self.duplicate_effect,
+            emit_recovery=self.emit_recovery,
+            fail_restart=self.fail_restart,
+            emit_effects=self.emit_effects,
+        )
+        if not baseline:
+            self.faulted_session = session
+        return session
+
+
+class _FakeSession(RunSession):
+    def __init__(
+        self,
+        *,
+        run: RunIdentity,
+        artifacts: RunArtifacts,
+        baseline: bool,
+        duplicate_effect: bool,
+        emit_recovery: bool,
+        fail_restart: bool,
+        emit_effects: bool,
+    ) -> None:
+        self._run = run
+        self.artifacts = artifacts
+        self.baseline = baseline
+        self.duplicate_effect = duplicate_effect
+        self.emit_recovery = emit_recovery
+        self.fail_restart = fail_restart
+        self.emit_effects = emit_effects
+        self.clock = 0.0
+        self.calls: list[str] = []
+        self.ledger = EffectLedger(artifacts.effects_path)
+
+    @property
+    def run(self) -> RunIdentity:
+        return self._run
+
+    def monotonic(self) -> float:
+        self.clock += 1.0
+        return self.clock
+
+    def restart_sandbox(self, fault: FaultSpec) -> JsonObject:
+        self.calls.append("restart_sandbox")
+        if self.fail_restart:
+            raise RuntimeError("restart failed")
+        if self.emit_effects:
+            self.ledger.record_commit(
+                operation_id="publish-1",
+                payload={"message": "hello"},
+                monotonic_seconds=self.monotonic(),
+            )
+        return {"boundary": "fake-agent-server", "mode": fault.parameters["mode"]}
+
+    def drop_dispatch_response(self, fault: FaultSpec) -> JsonObject:
+        self.calls.append("drop_dispatch_response")
+        return {"boundary": "tool-result-before-observation"}
+
+    def sigkill_mid_tool_call(self, fault: FaultSpec) -> JsonObject:
+        self.calls.append("sigkill_mid_tool_call")
+        return {"target": "fake-tool-process", "signal": "SIGKILL"}
+
+    def partition_network(self, fault: FaultSpec) -> JsonObject:
+        self.calls.append("partition_network")
+        return {"endpoint": "fake-agent-server", "direction": "both"}
+
+    def heal_network(self, fault: FaultSpec) -> JsonObject:
+        self.calls.append("heal_network")
+        return {"endpoint": "fake-agent-server"}
+
+    def execute(
+        self,
+        on_event: Callable[[EventRecord], None],
+    ) -> RunObservation:
+        if self.emit_effects:
+            self.ledger.record_intent(
+                operation_id="publish-1",
+                payload={"message": "hello"},
+                monotonic_seconds=self.monotonic(),
+            )
+        if self.baseline and self.emit_effects:
+            self.ledger.record_commit(
+                operation_id="publish-1",
+                payload={"message": "hello"},
+                monotonic_seconds=self.monotonic(),
+            )
+        on_event(_event("action-1", ordinal=1))
+        if not self.baseline and self.duplicate_effect and self.emit_effects:
+            self.ledger.record_commit(
+                operation_id="publish-1",
+                payload={"message": "hello"},
+                monotonic_seconds=self.monotonic(),
+            )
+        on_event(
+            EventRecord(
+                event_id="observation-1",
+                event_type="ObservationEvent",
+                monotonic_seconds=self.monotonic(),
+                tool_call_id="tool-1",
+            )
+        )
+        receipts = (
+            ()
+            if self.baseline or not self.emit_recovery
+            else (
+                RecoveryReceipt(
+                    run_id=self.run.run_id,
+                    receipt_type="conversation_history_restored",
+                    succeeded=True,
+                    monotonic_seconds=self.monotonic(),
+                ),
+                RecoveryReceipt(
+                    run_id=self.run.run_id,
+                    receipt_type="runtime_reattached",
+                    succeeded=True,
+                    monotonic_seconds=self.monotonic(),
+                ),
+            )
+        )
+        return RunObservation(
+            task_passed=True,
+            metrics=RunMetrics(
+                wall_time_seconds=10.0 if self.baseline else 15.0,
+                recovery_time_seconds=None if self.baseline else 3.0,
+                iterations=2 if self.baseline else 3,
+                event_count=2,
+                tool_calls=1,
+                tokens=100,
+                cost_usd=0.01,
+            ),
+            recovery_receipts=receipts,
+        )
+
+
+def _scenario() -> Scenario:
+    event_trigger = EventTrigger(
+        event_type="ActionEvent",
+        ordinal=1,
+    )
+    return Scenario(
+        schema_version=1,
+        scenario_id="e2e-faults",
+        benchmark="swebench",
+        instance_id="project__issue-1",
+        agent_config="default",
+        seed=23,
+        faults=(
+            FaultSpec(
+                fault_id="restart",
+                kind=FaultKind.SANDBOX_RESTART,
+                trigger=event_trigger,
+                parameters={"mode": "hard"},
+            ),
+            FaultSpec(
+                fault_id="lost-response",
+                kind=FaultKind.LOST_DISPATCH_RESPONSE,
+                trigger=event_trigger,
+            ),
+            FaultSpec(
+                fault_id="sigkill",
+                kind=FaultKind.SIGKILL_MID_TOOL_CALL,
+                trigger=event_trigger,
+            ),
+            FaultSpec(
+                fault_id="partition",
+                kind=FaultKind.NETWORK_PARTITION,
+                trigger=event_trigger,
+            ),
+        ),
+    )
+
+
+def _event(event_id: str, *, ordinal: int) -> EventRecord:
+    return EventRecord(
+        event_id=event_id,
+        event_type="ActionEvent",
+        monotonic_seconds=float(ordinal),
+        tool_call_id="tool-1",
+    )