Build a Self-Healing AWS Service

Introduction

30 Second Summary

Automated recovery sounds reassuring until one failure triggers the same restart three times. A fix without safety limits can make the outage worse.

In this project, you will build a deterministic local reliability lab in Python that converts simulated AWS service health signals into safe recovery decisions. You will prove its behavior with repeatable tests and audit reports.

What You'll Build

You will run a local simulation and watch 17 health signals become eight justified alarms while every restart, suppression, block, and escalation appears in an audit trail.

By the end of this project, you'll have:

• Detect sustained failures while keeping healthy services and one-off transient failures quiet.
• Compare unsafe and safe recovery by reproducing three restart requests for one incident before suppressing duplicates and bounding later attempts.
• Show reviewable safety evidence through eight passing pytest tests plus JSON, Markdown, and JSONL reports generated from SQLite audit history.
• Secret Mission: Replay a 24-delivery recovery storm across a simulated process restart and prove that only two restart actions execute.

Are there any prerequisites?

You need Python 3.11 or later plus Visual Studio Code with the Microsoft Python extension. You do not need AWS credentials, an AWS account, Docker, Terraform, or any paid service.

Before We Start

Before any hands-on work begins, commit to the safety problem this local reliability simulation solves. You will prevent duplicate or unsafe automated recovery actions while preserving deterministic policy evidence for every decision.

Set Up Health Signals

A repeatable reliability experiment depends on a fixed runtime. It also needs labeled input that produces the same evidence during every run.

This step creates that controlled workspace. You will isolate Python dependencies. You will also load the deterministic health signals that drive the simulation.

In this step, get ready to:

• Create the repository structure in Visual Studio Code.
• Install Python and pytest inside a project-local virtual environment.
• Validate the health fixtures through the inspection command.

Create the workspace structure

The repository separates deterministic inputs from application code. It also keeps generated evidence outside the source package.

• Press Cmd+Space on macOS or the Windows key on Windows to open system search.
• Type Visual Studio Code in the search field.
• Press Enter to launch Visual Studio Code.
• Select the folder-opening option on the Visual Studio Code start screen.
• Choose the empty project folder in the folder picker.
• Use the Explorer sidebar to reproduce the repository tree shown below:

.
├── .vscode/settings.json
├── data/
│   ├── health_signals.json
│   ├── policies.json
│   └── unsafe_policies.json
├── reports/
├── self_healing/
│   ├── __init__.py
│   ├── controller.py
│   ├── detector.py
│   ├── models.py
│   ├── policy.py
│   ├── reporting.py
│   ├── routing.py
│   ├── simulation.py
│   └── store.py
├── tests/test_self_healing.py
├── .gitignore
├── main.py
├── README.md
└── requirements.txt

How is the repository divided?

• The data/ directory holds deterministic health signals and recovery policies.
• The self_healing/ directory holds the simulation package.
• The reports/ directory receives generated evidence.
• The tests/ directory holds the executable safety contract.

Your Explorer sidebar should now show five top-level directories. This visible tree confirms that every later file has a defined location.

• Copy each project file from the Complete Code reference into its matching path.
• Save every project file.
• Compare the setup-specific files with the reference below.

✔️ Awesome, I've got everything!

Your setup files match the project reference. Confirm that every editor tab is saved before you configure the runtime.

ⓧ I'd like to double check the full code

pytest==9.2.0

The dependency file pins pytest to version 9.2.0 so every learner runs the same test release.

.venv/
__pycache__/
.pytest_cache/
*.pyc
data/*.db
reports/*.json
reports/*.jsonl
reports/*.md

The ignore rules keep the local environment out of version control. They also exclude generated databases and reports.

{
  "python.testing.pytestEnabled": true,
  "python.testing.unittestEnabled": false,
  "python.testing.pytestArgs": ["tests"]
}

The workspace settings enable pytest discovery. They scope discovery to the tests directory.

"""Deterministic local simulation of a safe AWS-style recovery controller."""

The package docstring identifies the code as a deterministic local simulation. It preserves the boundary between educational behavior and real AWS operations.

import argparse

from self_healing.simulation import run_detect, run_inspect, run_safe, run_unsafe


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Run the deterministic self-healing service simulation"
    )
    parser.add_argument(
        "command",
        choices=["inspect", "detect", "unsafe", "safe"],
    )
    args = parser.parse_args()
    commands = {
        "inspect": run_inspect,
        "detect": run_detect,
        "unsafe": run_unsafe,
        "safe": run_safe,
    }
    commands[args.command]()


if __name__ == "__main__":
    main()

The command-line entry point maps each supported command to its simulation function. The inspect command provides the setup checkpoint used in this step.

{
  "detection": {
    "consecutive_failures": 2,
    "latency_ms": 500,
    "error_rate": 0.1
  },
  "default": {
    "action": "WAIT_AND_RECHECK",
    "cooldown_seconds": 300,
    "max_attempts": 2,
    "allow_critical_restart": false
  },
  "services": {
    "checkout-prod": {
      "action": "NO_ACTION",
      "cooldown_seconds": 300,
      "max_attempts": 1,
      "allow_critical_restart": false
    },
    "search-prod": {
      "action": "WAIT_AND_RECHECK",
      "cooldown_seconds": 300,
      "max_attempts": 1,
      "allow_critical_restart": false
    },
    "payments-prod": {
      "action": "RESTART_TASK",
      "cooldown_seconds": 300,
      "max_attempts": 2,
      "allow_critical_restart": false
    },
    "admin-prod": {
      "action": "RESTART_TASK",
      "cooldown_seconds": 300,
      "max_attempts": 1,
      "allow_critical_restart": false
    },
    "catalog-prod": {
      "action": "ESCALATE",
      "cooldown_seconds": 300,
      "max_attempts": 1,
      "allow_critical_restart": false
    },
    "batch-prod": {
      "action": "NO_ACTION",
      "cooldown_seconds": 300,
      "max_attempts": 1,
      "allow_critical_restart": false
    }
  }
}

The valid policy defines the detector thresholds. It also assigns an explicit recovery action to each configured service.

{
  "detection": {
    "consecutive_failures": 2,
    "latency_ms": 500,
    "error_rate": 0.1
  },
  "default": {
    "action": "WAIT_AND_RECHECK",
    "cooldown_seconds": 300,
    "max_attempts": 2,
    "allow_critical_restart": false
  },
  "services": {
    "admin-prod": {
      "action": "RESTART_TASK",
      "cooldown_seconds": 0,
      "max_attempts": 99,
      "allow_critical_restart": true
    }
  }
}

The unsafe fixture enables a prohibited critical restart override. Later validation proves that this policy fails closed.

[
  {
    "event_id": "check-1",
    "service": "checkout-prod",
    "timestamp": "2026-01-01T00:00:00+00:00",
    "healthy_task_count": 3,
    "desired_task_count": 3,
    "latency_ms": 120,
    "error_rate": 0.01,
    "environment": "prod",
    "critical": false,
    "expected_alarm": false
  },
  {
    "event_id": "check-2",
    "service": "checkout-prod",
    "timestamp": "2026-01-01T00:01:00+00:00",
    "healthy_task_count": 3,
    "desired_task_count": 3,
    "latency_ms": 130,
    "error_rate": 0.01,
    "environment": "prod",
    "critical": false,
    "expected_alarm": false
  },
  {
    "event_id": "search-1",
    "service": "search-prod",
    "timestamp": "2026-01-01T00:00:00+00:00",
    "healthy_task_count": 1,
    "desired_task_count": 2,
    "latency_ms": 200,
    "error_rate": 0.02,
    "environment": "prod",
    "critical": false,
    "expected_alarm": false
  },
  {
    "event_id": "search-2",
    "service": "search-prod",
    "timestamp": "2026-01-01T00:01:00+00:00",
    "healthy_task_count": 2,
    "desired_task_count": 2,
    "latency_ms": 180,
    "error_rate": 0.01,
    "environment": "prod",
    "critical": false,
    "expected_alarm": false
  },
  {
    "event_id": "pay-1",
    "service": "payments-prod",
    "timestamp": "2026-01-01T00:00:00+00:00",
    "healthy_task_count": 0,
    "desired_task_count": 2,
    "latency_ms": 800,
    "error_rate": 0.25,
    "environment": "prod",
    "critical": false,
    "expected_alarm": false
  },
  {
    "event_id": "pay-2",
    "service": "payments-prod",
    "timestamp": "2026-01-01T00:01:00+00:00",
    "healthy_task_count": 0,
    "desired_task_count": 2,
    "latency_ms": 820,
    "error_rate": 0.28,
    "environment": "prod",
    "critical": false,
    "expected_alarm": true
  },
  {
    "event_id": "pay-3",
    "service": "payments-prod",
    "timestamp": "2026-01-01T00:02:00+00:00",
    "healthy_task_count": 0,
    "desired_task_count": 2,
    "latency_ms": 850,
    "error_rate": 0.3,
    "environment": "prod",
    "critical": false,
    "expected_alarm": true
  },
  {
    "event_id": "pay-4",
    "service": "payments-prod",
    "timestamp": "2026-01-01T00:07:00+00:00",
    "healthy_task_count": 0,
    "desired_task_count": 2,
    "latency_ms": 860,
    "error_rate": 0.31,
    "environment": "prod",
    "critical": false,
    "expected_alarm": true
  },
  {
    "event_id": "pay-5",
    "service": "payments-prod",
    "timestamp": "2026-01-01T00:13:00+00:00",
    "healthy_task_count": 0,
    "desired_task_count": 2,
    "latency_ms": 870,
    "error_rate": 0.32,
    "environment": "prod",
    "critical": false,
    "expected_alarm": true
  },
  {
    "event_id": "admin-1",
    "service": "admin-prod",
    "timestamp": "2026-01-01T00:00:00+00:00",
    "healthy_task_count": 0,
    "desired_task_count": 2,
    "latency_ms": 900,
    "error_rate": 0.4,
    "environment": "prod",
    "critical": true,
    "expected_alarm": false
  },
  {
    "event_id": "admin-2",
    "service": "admin-prod",
    "timestamp": "2026-01-01T00:01:00+00:00",
    "healthy_task_count": 0,
    "desired_task_count": 2,
    "latency_ms": 920,
    "error_rate": 0.42,
    "environment": "prod",
    "critical": true,
    "expected_alarm": true
  },
  {
    "event_id": "catalog-1",
    "service": "catalog-prod",
    "timestamp": "2026-01-01T00:00:00+00:00",
    "healthy_task_count": 2,
    "desired_task_count": 2,
    "latency_ms": 700,
    "error_rate": 0.02,
    "environment": "prod",
    "critical": false,
    "expected_alarm": false
  },
  {
    "event_id": "catalog-2",
    "service": "catalog-prod",
    "timestamp": "2026-01-01T00:01:00+00:00",
    "healthy_task_count": 2,
    "desired_task_count": 2,
    "latency_ms": 720,
    "error_rate": 0.02,
    "environment": "prod",
    "critical": false,
    "expected_alarm": true
  },
  {
    "event_id": "worker-1",
    "service": "worker-prod",
    "timestamp": "2026-01-01T00:00:00+00:00",
    "healthy_task_count": 2,
    "desired_task_count": 2,
    "latency_ms": 200,
    "error_rate": 0.2,
    "environment": "prod",
    "critical": false,
    "expected_alarm": false
  },
  {
    "event_id": "worker-2",
    "service": "worker-prod",
    "timestamp": "2026-01-01T00:01:00+00:00",
    "healthy_task_count": 2,
    "desired_task_count": 2,
    "latency_ms": 210,
    "error_rate": 0.25,
    "environment": "prod",
    "critical": false,
    "expected_alarm": true
  },
  {
    "event_id": "batch-1",
    "service": "batch-prod",
    "timestamp": "2026-01-01T00:00:00+00:00",
    "healthy_task_count": 0,
    "desired_task_count": 1,
    "latency_ms": 100,
    "error_rate": 0.01,
    "environment": "prod",
    "critical": false,
    "expected_alarm": false
  },
  {
    "event_id": "batch-2",
    "service": "batch-prod",
    "timestamp": "2026-01-01T00:01:00+00:00",
    "healthy_task_count": 0,
    "desired_task_count": 1,
    "latency_ms": 100,
    "error_rate": 0.01,
    "environment": "prod",
    "critical": false,
    "expected_alarm": true
  }
]

The fixture contains 17 ordered signals from seven services. Eight signals carry an expected alarm label for later detector verification.

Missing a file or directory?

Compare every path in the repository reference with the Explorer sidebar. Pay special attention to .vscode/settings.json because the leading dot makes the directory easy to overlook.

Still seeing a mismatch? Help me compare my repository with the required project structure.

Set up the Python environment

A project-local virtual environment keeps this repository's packages separate from other Python projects. Python 3.11+ also provides the language features used by the simulation.

• Open the integrated terminal from the Visual Studio Code workspace.
• Check the installed Python version with the command for your operating system.

macOS

• Check the macOS Python version by running this command:

python3 --version

The command prints the Python version available through python3.

Windows

• Check the Windows Python version by running this command:

py --version

The command prints the Python version available through the Windows Python launcher.

Choose the outcome that matches the version displayed in your terminal.

✔️ I see version 3.11 or higher

Your interpreter meets the project requirement. Continue with the environment commands below.

ⓧ I see an older version

The detected interpreter is below the project requirement. Install Python 3.11+ with the official installer for your operating system.

• Complete the Python installer for your operating system.
• Restart the Visual Studio Code terminal.
• Run the platform version command again.
• Confirm the result reports Python 3.11 or higher.
• Select the new Python installation in Visual Studio Code if the older system interpreter still appears.

ⓧ Command not found

The terminal cannot find a Python interpreter through the expected platform command. Install Python 3.11+ with the official installer for your operating system.

• Complete the Python installer for your operating system.
• Restart Visual Studio Code.
• Return to the integrated terminal.
• Run the platform version command again.
• Confirm the result reports Python 3.11 or higher.

A compatible interpreter is now available. The next commands create an isolated .venv inside this project.

macOS

• Create the project-local environment by running this command:

python3 -m venv .venv

The Explorer sidebar should now show a .venv directory inside the project workspace.

• Activate the environment by running this command:

source .venv/bin/activate

Your terminal prompt should now include .venv.

• Install the pinned test dependency by running this command:

python3 -m pip install -r requirements.txt

The installation output should identify pytest version 9.2.0.

Environment command failed?

Confirm the terminal is inside the folder containing requirements.txt. Check that .venv exists in that same folder.

Need help with the local environment? Help me diagnose my macOS virtual environment setup.

Windows

• Create the project-local environment by running this command:

py -m venv .venv

The Explorer sidebar should now show a .venv directory inside the project workspace.

• Activate the environment by running this command:

.venv\Scripts\activate

Your terminal prompt should now include .venv.

• Install the pinned test dependency by running this command:

py -m pip install -r requirements.txt

The installation output should identify pytest version 9.2.0.

PowerShell blocked activation?

PowerShell can block local activation scripts through its execution policy. Apply the documented user-scoped policy by running:

Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser

This setting permits locally created scripts for your user account. Return to the activation command after it completes.

Still blocked? Help me diagnose my Windows virtual environment setup.

The runtime is isolated now. Visual Studio Code also needs to use this same interpreter for running code and discovering tests.

• Select the extensions icon in the left activity bar.
• Search for the Microsoft Python extension.
• Install or enable the Microsoft Python extension.
• Select the Python interpreter display in the Status Bar.
• Choose the interpreter inside .venv.

Why select the workspace interpreter?

Visual Studio Code uses the selected interpreter for execution and test discovery. Selecting .venv keeps the editor aligned with the activated terminal.

That is the environment locked down. Your terminal and editor now use the same pinned local runtime.

Validate the health fixtures

The health fixture contains 17 timestamped signals from seven simulated services. Eight labels identify signals that should become alarms after sustained evidence is evaluated.

The policy loader uses a fail-closed schema. The valid fixture must pass before any detector or recovery controller can be trusted.

• Switch back to data/health_signals.json in the Explorer sidebar.
• Confirm the fixture contains 17 signal objects.
• Switch back to data/policies.json in the Explorer sidebar.
• Confirm consecutive_failures is set to 2.
• Switch back to data/unsafe_policies.json in the Explorer sidebar.
• Confirm allow_critical_restart is set to true for admin-prod.

Why keep an unsafe fixture?

The unsafe fixture creates a fixed policy violation without enabling a dangerous runtime path. Later tests can prove that validation rejects the same critical-workload override during every run.

Before you run the inspection, do you expect the valid policy and all fixture counts to pass together?

• Validate the repository from the active workspace terminal by running this command:

python main.py inspect

What does this command check?

• The inspection loads the deterministic health signals.
• It counts the unique simulated services.
• It counts the expected alarm labels.
• It loads the valid recovery policy through the fail-closed schema.

You should see Signals: 17. You should also see Services: 7.

The remaining output should include Expected alarms: 8 and Policy validation: PASS.

You have proved that the simulation starts from a stable runtime and validated input. That foundation makes every later safety result reproducible.

Inspection did not pass?

• Confirm the terminal is running from the folder containing main.py.
• Confirm every property in the three fixture files uses double quotes.
• Check that data/health_signals.json contains all 17 signal objects.

Still blocked? Help me debug the fixture inspection without changing the expected evidence.

Your reproducible reliability workspace is ready. Next, you will turn these labeled signals into alarms only when failures persist.

Detect Sustained Failures

Your fixtures already provide a controlled picture of healthy services, transient failures, persistent outages, critical workloads, and policy outcomes.

A recovery system that reacts to one bad datapoint can restart healthy work during a brief wobble. This step builds a detector that waits for sustained evidence before it creates an alarm.

In this step, get ready to:

• Model fixture signals with immutable data classes.
• Evaluate consecutive threshold breaches for each service.
• Route only matching alarm events through local envelopes.

Model signals and evaluations

Immutable dataclasses give every fixture a predictable shape. The detector can then return a separate result without changing the original signal.

The RecoveryAction enum also establishes the four decisions that later controller steps can request.

• Create self_healing/models.py in the existing self_healing folder with the code below by using the file controls in the VS Code sidebar:

from dataclasses import dataclass
from enum import StrEnum


class RecoveryAction(StrEnum):
    RESTART_TASK = "RESTART_TASK"
    WAIT_AND_RECHECK = "WAIT_AND_RECHECK"
    ESCALATE = "ESCALATE"
    NO_ACTION = "NO_ACTION"

What does this code do?

• The StrEnum base gives each recovery action a stable string value.
• The four members make the controller's possible requests explicit.
• The detector does not choose these actions. Later policy logic owns that responsibility.

• Save self_healing/models.py.
• Check the file for Python syntax diagnostics.

You should see no syntax errors around RecoveryAction or its four members.

Seeing an enum syntax error?

• Check that every member is indented with four spaces beneath RecoveryAction.
• Check that every string value uses matching quotation marks.

Still stuck? Help me fix the RecoveryAction enum in models.py.

• Place the cursor two blank lines below NO_ACTION = "NO_ACTION".
• Add the immutable health-signal model by copying this code:

@dataclass(frozen=True)
class HealthSignal:
    event_id: str
    service: str
    timestamp: str
    healthy_task_count: int
    desired_task_count: int
    latency_ms: float
    error_rate: float
    environment: str
    critical: bool
    expected_alarm: bool

    @classmethod
    def from_dict(cls, raw: dict[str, object]) -> "HealthSignal":
        return cls(
            event_id=str(raw["event_id"]),
            service=str(raw["service"]),
            timestamp=str(raw["timestamp"]),
            healthy_task_count=int(raw["healthy_task_count"]),
            desired_task_count=int(raw["desired_task_count"]),
            latency_ms=float(raw["latency_ms"]),
            error_rate=float(raw["error_rate"]),
            environment=str(raw["environment"]),
            critical=bool(raw["critical"]),
            expected_alarm=bool(raw["expected_alarm"]),
        )

How does HealthSignal protect the detector?

• The fields preserve every value from one deterministic fixture.
• The frozen=True setting prevents a signal from changing after creation.
• The from_dict() method converts raw JSON values into the expected Python types.
• The expected_alarm label supports the accuracy check at the end of the step.

• Save self_healing/models.py.
• Check the HealthSignal class for Python syntax diagnostics.

You should see a clean class definition with all fixture fields listed above from_dict().

Seeing an indentation error in HealthSignal?

• Align every field with four spaces beneath class HealthSignal:.
• Indent the constructor arguments beneath return cls(.

Need another pair of eyes? Help me fix the HealthSignal dataclass indentation.

• Place the cursor two blank lines below the closing parenthesis of HealthSignal.from_dict().
• Add the detector result model by copying this code:

@dataclass(frozen=True)
class Evaluation:
    signal: HealthSignal
    alarm: bool
    reasons: tuple[str, ...]
    consecutive_failures: int

    def to_dict(self) -> dict[str, object]:
        return {
            "event_id": self.signal.event_id,
            "service": self.signal.service,
            "timestamp": self.signal.timestamp,
            "alarm": self.alarm,
            "expected_alarm": self.signal.expected_alarm,
            "reasons": list(self.reasons),
            "consecutive_failures": self.consecutive_failures,
        }

What does Evaluation capture?

• The signal field links the result to its original health observation.
• The alarm field records the detector's final state.
• The reasons tuple preserves every breached threshold.
• The consecutive_failures field exposes the service's current failure streak.

• Save self_healing/models.py.
• Check the complete file for Python syntax diagnostics.

You should see no syntax errors across the enum or either immutable data model.

Seeing a type annotation error?

• Check that reasons uses tuple[str, ...].
• Check that to_dict() converts the tuple with list(self.reasons).

If the diagnostic remains, help me debug the Evaluation dataclass.

✔️ Awesome, I've got everything!

Your signal models now preserve fixture input and detector output in explicit immutable structures.

ⓧ I'd like to double check the full code

from dataclasses import dataclass
from enum import StrEnum


class RecoveryAction(StrEnum):
    RESTART_TASK = "RESTART_TASK"
    WAIT_AND_RECHECK = "WAIT_AND_RECHECK"
    ESCALATE = "ESCALATE"
    NO_ACTION = "NO_ACTION"


@dataclass(frozen=True)
class HealthSignal:
    event_id: str
    service: str
    timestamp: str
    healthy_task_count: int
    desired_task_count: int
    latency_ms: float
    error_rate: float
    environment: str
    critical: bool
    expected_alarm: bool

    @classmethod
    def from_dict(cls, raw: dict[str, object]) -> "HealthSignal":
        return cls(
            event_id=str(raw["event_id"]),
            service=str(raw["service"]),
            timestamp=str(raw["timestamp"]),
            healthy_task_count=int(raw["healthy_task_count"]),
            desired_task_count=int(raw["desired_task_count"]),
            latency_ms=float(raw["latency_ms"]),
            error_rate=float(raw["error_rate"]),
            environment=str(raw["environment"]),
            critical=bool(raw["critical"]),
            expected_alarm=bool(raw["expected_alarm"]),
        )


@dataclass(frozen=True)
class Evaluation:
    signal: HealthSignal
    alarm: bool
    reasons: tuple[str, ...]
    consecutive_failures: int

    def to_dict(self) -> dict[str, object]:
        return {
            "event_id": self.signal.event_id,
            "service": self.signal.service,
            "timestamp": self.signal.timestamp,
            "alarm": self.alarm,
            "expected_alarm": self.signal.expected_alarm,
            "reasons": list(self.reasons),
            "consecutive_failures": self.consecutive_failures,
        }

Evaluate sustained breaches

Each service needs an independent failure streak. A breach from payments-prod must never advance the count for search-prod.

The detector stores those streaks in _failure_counts. A healthy signal resets its service to zero.

• Create self_healing/detector.py in the existing self_healing folder with the code below by using the file controls in the VS Code sidebar:

from self_healing.models import Evaluation, HealthSignal


class HealthDetector:
    def __init__(
        self,
        consecutive_failures: int,
        latency_ms: float,
        error_rate: float,
    ) -> None:
        self.required_consecutive_failures = consecutive_failures
        self.latency_limit_ms = latency_ms
        self.error_rate_limit = error_rate
        self._failure_counts: dict[str, int] = {}

How is the detector configured?

• The constructor receives the required failure streak.
• The latency limit controls the response-time breach.
• The error-rate limit controls the failed-request breach.
• The _failure_counts dictionary keeps one count for each service name.

• Save self_healing/detector.py.
• Check the constructor for Python syntax diagnostics.

You should see no syntax errors around the threshold fields or the per-service dictionary.

Seeing an import or constructor error?

• Check that the import names match Evaluation and HealthSignal exactly.
• Check that every constructor assignment remains inside __init__().

For targeted help, help me debug the HealthDetector constructor.

• Place the cursor below the _failure_counts assignment inside HealthDetector.
• Add the evaluation methods by copying this code:

    def evaluate(self, signal: HealthSignal) -> Evaluation:
        reasons: list[str] = []
        if signal.healthy_task_count < signal.desired_task_count:
            reasons.append("healthy_task_count_below_desired")
        if signal.latency_ms > self.latency_limit_ms:
            reasons.append("latency_threshold_exceeded")
        if signal.error_rate > self.error_rate_limit:
            reasons.append("error_rate_threshold_exceeded")

        if reasons:
            count = self._failure_counts.get(signal.service, 0) + 1
            self._failure_counts[signal.service] = count
        else:
            count = 0
            self._failure_counts[signal.service] = 0

        return Evaluation(
            signal=signal,
            alarm=bool(reasons) and count >= self.required_consecutive_failures,
            reasons=tuple(reasons),
            consecutive_failures=count,
        )

    def evaluate_all(self, signals: list[HealthSignal]) -> list[Evaluation]:
        return [self.evaluate(signal) for signal in signals]

How does sustained detection work?

• Availability breaches when the healthy task count falls below the desired task count.
• Latency breaches when the observed value exceeds the configured limit.
• Error rate breaches when the observed value exceeds its configured limit.
• Any breach advances only the current service's streak.
• A healthy signal resets only the current service's streak.
• The alarm becomes true after the configured streak reaches its required length.

• Save self_healing/detector.py.
• Check the complete class for Python syntax diagnostics.

You should see evaluate() and evaluate_all() nested at the same indentation level inside HealthDetector.

Seeing a method indentation error?

• Align both method definitions with four spaces beneath class HealthDetector:.
• Keep the threshold checks inside evaluate().
• Keep the list comprehension inside evaluate_all().

If the class still fails to parse, help me fix the HealthDetector method indentation.

✔️ Awesome, I've got everything!

Your detector now tracks independent service streaks across all three health dimensions.

ⓧ I'd like to double check the full code

from self_healing.models import Evaluation, HealthSignal


class HealthDetector:
    def __init__(
        self,
        consecutive_failures: int,
        latency_ms: float,
        error_rate: float,
    ) -> None:
        self.required_consecutive_failures = consecutive_failures
        self.latency_limit_ms = latency_ms
        self.error_rate_limit = error_rate
        self._failure_counts: dict[str, int] = {}

    def evaluate(self, signal: HealthSignal) -> Evaluation:
        reasons: list[str] = []
        if signal.healthy_task_count < signal.desired_task_count:
            reasons.append("healthy_task_count_below_desired")
        if signal.latency_ms > self.latency_limit_ms:
            reasons.append("latency_threshold_exceeded")
        if signal.error_rate > self.error_rate_limit:
            reasons.append("error_rate_threshold_exceeded")

        if reasons:
            count = self._failure_counts.get(signal.service, 0) + 1
            self._failure_counts[signal.service] = count
        else:
            count = 0
            self._failure_counts[signal.service] = 0

        return Evaluation(
            signal=signal,
            alarm=bool(reasons) and count >= self.required_consecutive_failures,
            reasons=tuple(reasons),
            consecutive_failures=count,
        )

    def evaluate_all(self, signals: list[HealthSignal]) -> list[Evaluation]:
        return [self.evaluate(signal) for signal in signals]

Route alarm events and verify detection

Alarm evaluations become local envelopes inspired by CloudWatch alarm events. The simulated source label keeps the local safety boundary visible.

An EventBridge-style router accepts an envelope only when its source, detail type, and alarm state match the local rule.

• Create self_healing/routing.py in the existing self_healing folder with the code below by using the file controls in the VS Code sidebar:

from self_healing.models import Evaluation


SIMULATED_SOURCE = "local.cloudwatch.simulation"
SIMULATED_DETAIL_TYPE = "Simulated CloudWatch Alarm State Change"


def cloudwatch_style_event(evaluation: Evaluation) -> dict[str, object]:
    if not evaluation.alarm:
        raise ValueError("Only alarm evaluations can become routed events")

    signal = evaluation.signal
    return {
        "version": "0-local",
        "id": signal.event_id,
        "detail-type": SIMULATED_DETAIL_TYPE,
        "source": SIMULATED_SOURCE,
        "time": signal.timestamp,
        "region": "local",
        "resources": [f"simulated-ecs-service:{signal.service}"],
        "detail": {
            "alarmName": f"{signal.service}-health",
            "state": {"value": "ALARM"},
            "service": signal.service,
            "environment": signal.environment,
            "critical": signal.critical,
            "reasons": list(evaluation.reasons),
            "consecutiveFailures": evaluation.consecutive_failures,
        },
    }

What does the envelope contain?

• The local source identifies the event as simulation data.
• The event ID preserves the fixture's incident identity.
• The detail object carries the service context.
• The detail object also carries the breach evidence.
• The initial guard prevents an OK evaluation from becoming a routed alarm.

• Save self_healing/routing.py.
• Check cloudwatch_style_event() for Python syntax diagnostics.

You should see no syntax errors around the nested event dictionary.

Seeing a dictionary syntax error?

• Check that every nested opening brace has a matching closing brace.
• Check that detail-type remains inside quotation marks.

For help tracing the nesting, help me debug the local alarm envelope.

• Place the cursor two blank lines below the closing brace of cloudwatch_style_event().
• Add the router validation by copying this code:

class EventBridgeStyleRouter:
    def route(self, event: dict[str, object]) -> dict[str, object]:
        detail = event.get("detail")
        if not isinstance(detail, dict):
            raise ValueError("Event detail must be an object")
        state = detail.get("state")
        if not isinstance(state, dict):
            raise ValueError("Event state must be an object")
        if (
            event.get("source") != SIMULATED_SOURCE
            or event.get("detail-type") != SIMULATED_DETAIL_TYPE
            or state.get("value") != "ALARM"
        ):
            raise ValueError("Event does not match the simulated alarm rule")
        return event

How does the router fail closed?

• The router requires detail to be an object.
• The router requires state to be an object.
• The router compares the source with the local simulation source.
• The router compares the detail type with the local alarm detail type.
• The router accepts only an ALARM state.

• Save self_healing/routing.py.
• Check the complete router class for Python syntax diagnostics.

You should see no syntax errors around the nested validation checks.

Seeing a router validation error?

• Check that each comparison remains inside the multiline if condition.
• Check that the accepted event returns after the validation block.

If the condition is still unclear, help me debug EventBridgeStyleRouter.route().

✔️ Awesome, I've got everything!

Your routing boundary now creates alarm-only envelopes and rejects events that do not match the local rule.

ⓧ I'd like to double check the full code

from self_healing.models import Evaluation


SIMULATED_SOURCE = "local.cloudwatch.simulation"
SIMULATED_DETAIL_TYPE = "Simulated CloudWatch Alarm State Change"


def cloudwatch_style_event(evaluation: Evaluation) -> dict[str, object]:
    if not evaluation.alarm:
        raise ValueError("Only alarm evaluations can become routed events")

    signal = evaluation.signal
    return {
        "version": "0-local",
        "id": signal.event_id,
        "detail-type": SIMULATED_DETAIL_TYPE,
        "source": SIMULATED_SOURCE,
        "time": signal.timestamp,
        "region": "local",
        "resources": [f"simulated-ecs-service:{signal.service}"],
        "detail": {
            "alarmName": f"{signal.service}-health",
            "state": {"value": "ALARM"},
            "service": signal.service,
            "environment": signal.environment,
            "critical": signal.critical,
            "reasons": list(evaluation.reasons),
            "consecutiveFailures": evaluation.consecutive_failures,
        },
    }


class EventBridgeStyleRouter:
    def route(self, event: dict[str, object]) -> dict[str, object]:
        detail = event.get("detail")
        if not isinstance(detail, dict):
            raise ValueError("Event detail must be an object")
        state = detail.get("state")
        if not isinstance(state, dict):
            raise ValueError("Event state must be an object")
        if (
            event.get("source") != SIMULATED_SOURCE
            or event.get("detail-type") != SIMULATED_DETAIL_TYPE
            or state.get("value") != "ALARM"
        ):
            raise ValueError("Event does not match the simulated alarm rule")
        return event

Before you run the detector, which signal do you expect to become the first alarm for payments-prod?

• Run the detector from the project terminal with this command:

python main.py detect

What should you see?

• The search-1 signal remains OK with a consecutive count of 1.
• The recovered search-2 signal remains OK with a consecutive count of 0.
• The pay-1 signal remains OK while its streak reaches 1.
• The pay-2 signal becomes the first ALARM for payments-prod.
• The summary prints Detection accuracy: 100.0% (17/17).
• The final line prints Alarms: 8.

That safety gate is working. A brief search-prod failure stays quiet while the sustained payments-prod outage becomes actionable.

Detection results do not match?

• Check that healthy signals reset their service count to 0.
• Check that the alarm condition includes the configured consecutive-failure threshold.
• Check that each threshold uses a strict greater-than comparison.

For help comparing the output, help me debug my sustained failure detection results.

Your pipeline now separates isolated noise from sustained failure evidence. Next, you will replay one routed incident through an unsafe controller to expose duplicate remediation.

Expose Duplicate Recovery

Your detector now turns sustained failures into eight routed incidents. The payments service first reaches alarm state at pay-2.

Repeated delivery exposes a dangerous gap in a controller without idempotency. In this step, you will replay one routed incident through a stateless controller and preserve the resulting recovery failure as evidence.

In this step, get ready to:

• Add a controller that requests a simulated restart for every delivery.
• Replay the same routed incident through that controller.
• Capture the unsafe decisions in a reproducible JSON report.

Build the unsafe controller

An unsafe recovery controller treats every delivery as a new incident. It keeps no state that could connect a repeated event to an earlier decision.

• Return to the self_healing folder in the VS Code Explorer.
• Select the New File button in the Explorer toolbar.
• Name the file controller.py.
• Add the unsafe recovery handler by pasting this code into controller.py:

from self_healing.models import RecoveryAction


class UnsafeController:
    def handle(self, event: dict[str, object]) -> dict[str, str]:
        detail = event["detail"]
        if not isinstance(detail, dict):
            raise ValueError("Event detail must be an object")
        return {
            "event_id": str(event["id"]),
            "service": str(detail["service"]),
            "requested_action": RecoveryAction.RESTART_TASK.value,
            "final_action": RecoveryAction.RESTART_TASK.value,
            "outcome": "UNSAFE_EXECUTED",
            "reason": "No idempotency, cooldown, or attempt limit was checked",
            "event_time": str(event["time"]),
        }

What does this controller do?

• The handle() method reads the routed event details needed for the decision record.
• The requested action always remains RESTART_TASK.
• The UNSAFE_EXECUTED outcome makes the missing safety checks visible in the evidence.
• The controller stores no processed event IDs or service recovery history.

• Save controller.py.
• Confirm the new file appears inside the self_healing folder in the Explorer.

You now have a callable controller that produces a complete recovery decision for any routed event.

Controller file missing from the package?

Check that controller.py sits inside self_healing next to models.py. A file created beside main.py belongs to the wrong package location.

If the file location still looks unclear, help me verify my controller package structure.

✔️ Awesome, I've got everything!

Great. Save controller.py before connecting it to the incident replay.

ⓧ I'd like to double check the full code

from self_healing.models import RecoveryAction


class UnsafeController:
    def handle(self, event: dict[str, object]) -> dict[str, str]:
        detail = event["detail"]
        if not isinstance(detail, dict):
            raise ValueError("Event detail must be an object")
        return {
            "event_id": str(event["id"]),
            "service": str(detail["service"]),
            "requested_action": RecoveryAction.RESTART_TASK.value,
            "final_action": RecoveryAction.RESTART_TASK.value,
            "outcome": "UNSAFE_EXECUTED",
            "reason": "No idempotency, cooldown, or attempt limit was checked",
            "event_time": str(event["time"]),
        }

How should this file compare?

Your file should contain one import and one UnsafeController class. Its handler should return the routed event identity with an UNSAFE_EXECUTED restart decision.

Connect the replay to the simulation

The simulation already creates the eight alarm events accepted by the EventBridge-style router. The unsafe run needs to select pay-2 from those events and deliver that same object repeatedly.

• Return to self_healing/simulation.py in the VS Code Explorer.
• Add this controller import above the existing detector import:

from self_healing.controller import UnsafeController

Why import the controller here?

The simulation coordinates fixture evaluation and routed event delivery. Importing UnsafeController lets the replay pass an accepted event into the intentionally stateless handler.

• Find the path constants below ROOT.
• Add this report directory path below POLICIES_PATH:

REPORTS_PATH = ROOT / "reports"

Why define a report path?

The shared REPORTS_PATH points every evidence writer at the same local directory. The unsafe baseline can now be regenerated without choosing a new destination.

• Scroll below run_detect() in self_healing/simulation.py.
• Add the unsafe replay function by pasting this code:

def run_unsafe() -> None:
    _, evaluations = build_evaluations()
    events = build_events(evaluations)
    event = next(item for item in events if item["id"] == "pay-2")
    controller = UnsafeController()
    actions = [controller.handle(event) for _ in range(3)]
    REPORTS_PATH.mkdir(parents=True, exist_ok=True)
    evidence = {"event_id": "pay-2", "unsafe_restart_requests": 3, "actions": actions}
    (REPORTS_PATH / "unsafe_failure.json").write_text(
        json.dumps(evidence, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )
    print("Unsafe restart requests: 3 for event pay-2")
    print("Evidence: reports/unsafe_failure.json")

What does the replay do?

• The function reuses build_evaluations() and build_events() to preserve the detector and router behavior.
• The event search selects the routed incident whose ID is pay-2.
• The list comprehension delivers the same event object to one controller three times.
• The evidence writer stores every returned decision in reports/unsafe_failure.json.

• Save self_healing/simulation.py.
• Confirm VS Code shows no syntax warning on run_unsafe().

Seeing an unresolved controller import?

Confirm controller.py is saved inside self_healing. Check that the class name is exactly UnsafeController.

If VS Code still cannot resolve the import, help me debug the UnsafeController import.

✔️ Awesome, I've got everything!

Your simulation can now select the routed pay-2 incident and preserve every unsafe decision.

ⓧ I'd like to double check the full code

import json
from pathlib import Path

from self_healing.controller import UnsafeController
from self_healing.detector import HealthDetector
from self_healing.models import Evaluation, HealthSignal
from self_healing.policy import PolicySet, load_policy_set
from self_healing.routing import EventBridgeStyleRouter, cloudwatch_style_event


ROOT = Path(__file__).resolve().parents[1]
SIGNALS_PATH = ROOT / "data" / "health_signals.json"
POLICIES_PATH = ROOT / "data" / "policies.json"
REPORTS_PATH = ROOT / "reports"


def load_signals(path: Path = SIGNALS_PATH) -> list[HealthSignal]:
    raw = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(raw, list):
        raise ValueError("Health fixture must be a list")
    return [HealthSignal.from_dict(item) for item in raw]


def build_evaluations(
    policy_path: Path = POLICIES_PATH,
) -> tuple[PolicySet, list[Evaluation]]:
    policies = load_policy_set(policy_path)
    detector = HealthDetector(
        consecutive_failures=int(policies.detection["consecutive_failures"]),
        latency_ms=float(policies.detection["latency_ms"]),
        error_rate=float(policies.detection["error_rate"]),
    )
    return policies, detector.evaluate_all(load_signals())


def build_events(evaluations: list[Evaluation]) -> list[dict[str, object]]:
    router = EventBridgeStyleRouter()
    return [
        router.route(cloudwatch_style_event(evaluation))
        for evaluation in evaluations
        if evaluation.alarm
    ]


def run_inspect() -> None:
    policies, evaluations = build_evaluations()
    signals = [evaluation.signal for evaluation in evaluations]
    print(f"Signals: {len(signals)}")
    print(f"Services: {len({signal.service for signal in signals})}")
    print(f"Expected alarms: {sum(signal.expected_alarm for signal in signals)}")
    print(f"Configured service policies: {len(policies.services)}")
    print("Policy validation: PASS")


def run_detect() -> None:
    _, evaluations = build_evaluations()
    for evaluation in evaluations:
        status = "ALARM" if evaluation.alarm else "OK"
        reasons = ",".join(evaluation.reasons) if evaluation.reasons else "none"
        print(
            f"[{status}] {evaluation.signal.service} "
            f"{evaluation.signal.event_id} "
            f"consecutive={evaluation.consecutive_failures} reasons={reasons}"
        )
    correct = sum(
        evaluation.alarm == evaluation.signal.expected_alarm
        for evaluation in evaluations
    )
    print(f"Detection accuracy: {100 * correct / len(evaluations):.1f}% "
          f"({correct}/{len(evaluations)})")
    print(f"Alarms: {sum(evaluation.alarm for evaluation in evaluations)}")


def run_unsafe() -> None:
    _, evaluations = build_evaluations()
    events = build_events(evaluations)
    event = next(item for item in events if item["id"] == "pay-2")
    controller = UnsafeController()
    actions = [controller.handle(event) for _ in range(3)]
    REPORTS_PATH.mkdir(parents=True, exist_ok=True)
    evidence = {"event_id": "pay-2", "unsafe_restart_requests": 3, "actions": actions}
    (REPORTS_PATH / "unsafe_failure.json").write_text(
        json.dumps(evidence, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )
    print("Unsafe restart requests: 3 for event pay-2")
    print("Evidence: reports/unsafe_failure.json")

How should this file compare?

The file should retain the existing fixture loading, evaluation, routing, inspection, and detection functions. The new pieces are the controller import, REPORTS_PATH, and run_unsafe().

Replay the duplicated incident

The controller has no memory between calls. Before you run the replay, consider how many restart requests it records for three deliveries of the same event.

• Run the unsafe scenario from the terminal panel in the open workspace:

python main.py unsafe

What just happened?

Your terminal prints Unsafe restart requests: 3 for event pay-2. One routed incident produced three simulated restart requests.

The command also prints the path to reports/unsafe_failure.json. This is the reproducible baseline for the missing safety controls.

• Expand the reports folder in the VS Code Explorer.
• Select unsafe_failure.json to inspect the generated evidence.
• Confirm the top-level event_id value is pay-2.
• Confirm the actions array contains three records with the UNSAFE_EXECUTED outcome.

You reproduced the recovery-loop risk on demand. The same incident now has three audit-ready examples of an unsafe restart decision.

Report file missing?

Check that the terminal is using the open project workspace before rerunning the command. Confirm that run_unsafe() creates REPORTS_PATH before writing the file.

If the command completes without creating the report, help me trace the unsafe evidence path.

That is the failure exposed: one incident can trigger repeated remediation when the controller trusts every delivery. Next, you will turn this baseline into a safety contract with persistent controls.

Enforce Safe Recovery

Your unsafe baseline proves that one duplicated incident can trigger three restart requests. You now have evidence of the recovery loop that this controller must stop.

Safe recovery needs a fail-closed policy boundary. It also needs persistent idempotency claims in SQLite. This step adds those safeguards before any simulated restart can execute.

In this step, get ready to:

• Reject unsafe recovery policies before controller execution.
• Persist incident claims plus recovery history in SQLite.
• Enforce recovery controls in a deterministic order.

Validate recovery policies

A recovery policy controls which action a service may request. Fail-closed validation rejects malformed or dangerous JSON before the controller handles an incident.

• In the Explorer sidebar of Visual Studio Code, select the existing self_healing folder.
• Create policy.py inside self_healing with the New File control.
• Add the policy data structures plus numeric validator by pasting this first section:

import json
from dataclasses import dataclass
from pathlib import Path

from self_healing.models import RecoveryAction


@dataclass(frozen=True)
class RecoveryPolicy:
    action: RecoveryAction
    cooldown_seconds: int
    max_attempts: int
    allow_critical_restart: bool


@dataclass(frozen=True)
class PolicySet:
    detection: dict[str, float | int]
    default: RecoveryPolicy
    services: dict[str, RecoveryPolicy]

    def for_service(self, service: str) -> RecoveryPolicy:
        return self.services.get(service, self.default)


def _number(value: object, name: str, minimum: float) -> float:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ValueError(f"{name} must be numeric")
    number = float(value)
    if number < minimum:
        raise ValueError(f"{name} must be at least {minimum}")
    return number

What does this code do?

• RecoveryPolicy stores one requested action plus its cooldown and attempt limits.
• PolicySet keeps the detector configuration beside the default and service-specific policies.
• _number() rejects booleans and non-numeric values before checking the required minimum.

• Continue policy.py with the policy shape validation:

def _policy(raw: object, location: str) -> RecoveryPolicy:
    if not isinstance(raw, dict):
        raise ValueError(f"{location} must be an object")
    expected = {
        "action",
        "cooldown_seconds",
        "max_attempts",
        "allow_critical_restart",
    }
    if set(raw) != expected:
        raise ValueError(f"{location} contains missing or unknown keys")

Why check exact keys?

The equality check rejects missing fields plus unrecognized fields. A misspelled safety setting cannot be silently ignored.

• Complete the _policy() function with action and critical-workload validation:

    try:
        action = RecoveryAction(str(raw["action"]))
    except ValueError as error:
        raise ValueError(f"{location} has an unknown recovery action") from error

    cooldown = _number(raw["cooldown_seconds"], "cooldown_seconds", 0)
    attempts = _number(raw["max_attempts"], "max_attempts", 1)
    allow_critical = raw["allow_critical_restart"]
    if not isinstance(allow_critical, bool):
        raise ValueError("allow_critical_restart must be boolean")
    if allow_critical:
        raise ValueError("Unsafe policy rejected: critical restart override is prohibited")

    return RecoveryPolicy(
        action=action,
        cooldown_seconds=int(cooldown),
        max_attempts=int(attempts),
        allow_critical_restart=allow_critical,
    )

How does validation fail closed?

• RecoveryAction rejects action names outside the four defined controller outcomes.
• Cooldowns may be zero while attempt limits must be at least one.
• A critical restart override always raises an error before the controller starts.

• Add the first half of load_policy_set() below _policy() to validate the document structure:

def load_policy_set(path: Path) -> PolicySet:
    document = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(document, dict):
        raise ValueError("Policy document must be an object")
    if set(document) != {"detection", "default", "services"}:
        raise ValueError("Policy document contains missing or unknown top-level keys")

    detection = document["detection"]
    if not isinstance(detection, dict):
        raise ValueError("detection must be an object")
    if set(detection) != {"consecutive_failures", "latency_ms", "error_rate"}:
        raise ValueError("detection contains missing or unknown keys")

    consecutive = _number(
        detection["consecutive_failures"], "consecutive_failures", 2
    )
    latency = _number(detection["latency_ms"], "latency_ms", 0.000001)
    error_rate = _number(detection["error_rate"], "error_rate", 0.000001)

What gets validated first?

The loader validates each container before reading its values. Detector thresholds receive the same numeric safeguards as recovery settings.

• Complete load_policy_set() with service-policy conversion plus the final PolicySet result:

    raw_services = document["services"]
    if not isinstance(raw_services, dict):
        raise ValueError("services must be an object")
    services = {
        str(service): _policy(raw, f"services.{service}")
        for service, raw in raw_services.items()
    }

    return PolicySet(
        detection={
            "consecutive_failures": int(consecutive),
            "latency_ms": latency,
            "error_rate": error_rate,
        },
        default=_policy(document["default"], "default"),
        services=services,
    )

How are service policies resolved?

Each named service passes through the same strict validator. PolicySet returns the default policy whenever a routed service has no explicit entry.

• Save self_healing/policy.py.
• Confirm that policy.py now appears under self_healing in the Explorer sidebar.

Policy file showing an error?

Check that each continuation stays inside its function indentation. Confirm that the closing parentheses remain below their matching constructors.

If the editor still highlights the file, ask for help checking the policy module structure.

Persist recovery history in SQLite

In-memory checks disappear when a process exits. AuditStore gives duplicate claims plus recovery attempts a durable home in data/audit.db.

• In the Explorer sidebar, select the existing self_healing folder.
• Create store.py inside self_healing with the New File control.
• Add the database connection plus the first two tables by pasting this section:

import sqlite3
from pathlib import Path


class AuditStore:
    def __init__(self, path: Path) -> None:
        path.parent.mkdir(parents=True, exist_ok=True)
        self.connection = sqlite3.connect(path)
        self.connection.row_factory = sqlite3.Row
        self.connection.executescript(
            """
            CREATE TABLE IF NOT EXISTS processed_events (
                event_id TEXT PRIMARY KEY,
                service TEXT NOT NULL,
                received_at TEXT NOT NULL
            );

            CREATE TABLE IF NOT EXISTS recovery_state (
                service TEXT PRIMARY KEY,
                attempts INTEGER NOT NULL,
                last_attempt_at TEXT
            );

What do these tables protect?

• processed_events uses event_id as a primary key so one incident can receive only one successful claim.
• recovery_state stores the restart count plus the latest simulated attempt time for each service.

• Complete the schema with the append-only audit table:

            CREATE TABLE IF NOT EXISTS audit_events (
                audit_id INTEGER PRIMARY KEY AUTOINCREMENT,
                event_id TEXT NOT NULL,
                service TEXT NOT NULL,
                requested_action TEXT NOT NULL,
                final_action TEXT NOT NULL,
                outcome TEXT NOT NULL,
                reason TEXT NOT NULL,
                event_time TEXT NOT NULL
            );
            """
        )
        self.connection.commit()

Why keep append-only audits?

Each delivery receives its own audit row. Duplicate suppression and blocked actions remain visible evidence instead of disappearing behind a final service state.

• Add the persistent event claim below __init__().

    def reserve_event(self, event_id: str, service: str, event_time: str) -> bool:
        cursor = self.connection.execute(
            """
            INSERT OR IGNORE INTO processed_events(event_id, service, received_at)
            VALUES (?, ?, ?)
            """,
            (event_id, service, event_time),
        )
        self.connection.commit()
        return cursor.rowcount == 1

How does the claim suppress duplicates?

The primary key accepts the first event ID. INSERT OR IGNORE leaves the existing row untouched when the same ID arrives again.

• Add the recovery-state lookup below reserve_event().

    def recovery_state(self, service: str) -> tuple[int, str | None]:
        row = self.connection.execute(
            "SELECT attempts, last_attempt_at FROM recovery_state WHERE service = ?",
            (service,),
        ).fetchone()
        if row is None:
            return 0, None
        return int(row["attempts"]), row["last_attempt_at"]

What state comes back?

A service with no recovery history starts at zero attempts. Existing services return their persisted attempt count plus the last attempt timestamp.

• Add record_restart() below recovery_state().

    def record_restart(self, service: str, event_time: str) -> None:
        attempts, _ = self.recovery_state(service)
        if attempts == 0:
            self.connection.execute(
                """
                INSERT INTO recovery_state(service, attempts, last_attempt_at)
                VALUES (?, ?, ?)
                """,
                (service, 1, event_time),
            )
        else:
            self.connection.execute(
                """
                UPDATE recovery_state
                SET attempts = ?, last_attempt_at = ?
                WHERE service = ?
                """,
                (attempts + 1, event_time, service),
            )
        self.connection.commit()

When does an attempt increase?

The store records an attempt only after the controller permits a simulated restart. Blocked decisions do not consume the service attempt budget.

• Add record_audit() below record_restart().

    def record_audit(self, decision: dict[str, str]) -> None:
        self.connection.execute(
            """
            INSERT INTO audit_events(
                event_id,
                service,
                requested_action,
                final_action,
                outcome,
                reason,
                event_time
            ) VALUES (?, ?, ?, ?, ?, ?, ?)
            """,
            (
                decision["event_id"],
                decision["service"],
                decision["requested_action"],
                decision["final_action"],
                decision["outcome"],
                decision["reason"],
                decision["event_time"],
            ),
        )
        self.connection.commit()

What makes an audit useful?

The row preserves the requested action beside the final action. Its outcome plus reason explain which safety control changed the decision.

• Finish AuditStore with audit retrieval plus connection cleanup:

    def all_audits(self) -> list[dict[str, object]]:
        rows = self.connection.execute(
            """
            SELECT audit_id, event_id, service, requested_action, final_action,
                   outcome, reason, event_time
            FROM audit_events
            ORDER BY audit_id
            """
        ).fetchall()
        return [dict(row) for row in rows]

    def close(self) -> None:
        self.connection.close()

Why order audits by ID?

audit_id preserves controller delivery order for later evidence. close() releases the database connection before another controller instance opens the same file.

• Save self_healing/store.py.
• Confirm that store.py now appears under self_healing in the Explorer sidebar.

Store file showing SQL errors?

Check that each table definition remains inside the triple-quoted string. Confirm that each parameter list has the same number of values as its question-mark placeholders.

For a closer comparison, ask for help checking the SQLite schema and bindings.

Enforce controls and verify the safe run

The store supplies durable facts. SafeRecoveryController now evaluates those facts in a fixed order so every delivery reaches one auditable outcome.

• In self_healing/controller.py, replace the current import section with this expanded import section:

from datetime import datetime

from self_healing.models import RecoveryAction
from self_healing.policy import PolicySet
from self_healing.store import AuditStore

Why does the controller need these imports?

datetime calculates cooldown age from deterministic event timestamps. PolicySet supplies the requested action while AuditStore supplies persistent safety state.

• Keep the existing UnsafeController unchanged.
• Add SafeRecoveryController below UnsafeController with its shared audit helper:

class SafeRecoveryController:
    def __init__(self, policies: PolicySet, store: AuditStore) -> None:
        self.policies = policies
        self.store = store

    def _finish(
        self,
        event_id: str,
        service: str,
        requested: RecoveryAction,
        final: RecoveryAction,
        outcome: str,
        reason: str,
        event_time: str,
    ) -> dict[str, str]:
        decision = {
            "event_id": event_id,
            "service": service,
            "requested_action": requested.value,
            "final_action": final.value,
            "outcome": outcome,
            "reason": reason,
            "event_time": event_time,
        }
        self.store.record_audit(decision)
        return decision

Why use one finish path?

Every branch sends the same decision shape to record_audit(). Executed actions and safety blocks therefore produce comparable evidence.

• Start handle() by extracting the routed event context:

    def handle(self, event: dict[str, object]) -> dict[str, str]:
        detail = event["detail"]
        if not isinstance(detail, dict):
            raise ValueError("Event detail must be an object")

        event_id = str(event["id"])
        service = str(detail["service"])
        event_time = str(event["time"])
        environment = str(detail["environment"])
        critical = bool(detail["critical"])
        policy = self.policies.for_service(service)
        requested = policy.action

What context drives a decision?

The controller reads identity, service, timestamp, environment, and criticality from the routed event. The service name resolves the explicit recovery policy.

• Add the duplicate claim as the first safety decision:

        if not self.store.reserve_event(event_id, service, event_time):
            return self._finish(
                event_id,
                service,
                requested,
                RecoveryAction.NO_ACTION,
                "DUPLICATE_SUPPRESSED",
                "The event ID already has a persistent claim",
                event_time,
            )

Why claim before policy evaluation?

A repeated event stops before any policy can request another action. The duplicate still receives an audit row that explains its suppression.

• Add the direct NO_ACTION plus WAIT_AND_RECHECK branches:

        if requested is RecoveryAction.NO_ACTION:
            return self._finish(
                event_id,
                service,
                requested,
                RecoveryAction.NO_ACTION,
                "NO_ACTION",
                "The explicit policy requires observation only",
                event_time,
            )

        if requested is RecoveryAction.WAIT_AND_RECHECK:
            return self._finish(
                event_id,
                service,
                requested,
                RecoveryAction.WAIT_AND_RECHECK,
                "WAIT_AND_RECHECK",
                "The explicit policy defers remediation",
                event_time,
            )

Why return explicit non-restart actions early?

Observation and waiting policies need no restart safeguards. Their early returns preserve the exact policy decision in the audit history.

• Add explicit escalation plus critical-production protection:

        if requested is RecoveryAction.ESCALATE:
            return self._finish(
                event_id,
                service,
                requested,
                RecoveryAction.ESCALATE,
                "ESCALATED",
                "The explicit policy requires human review",
                event_time,
            )

        if critical and environment == "prod":
            return self._finish(
                event_id,
                service,
                requested,
                RecoveryAction.ESCALATE,
                "BLOCKED_CRITICAL",
                "Critical production workloads cannot be restarted automatically",
                event_time,
            )

How are critical workloads protected?

A critical production service cannot reach simulated restart execution. Its restart request becomes an escalation with a specific audit reason.

• Add the maximum-attempt check below the critical-workload guard:

        attempts, last_attempt_at = self.store.recovery_state(service)
        if attempts >= policy.max_attempts:
            return self._finish(
                event_id,
                service,
                requested,
                RecoveryAction.ESCALATE,
                "MAX_ATTEMPTS_ESCALATED",
                "The maximum simulated restart attempts has been reached",
                event_time,
            )

Why check attempts before cooldown?

An exhausted service must escalate regardless of elapsed time. This ordering prevents a later timestamp from bypassing the bounded retry policy.

• Add the cooldown calculation below the attempt check:

        if last_attempt_at is not None:
            elapsed = (
                datetime.fromisoformat(event_time)
                - datetime.fromisoformat(last_attempt_at)
            ).total_seconds()
            if elapsed < policy.cooldown_seconds:
                return self._finish(
                    event_id,
                    service,
                    requested,
                    RecoveryAction.WAIT_AND_RECHECK,
                    "COOLDOWN_BLOCKED",
                    "The service is still inside its recovery cooldown",
                    event_time,
                )

Why use event timestamps?

Fixture timestamps make cooldown results deterministic. The controller needs no sleep or wall-clock dependency to reproduce its decisions.

• Finish handle() with the permitted restart path:

        self.store.record_restart(service, event_time)
        return self._finish(
            event_id,
            service,
            requested,
            RecoveryAction.RESTART_TASK,
            "RECOVERY_EXECUTED",
            "A task restart was simulated and no AWS API was called",
            event_time,
        )

What does execution mean here?

The controller updates local recovery state plus the audit log. RECOVERY_EXECUTED represents a simulated restart decision with no AWS API call.

• Save self_healing/controller.py.
• In self_healing/simulation.py, replace the import section with this final import section:

import json
from pathlib import Path

from self_healing.controller import SafeRecoveryController, UnsafeController
from self_healing.detector import HealthDetector
from self_healing.models import Evaluation, HealthSignal
from self_healing.policy import PolicySet, load_policy_set
from self_healing.reporting import build_report, write_reports
from self_healing.routing import EventBridgeStyleRouter, cloudwatch_style_event
from self_healing.store import AuditStore

What changes in the simulation runner?

The runner now has access to validated policies plus the safe controller. It can also open the audit store and pass completed decisions into the existing reporting helpers.

• Find the existing path constants near the top of simulation.py.
• Add the audit database path below UNSAFE_POLICIES_PATH:

DATABASE_PATH = ROOT / "data" / "audit.db"

Why keep the database path central?

One shared path lets the safe run reset its generated state predictably. Future controller instances can reopen the same database to prove persistence.

• Add the first half of run_safe() below run_unsafe():

def run_safe() -> None:
    policies, evaluations = build_evaluations()
    events = build_events(evaluations)
    DATABASE_PATH.unlink(missing_ok=True)
    store = AuditStore(DATABASE_PATH)
    controller = SafeRecoveryController(policies, store)

    for event in events:
        controller.handle(event)
        if event["id"] == "pay-2":
            controller.handle(event)

    audits = store.all_audits()
    store.close()
    report = build_report(evaluations, audits)
    write_reports(report, REPORTS_PATH)
    metrics = report["metrics"]
    if not isinstance(metrics, dict):
        raise ValueError("metrics must be an object")

How does the safe scenario run?

• The previous audit database is removed so every benchmark starts from the same state.
• All eight routed incidents pass through the safe controller.
• pay-2 receives one extra delivery to exercise persistent duplicate suppression.

• Complete run_safe() with its deterministic summary:

    print(f"Incidents: {metrics['incidents_detected']}")
    print(f"Controller deliveries: {metrics['controller_deliveries']}")
    print(f"Recovery actions executed: {metrics['recovery_actions_executed']}")
    print(
        "Duplicate deliveries suppressed: "
        f"{metrics['duplicate_deliveries_suppressed']}"
    )
    print(f"Escalations: {metrics['escalations']}")
    print(f"Unsafe actions executed: {metrics['unsafe_actions_executed']}")
    print("Reports: reports/reliability_report.json, "
          "reports/reliability_report.md, reports/audit.jsonl")

What does the summary expose?

The printed counters separate controller traffic from executed recovery actions. Suppressions and escalations remain visible as successful safety outcomes.

• Save self_healing/simulation.py.

✔️ Awesome, I've got everything!

Your policy loader, SQLite store, safe controller, and simulation runner are ready. Save every changed file before running the verification.

ⓧ I'd like to double check the full code

import json
from pathlib import Path

from self_healing.controller import SafeRecoveryController, UnsafeController
from self_healing.detector import HealthDetector
from self_healing.models import Evaluation, HealthSignal
from self_healing.policy import PolicySet, load_policy_set
from self_healing.reporting import build_report, write_reports
from self_healing.routing import EventBridgeStyleRouter, cloudwatch_style_event
from self_healing.store import AuditStore


ROOT = Path(__file__).resolve().parents[1]
SIGNALS_PATH = ROOT / "data" / "health_signals.json"
POLICIES_PATH = ROOT / "data" / "policies.json"
UNSAFE_POLICIES_PATH = ROOT / "data" / "unsafe_policies.json"
DATABASE_PATH = ROOT / "data" / "audit.db"
REPORTS_PATH = ROOT / "reports"


def load_signals(path: Path = SIGNALS_PATH) -> list[HealthSignal]:
    raw = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(raw, list):
        raise ValueError("Health fixture must be a list")
    return [HealthSignal.from_dict(item) for item in raw]


def build_evaluations(
    policy_path: Path = POLICIES_PATH,
) -> tuple[PolicySet, list[Evaluation]]:
    policies = load_policy_set(policy_path)
    detector = HealthDetector(
        consecutive_failures=int(policies.detection["consecutive_failures"]),
        latency_ms=float(policies.detection["latency_ms"]),
        error_rate=float(policies.detection["error_rate"]),
    )
    return policies, detector.evaluate_all(load_signals())


def build_events(evaluations: list[Evaluation]) -> list[dict[str, object]]:
    router = EventBridgeStyleRouter()
    return [
        router.route(cloudwatch_style_event(evaluation))
        for evaluation in evaluations
        if evaluation.alarm
    ]


def run_inspect() -> None:
    policies, evaluations = build_evaluations()
    signals = [evaluation.signal for evaluation in evaluations]
    print(f"Signals: {len(signals)}")
    print(f"Services: {len({signal.service for signal in signals})}")
    print(f"Expected alarms: {sum(signal.expected_alarm for signal in signals)}")
    print(f"Configured service policies: {len(policies.services)}")
    print("Policy validation: PASS")


def run_detect() -> None:
    _, evaluations = build_evaluations()
    for evaluation in evaluations:
        status = "ALARM" if evaluation.alarm else "OK"
        reasons = ",".join(evaluation.reasons) if evaluation.reasons else "none"
        print(
            f"[{status}] {evaluation.signal.service} "
            f"{evaluation.signal.event_id} "
            f"consecutive={evaluation.consecutive_failures} reasons={reasons}"
        )
    correct = sum(
        evaluation.alarm == evaluation.signal.expected_alarm
        for evaluation in evaluations
    )
    print(f"Detection accuracy: {100 * correct / len(evaluations):.1f}% "
          f"({correct}/{len(evaluations)})")
    print(f"Alarms: {sum(evaluation.alarm for evaluation in evaluations)}")


def run_unsafe() -> None:
    _, evaluations = build_evaluations()
    events = build_events(evaluations)
    event = next(item for item in events if item["id"] == "pay-2")
    controller = UnsafeController()
    actions = [controller.handle(event) for _ in range(3)]
    REPORTS_PATH.mkdir(parents=True, exist_ok=True)
    evidence = {"event_id": "pay-2", "unsafe_restart_requests": 3, "actions": actions}
    (REPORTS_PATH / "unsafe_failure.json").write_text(
        json.dumps(evidence, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )
    print("Unsafe restart requests: 3 for event pay-2")
    print("Evidence: reports/unsafe_failure.json")


def run_safe() -> None:
    policies, evaluations = build_evaluations()
    events = build_events(evaluations)
    DATABASE_PATH.unlink(missing_ok=True)
    store = AuditStore(DATABASE_PATH)
    controller = SafeRecoveryController(policies, store)

    for event in events:
        controller.handle(event)
        if event["id"] == "pay-2":
            controller.handle(event)

    audits = store.all_audits()
    store.close()
    report = build_report(evaluations, audits)
    write_reports(report, REPORTS_PATH)
    metrics = report["metrics"]
    if not isinstance(metrics, dict):
        raise ValueError("metrics must be an object")

    print(f"Incidents: {metrics['incidents_detected']}")
    print(f"Controller deliveries: {metrics['controller_deliveries']}")
    print(f"Recovery actions executed: {metrics['recovery_actions_executed']}")
    print(
        "Duplicate deliveries suppressed: "
        f"{metrics['duplicate_deliveries_suppressed']}"
    )
    print(f"Escalations: {metrics['escalations']}")
    print(f"Unsafe actions executed: {metrics['unsafe_actions_executed']}")
    print("Reports: reports/reliability_report.json, "
          "reports/reliability_report.md, reports/audit.jsonl")

Before you run the safe controller, which number do you expect to shrink most sharply compared with the three unsafe restart requests?

• Run the safe recovery scenario from the existing project terminal with this command:

python main.py safe

What should you see?

• Incidents prints 8.
• Controller deliveries prints 9.
• Recovery actions executed prints 2.
• Duplicate deliveries suppressed prints 1.
• Escalations prints 3.
• Unsafe actions executed prints 0.

That is the safety boundary working. Nine deliveries now produce two permitted restart actions while every duplicate or unsafe path remains visible in the audit history.

• Confirm that audit.db now appears under the existing data folder in the Explorer sidebar.

Safe command not reaching the summary?

Check the first traceback line that points into your project. A policy error usually means a key or indentation mismatch in policy.py.

A database error usually points to an incomplete SQL string or parameter tuple in store.py. Ask for help tracing the failed safety layer.

Your controller now turns duplicated incidents into deterministic evidence instead of repeated recovery loops. Next, you will make those safety claims executable with tests and reviewable reports.

Test and Report Reliability

Your safe controller now converts routed incidents into deterministic decisions. Its SQLite history preserves every executed action plus every blocked action.

Those safety claims need executable evidence that survives a process restart. This step adds deterministic tests plus reviewable reliability reports.

In this step, get ready to:

• Test the detector plus every recovery safeguard with deterministic fixtures.
• Calculate reliability metrics from detector evaluations plus persisted audit rows.
• Generate JSON plus Markdown plus JSONL evidence for the safety contract.

Add deterministic reliability tests

pytest turns each reliability claim into a repeatable check. Temporary databases supplied through tmp_path keep controller tests isolated from the audit history in data/audit.db.

• In the Visual Studio Code file tree, select the existing tests folder.
• Create test_self_healing.py inside the selected folder.
• Add the shared imports plus helper functions by copying this code into tests/test_self_healing.py:

from pathlib import Path

from self_healing.controller import SafeRecoveryController
from self_healing.detector import HealthDetector
from self_healing.policy import load_policy_set
from self_healing.simulation import (
    POLICIES_PATH,
    UNSAFE_POLICIES_PATH,
    build_evaluations,
    build_events,
    load_signals,
)
from self_healing.store import AuditStore


def event_by_id(event_id: str) -> dict[str, object]:
    _, evaluations = build_evaluations()
    return next(event for event in build_events(evaluations) if event["id"] == event_id)


def controller(tmp_path: Path) -> tuple[SafeRecoveryController, AuditStore]:
    store = AuditStore(tmp_path / "audit.db")
    return SafeRecoveryController(load_policy_set(POLICIES_PATH), store), store

What Do These Helpers Provide?

• The imports connect each test to the real detector plus policy loader plus controller plus audit store.
• The event_by_id() helper selects one routed alarm from the deterministic fixture.
• The controller() helper creates an isolated database for each controller test.

• Save tests/test_self_healing.py.
• Confirm the file tree shows test_self_healing.py directly inside tests.

Cannot Import the Project Modules?

Confirm test_self_healing.py sits inside the existing tests folder. Confirm the workspace virtual environment remains selected.

Still blocked? Help me diagnose the test import paths in this local Python project.

The first pair checks the detector's sustained-evidence rule. A transient breach remains quiet while a persistent outage alarms after its second consecutive failure.

• Add the detector tests below the controller() helper:

def test_transient_failure_does_not_alarm() -> None:
    policies = load_policy_set(POLICIES_PATH)
    detector = HealthDetector(
        int(policies.detection["consecutive_failures"]),
        float(policies.detection["latency_ms"]),
        float(policies.detection["error_rate"]),
    )
    search = [signal for signal in load_signals() if signal.service == "search-prod"]
    assert [result.alarm for result in detector.evaluate_all(search)] == [False, False]


def test_persistent_outage_alarms_after_second_failure() -> None:
    policies = load_policy_set(POLICIES_PATH)
    detector = HealthDetector(
        int(policies.detection["consecutive_failures"]),
        float(policies.detection["latency_ms"]),
        float(policies.detection["error_rate"]),
    )
    payments = [
        signal for signal in load_signals() if signal.service == "payments-prod"
    ]
    assert [result.alarm for result in detector.evaluate_all(payments)] == [
        False,
        True,
        True,
        True,
        True,
    ]

What Do the Detector Tests Prove?

• The transient test confirms that one breached search-prod signal cannot trigger an alarm.
• The persistent test confirms that payments-prod starts alarming when the configured evidence threshold is met.

• Save tests/test_self_healing.py.
• Run the current detector tests from the existing terminal with this command:

python -m pytest -q

What Should This Run Confirm?

Both detector checks pass against the unchanged fixture. This proves the labeled transient plus persistent scenarios match the sustained-breach rule.

Are the Detector Tests Failing?

Verify the selected interpreter points to the workspace .venv. Confirm data/health_signals.json still contains the deterministic fixture from earlier.

Need a closer diagnosis? Help me trace the failing detector assertion.

Duplicate claims plus cooldowns stop two different recovery-loop paths. One blocks a repeated event ID while the other blocks a new incident that arrives too soon.

• Add the duplicate plus cooldown tests below the detector tests:

def test_duplicate_event_is_suppressed(tmp_path: Path) -> None:
    safe, store = controller(tmp_path)
    event = event_by_id("pay-2")
    assert safe.handle(event)["outcome"] == "RECOVERY_EXECUTED"
    assert safe.handle(event)["outcome"] == "DUPLICATE_SUPPRESSED"
    store.close()


def test_cooldown_blocks_early_restart(tmp_path: Path) -> None:
    safe, store = controller(tmp_path)
    assert safe.handle(event_by_id("pay-2"))["outcome"] == "RECOVERY_EXECUTED"
    assert safe.handle(event_by_id("pay-3"))["outcome"] == "COOLDOWN_BLOCKED"
    store.close()

How Do These Controls Differ?

• The duplicate test delivers pay-2 twice. Its persistent event claim allows only the first delivery to execute.
• The cooldown test delivers a new pay-3 incident before the configured wait has elapsed. The controller records a blocked decision.

• Save tests/test_self_healing.py.
• Run the expanded safety suite with this command:

python -m pytest -q

What Should This Run Confirm?

The detector checks plus both early recovery controls pass. Each controller test uses a separate temporary SQLite database.

Is a Recovery Outcome Different?

Confirm the duplicate test reuses pay-2. Confirm the cooldown test delivers pay-3 after pay-2.

Still seeing a mismatch? Help me inspect the duplicate claim or cooldown outcome.

The next pair covers hard recovery boundaries. The attempt ceiling escalates repeated restarts while critical-production protection blocks automatic remediation.

• Add the attempt-limit plus critical-workload tests below the cooldown test:

def test_maximum_attempts_escalates(tmp_path: Path) -> None:
    safe, store = controller(tmp_path)
    assert safe.handle(event_by_id("pay-2"))["outcome"] == "RECOVERY_EXECUTED"
    assert safe.handle(event_by_id("pay-4"))["outcome"] == "RECOVERY_EXECUTED"
    assert safe.handle(event_by_id("pay-5"))["outcome"] == "MAX_ATTEMPTS_ESCALATED"
    store.close()


def test_critical_production_restart_is_blocked(tmp_path: Path) -> None:
    safe, store = controller(tmp_path)
    decision = safe.handle(event_by_id("admin-2"))
    assert decision["outcome"] == "BLOCKED_CRITICAL"
    assert decision["final_action"] == "ESCALATE"
    store.close()

Which Boundaries Are Covered?

• The attempt test allows two simulated restarts for payments-prod. Its later incident becomes an escalation.
• The critical-workload test confirms that admin-prod produces an escalation instead of an automatic restart.

• Save tests/test_self_healing.py.
• Run the expanded safety suite with this command:

python -m pytest -q

What Should This Run Confirm?

The suite now passes through the attempt-limit plus critical-production scenarios. Blocked actions are expected safety outcomes.

Did a Restart Execute Unexpectedly?

Confirm pay-2 plus pay-4 execute before pay-5 reaches the attempt ceiling. Confirm the critical test routes admin-2.

Need help following the decision order? Help me trace the unexpected restart.

The final pair crosses process boundaries. Unsafe policy input fails closed while the persisted recovery state remains available to a new controller instance.

• Add the policy-rejection plus process-restart tests at the end of tests/test_self_healing.py:

def test_unsafe_policy_fails_closed() -> None:
    try:
        load_policy_set(UNSAFE_POLICIES_PATH)
    except ValueError as error:
        assert "Unsafe policy rejected" in str(error)
    else:
        raise AssertionError("Unsafe policy was accepted")


def test_recovery_history_survives_controller_restart(tmp_path: Path) -> None:
    database = tmp_path / "restart.db"
    policies = load_policy_set(POLICIES_PATH)

    first_store = AuditStore(database)
    first_controller = SafeRecoveryController(policies, first_store)
    first_controller.handle(event_by_id("pay-2"))
    first_controller.handle(event_by_id("pay-4"))
    first_store.close()

    second_store = AuditStore(database)
    second_controller = SafeRecoveryController(policies, second_store)
    decision = second_controller.handle(event_by_id("pay-5"))
    assert decision["outcome"] == "MAX_ATTEMPTS_ESCALATED"
    assert second_store.recovery_state("payments-prod")[0] == 2
    second_store.close()

Why Test a Controller Restart?

• The unsafe-policy test proves that a critical restart override is rejected during policy loading.
• The restart test opens two controller instances against the same database. The second controller still sees two prior attempts.
• Persistent recovery state prevents a process restart from resetting the attempt ceiling.

• Save tests/test_self_healing.py.
• Run the complete safety suite with this command:

python -m pytest -q

What Should the Complete Suite Show?

The terminal ends with 8 passed. Strong result. Every detector plus controller safety claim now has an executable check.

Is the Complete Suite Failing?

Read the first failed test name before changing any implementation. That name identifies the safety contract that drifted.

Want help interpreting the failure? Help me diagnose the first failing reliability test.

✔️ Awesome, I've got everything!

Your tests/test_self_healing.py file now contains two helpers plus eight deterministic tests. Save it before continuing.

ⓧ I'd like to double check the full code

from pathlib import Path

from self_healing.controller import SafeRecoveryController
from self_healing.detector import HealthDetector
from self_healing.policy import load_policy_set
from self_healing.simulation import (
    POLICIES_PATH,
    UNSAFE_POLICIES_PATH,
    build_evaluations,
    build_events,
    load_signals,
)
from self_healing.store import AuditStore


def event_by_id(event_id: str) -> dict[str, object]:
    _, evaluations = build_evaluations()
    return next(event for event in build_events(evaluations) if event["id"] == event_id)


def controller(tmp_path: Path) -> tuple[SafeRecoveryController, AuditStore]:
    store = AuditStore(tmp_path / "audit.db")
    return SafeRecoveryController(load_policy_set(POLICIES_PATH), store), store


def test_transient_failure_does_not_alarm() -> None:
    policies = load_policy_set(POLICIES_PATH)
    detector = HealthDetector(
        int(policies.detection["consecutive_failures"]),
        float(policies.detection["latency_ms"]),
        float(policies.detection["error_rate"]),
    )
    search = [signal for signal in load_signals() if signal.service == "search-prod"]
    assert [result.alarm for result in detector.evaluate_all(search)] == [False, False]


def test_persistent_outage_alarms_after_second_failure() -> None:
    policies = load_policy_set(POLICIES_PATH)
    detector = HealthDetector(
        int(policies.detection["consecutive_failures"]),
        float(policies.detection["latency_ms"]),
        float(policies.detection["error_rate"]),
    )
    payments = [
        signal for signal in load_signals() if signal.service == "payments-prod"
    ]
    assert [result.alarm for result in detector.evaluate_all(payments)] == [
        False,
        True,
        True,
        True,
        True,
    ]


def test_duplicate_event_is_suppressed(tmp_path: Path) -> None:
    safe, store = controller(tmp_path)
    event = event_by_id("pay-2")
    assert safe.handle(event)["outcome"] == "RECOVERY_EXECUTED"
    assert safe.handle(event)["outcome"] == "DUPLICATE_SUPPRESSED"
    store.close()


def test_cooldown_blocks_early_restart(tmp_path: Path) -> None:
    safe, store = controller(tmp_path)
    assert safe.handle(event_by_id("pay-2"))["outcome"] == "RECOVERY_EXECUTED"
    assert safe.handle(event_by_id("pay-3"))["outcome"] == "COOLDOWN_BLOCKED"
    store.close()


def test_maximum_attempts_escalates(tmp_path: Path) -> None:
    safe, store = controller(tmp_path)
    assert safe.handle(event_by_id("pay-2"))["outcome"] == "RECOVERY_EXECUTED"
    assert safe.handle(event_by_id("pay-4"))["outcome"] == "RECOVERY_EXECUTED"
    assert safe.handle(event_by_id("pay-5"))["outcome"] == "MAX_ATTEMPTS_ESCALATED"
    store.close()


def test_critical_production_restart_is_blocked(tmp_path: Path) -> None:
    safe, store = controller(tmp_path)
    decision = safe.handle(event_by_id("admin-2"))
    assert decision["outcome"] == "BLOCKED_CRITICAL"
    assert decision["final_action"] == "ESCALATE"
    store.close()


def test_unsafe_policy_fails_closed() -> None:
    try:
        load_policy_set(UNSAFE_POLICIES_PATH)
    except ValueError as error:
        assert "Unsafe policy rejected" in str(error)
    else:
        raise AssertionError("Unsafe policy was accepted")


def test_recovery_history_survives_controller_restart(tmp_path: Path) -> None:
    database = tmp_path / "restart.db"
    policies = load_policy_set(POLICIES_PATH)

    first_store = AuditStore(database)
    first_controller = SafeRecoveryController(policies, first_store)
    first_controller.handle(event_by_id("pay-2"))
    first_controller.handle(event_by_id("pay-4"))
    first_store.close()

    second_store = AuditStore(database)
    second_controller = SafeRecoveryController(policies, second_store)
    decision = second_controller.handle(event_by_id("pay-5"))
    assert decision["outcome"] == "MAX_ATTEMPTS_ESCALATED"
    assert second_store.recovery_state("payments-prod")[0] == 2
    second_store.close()

What Should Match?

The file contains two helpers followed by eight tests. The final test reuses one database across two audit-store instances.

Generate reliability reports

The report engine combines detector evaluations plus persisted audit decisions into one evidence object. It writes structured JSON metrics plus a human-readable Markdown summary plus an append-only JSONL audit record.

• In the VS Code file tree, select the existing self_healing folder.
• Create reporting.py inside the selected folder.
• Add the complete reporting implementation shown in the double-check tab below.

✔️ Awesome, I've got everything!

Your reporting module calculates the metrics before writing all three evidence formats. Save self_healing/reporting.py.

ⓧ I'd like to double check the full code

import json
from pathlib import Path

from self_healing.models import Evaluation


def build_report(
    evaluations: list[Evaluation], audits: list[dict[str, object]]
) -> dict[str, object]:
    correct = sum(
        evaluation.alarm == evaluation.signal.expected_alarm
        for evaluation in evaluations
    )
    incidents = sum(evaluation.alarm for evaluation in evaluations)
    outcomes: dict[str, int] = {}
    for audit in audits:
        outcome = str(audit["outcome"])
        outcomes[outcome] = outcomes.get(outcome, 0) + 1

    restart_decisions = [
        audit
        for audit in audits
        if audit["requested_action"] == "RESTART_TASK"
        and audit["outcome"] != "DUPLICATE_SUPPRESSED"
    ]
    executed = outcomes.get("RECOVERY_EXECUTED", 0)
    success_rate = (
        round(100 * executed / len(restart_decisions), 1)
        if restart_decisions
        else 0.0
    )
    escalations = sum(
        outcomes.get(name, 0)
        for name in ("ESCALATED", "BLOCKED_CRITICAL", "MAX_ATTEMPTS_ESCALATED")
    )

    return {
        "metrics": {
            "signals_evaluated": len(evaluations),
            "incidents_detected": incidents,
            "controller_deliveries": len(audits),
            "detection_accuracy_percent": round(100 * correct / len(evaluations), 1),
            "recovery_success_rate_percent": success_rate,
            "recovery_actions_executed": executed,
            "duplicate_deliveries_suppressed": outcomes.get(
                "DUPLICATE_SUPPRESSED", 0
            ),
            "escalations": escalations,
            "unsafe_actions_executed": outcomes.get("UNSAFE_EXECUTED", 0),
        },
        "outcomes": dict(sorted(outcomes.items())),
        "evaluations": [evaluation.to_dict() for evaluation in evaluations],
        "audit_events": audits,
        "metric_definitions": {
            "detection_accuracy_percent": (
                "Fixture labels matched divided by all evaluated signals"
            ),
            "recovery_success_rate_percent": (
                "Executed simulated restarts divided by non-duplicate "
                "RESTART_TASK decisions"
            ),
            "unsafe_actions_executed": (
                "Actions executed after a critical, cooldown, duplicate, or "
                "attempt-limit safety condition"
            ),
        },
    }


def write_reports(report: dict[str, object], directory: Path) -> None:
    directory.mkdir(parents=True, exist_ok=True)
    (directory / "reliability_report.json").write_text(
        json.dumps(report, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )

    audits = report["audit_events"]
    if not isinstance(audits, list):
        raise ValueError("audit_events must be a list")
    jsonl = "\n".join(json.dumps(row, sort_keys=True) for row in audits) + "\n"
    (directory / "audit.jsonl").write_text(jsonl, encoding="utf-8")

    metrics = report["metrics"]
    if not isinstance(metrics, dict):
        raise ValueError("metrics must be an object")
    markdown = "\n".join(
        [
            "# Reliability Report",
            "",
            "All recovery actions in this report are local simulations.",
            "",
            "## Metrics",
            "",
            f"- Signals evaluated: {metrics['signals_evaluated']}",
            f"- Incidents detected: {metrics['incidents_detected']}",
            f"- Controller deliveries: {metrics['controller_deliveries']}",
            f"- Detection accuracy: {metrics['detection_accuracy_percent']}%",
            f"- Recovery success rate: {metrics['recovery_success_rate_percent']}%",
            f"- Recovery actions executed: {metrics['recovery_actions_executed']}",
            (
                "- Duplicate deliveries suppressed: "
                f"{metrics['duplicate_deliveries_suppressed']}"
            ),
            f"- Escalations: {metrics['escalations']}",
            f"- Unsafe actions executed: {metrics['unsafe_actions_executed']}",
            "",
            "## Safety interpretation",
            "",
            "A simulated restart counts as successful execution, not proof that a "
            "real ECS workload recovered. Blocked and escalated decisions are "
            "expected safety outcomes.",
            "",
        ]
    )
    (directory / "reliability_report.md").write_text(
        markdown, encoding="utf-8"
    )

How Does Reporting Work?

• The build_report() function compares detector results with all fixture labels.
• The recovery success rate divides executed simulated restarts by non-duplicate RESTART_TASK decisions.
• The write_reports() function writes three evidence formats from the same report object.

• Save self_healing/reporting.py.
• Confirm the file tree shows reporting.py beside simulation.py.

Is the Reporting File in the Wrong Folder?

Confirm reporting.py sits inside self_healing. A file beside main.py cannot satisfy the package import.

Need help checking the layout? Help me place the reporting module correctly.

The reporting module now owns metric calculation plus evidence serialization. The safe simulation needs to pass its evaluations plus audit rows into that module.

• In self_healing/simulation.py, replace the current import group with this completed import group:

import json
from pathlib import Path

from self_healing.controller import SafeRecoveryController, UnsafeController
from self_healing.detector import HealthDetector
from self_healing.models import Evaluation, HealthSignal
from self_healing.policy import PolicySet, load_policy_set
from self_healing.reporting import build_report, write_reports
from self_healing.routing import EventBridgeStyleRouter, cloudwatch_style_event
from self_healing.store import AuditStore

What Does the New Import Connect?

The simulation can now create one report object through build_report(). It can write the three evidence files through write_reports().

• Save self_healing/simulation.py.
• In run_safe(), find the store.close() line that follows store.all_audits().
• Replace the existing summary code after store.close() with this report plus metric block:

    report = build_report(evaluations, audits)
    write_reports(report, REPORTS_PATH)
    metrics = report["metrics"]
    if not isinstance(metrics, dict):
        raise ValueError("metrics must be an object")

    print(f"Incidents: {metrics['incidents_detected']}")
    print(f"Controller deliveries: {metrics['controller_deliveries']}")
    print(f"Recovery actions executed: {metrics['recovery_actions_executed']}")
    print(
        "Duplicate deliveries suppressed: "
        f"{metrics['duplicate_deliveries_suppressed']}"
    )
    print(f"Escalations: {metrics['escalations']}")
    print(f"Unsafe actions executed: {metrics['unsafe_actions_executed']}")
    print("Reports: reports/reliability_report.json, "
          "reports/reliability_report.md, reports/audit.jsonl")

What Changes in the Safe Run?

• The safe run builds metrics from the same evaluations plus audits that produced its decisions.
• The report writer creates all three evidence files under reports.
• The terminal summary reads from the report's metric object. The saved evidence plus console output stay aligned.

• Save self_healing/simulation.py.

Before you run the safe simulation, which metric should remain at zero when every safeguard holds?

• Generate the SQLite-backed evidence by running this command:

python main.py safe

What Should the Safe Run Show?

• The summary prints Incidents: 8 plus Controller deliveries: 9.
• The summary prints Recovery actions executed: 2 plus Duplicate deliveries suppressed: 1.
• The summary prints Escalations: 3 plus Unsafe actions executed: 0.
• The final line lists reports/reliability_report.json plus reports/reliability_report.md plus reports/audit.jsonl.

Are the Evidence Files Missing?

Confirm you saved both self_healing/reporting.py plus self_healing/simulation.py before running the command. Confirm reporting.py remains inside the package folder.

Still blocked? Help me trace the report generation path.

✔️ Awesome, I've got everything!

Your safe simulation now builds the report after reading the audit rows. It writes all three evidence formats before printing the metric summary.

ⓧ I'd like to double check the full code

import json
from pathlib import Path

from self_healing.controller import SafeRecoveryController, UnsafeController
from self_healing.detector import HealthDetector
from self_healing.models import Evaluation, HealthSignal
from self_healing.policy import PolicySet, load_policy_set
from self_healing.reporting import build_report, write_reports
from self_healing.routing import EventBridgeStyleRouter, cloudwatch_style_event
from self_healing.store import AuditStore


ROOT = Path(__file__).resolve().parents[1]
SIGNALS_PATH = ROOT / "data" / "health_signals.json"
POLICIES_PATH = ROOT / "data" / "policies.json"
UNSAFE_POLICIES_PATH = ROOT / "data" / "unsafe_policies.json"
DATABASE_PATH = ROOT / "data" / "audit.db"
REPORTS_PATH = ROOT / "reports"


def load_signals(path: Path = SIGNALS_PATH) -> list[HealthSignal]:
    raw = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(raw, list):
        raise ValueError("Health fixture must be a list")
    return [HealthSignal.from_dict(item) for item in raw]


def build_evaluations(
    policy_path: Path = POLICIES_PATH,
) -> tuple[PolicySet, list[Evaluation]]:
    policies = load_policy_set(policy_path)
    detector = HealthDetector(
        consecutive_failures=int(policies.detection["consecutive_failures"]),
        latency_ms=float(policies.detection["latency_ms"]),
        error_rate=float(policies.detection["error_rate"]),
    )
    return policies, detector.evaluate_all(load_signals())


def build_events(evaluations: list[Evaluation]) -> list[dict[str, object]]:
    router = EventBridgeStyleRouter()
    return [
        router.route(cloudwatch_style_event(evaluation))
        for evaluation in evaluations
        if evaluation.alarm
    ]


def run_inspect() -> None:
    policies, evaluations = build_evaluations()
    signals = [evaluation.signal for evaluation in evaluations]
    print(f"Signals: {len(signals)}")
    print(f"Services: {len({signal.service for signal in signals})}")
    print(f"Expected alarms: {sum(signal.expected_alarm for signal in signals)}")
    print(f"Configured service policies: {len(policies.services)}")
    print("Policy validation: PASS")


def run_detect() -> None:
    _, evaluations = build_evaluations()
    for evaluation in evaluations:
        status = "ALARM" if evaluation.alarm else "OK"
        reasons = ",".join(evaluation.reasons) if evaluation.reasons else "none"
        print(
            f"[{status}] {evaluation.signal.service} "
            f"{evaluation.signal.event_id} "
            f"consecutive={evaluation.consecutive_failures} reasons={reasons}"
        )
    correct = sum(
        evaluation.alarm == evaluation.signal.expected_alarm
        for evaluation in evaluations
    )
    print(f"Detection accuracy: {100 * correct / len(evaluations):.1f}% "
          f"({correct}/{len(evaluations)})")
    print(f"Alarms: {sum(evaluation.alarm for evaluation in evaluations)}")


def run_unsafe() -> None:
    _, evaluations = build_evaluations()
    events = build_events(evaluations)
    event = next(item for item in events if item["id"] == "pay-2")
    controller = UnsafeController()
    actions = [controller.handle(event) for _ in range(3)]
    REPORTS_PATH.mkdir(parents=True, exist_ok=True)
    evidence = {"event_id": "pay-2", "unsafe_restart_requests": 3, "actions": actions}
    (REPORTS_PATH / "unsafe_failure.json").write_text(
        json.dumps(evidence, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )
    print("Unsafe restart requests: 3 for event pay-2")
    print("Evidence: reports/unsafe_failure.json")


def run_safe() -> None:
    policies, evaluations = build_evaluations()
    events = build_events(evaluations)
    DATABASE_PATH.unlink(missing_ok=True)
    store = AuditStore(DATABASE_PATH)
    controller = SafeRecoveryController(policies, store)

    for event in events:
        controller.handle(event)
        if event["id"] == "pay-2":
            controller.handle(event)

    audits = store.all_audits()
    store.close()
    report = build_report(evaluations, audits)
    write_reports(report, REPORTS_PATH)
    metrics = report["metrics"]
    if not isinstance(metrics, dict):
        raise ValueError("metrics must be an object")

    print(f"Incidents: {metrics['incidents_detected']}")
    print(f"Controller deliveries: {metrics['controller_deliveries']}")
    print(f"Recovery actions executed: {metrics['recovery_actions_executed']}")
    print(
        "Duplicate deliveries suppressed: "
        f"{metrics['duplicate_deliveries_suppressed']}"
    )
    print(f"Escalations: {metrics['escalations']}")
    print(f"Unsafe actions executed: {metrics['unsafe_actions_executed']}")
    print("Reports: reports/reliability_report.json, "
          "reports/reliability_report.md, reports/audit.jsonl")

What Should Match?

Compare the reporting import plus the final half of run_safe() closely. The other simulation commands remain unchanged.

Review the safety evidence

The completed lab has two independent forms of proof. Tests check individual contracts while generated reports summarize one deterministic safe run.

Before you run the final test check, do you expect every safety contract to pass against the unchanged fixtures?

• Run all eight tests from the existing terminal with this command:

python -m pytest -q

What Should the Final Test Check Show?

The terminal ends with 8 passed. This covers transient suppression plus persistent detection plus every safe-controller boundary.

Did One of the Eight Tests Fail?

Start with the first failed test name. Compare that test with the full-file tab before changing detector or controller behavior.

Need help reading the failure? Help me diagnose the first failing pytest check.

Before you regenerate the reports, which result should remain at zero if every controller safeguard holds?

• Regenerate the final reliability evidence with this command:

python main.py safe

What Should the Final Safe Run Show?

• The terminal prints Incidents: 8.
• The terminal prints Controller deliveries: 9.
• The terminal prints Recovery actions executed: 2.
• The terminal prints Duplicate deliveries suppressed: 1.
• The terminal prints Escalations: 3.
• The terminal prints Unsafe actions executed: 0.

• Expand the reports folder in the VS Code file tree.
• Open reports/reliability_report.json.
• Confirm detection_accuracy_percent is 100.0.
• Confirm recovery_success_rate_percent is 40.0.
• Confirm unsafe_actions_executed is 0.
• Open reports/reliability_report.md to review the safety interpretation.
• Open reports/audit.jsonl to review the append-only controller decisions.

The 40.0 percent recovery rate represents two executed restarts across five non-duplicate restart decisions. Cooldown blocks plus escalations remain expected safety outcomes.

Are the Report Metrics Different?

Confirm the safe command completed after both changed Python files were saved. The command recreates data/audit.db before calculating the metrics.

Still seeing different evidence? Help me trace the generated reliability metrics.

That completes the evidence layer. Your local reliability lab now proves its safety contract with restart-safe tests plus reproducible reports.

Secret mission

Stop a Cascading Recovery Storm

One safe incident does not prove that a controller can survive a recovery storm. Replay 24 deliveries across a simulated process restart to prove that persistent claims still suppress duplicates while every safety boundary holds.

Clean Up Your Resources

Clean Up Your Resources

The entire lab runs inside your local Visual Studio Code workspace with no cloud resources or ongoing costs. Choose whether to keep the complete lab, pause your local environment, or remove its generated artifacts.

Resources you used:

• A project-local Python .venv/ containing pytest 9.2.0.
• Generated SQLite databases at data/audit.db and data/storm.db.
• Generated JSON, Markdown, and JSONL evidence under reports/.

Keep everything running

No action needed. Choose this if you want to rerun the simulations or extend the safety controls later.

• Keep .venv/ so the pinned test environment remains ready.
• Retain data/audit.db and data/storm.db so the persisted recovery history remains available for review.
• Retain the generated files under reports/ so you can compare the unsafe baseline with the safe controller and storm benchmark.

Pause - I'll come back to this later

Deactivate the virtual environment to return the current terminal to its normal interpreter. Every source file and generated evidence file stays intact.

• Return to the terminal session where .venv is active.
• Deactivate the project environment by running this command:

deactivate

What does this command do?

This removes the virtual environment settings from the current terminal session. The environment and its installed packages remain stored in .venv/.

• Close the Visual Studio Code workspace.

The virtual environment is now inactive. No simulation keeps running after its command finishes.

Delete - I don't want to use this again

Remove the local runtime environment and generated evidence while preserving the reproducible project source. You can rebuild every deleted artifact from the retained files.

Generated Evidence Will Be Removed

Removing these artifacts clears your saved audit history and generated reports. Your source code, fixtures, policies, tests, settings, dependency pin, and documentation stay in the repository.

• Select the Explorer icon in the left Activity Bar.
• Right-click .venv/ in the Explorer sidebar.
• Select Delete from the menu.
• Delete data/audit.db using the same right-click menu.
• Delete data/storm.db using the same right-click menu.
• In reports/, delete audit.jsonl, reliability_report.json, reliability_report.md, storm_audit.jsonl, storm_report.json, and unsafe_failure.json.
• Review data/ and reports/ in the Explorer sidebar.

You should still see data/health_signals.json, data/policies.json, and data/unsafe_policies.json under data/. The .venv/, both database files, and all six generated report files should be gone.

Nice Work!

Nice Work!

Mission complete! You built a deterministic local reliability lab that turns sustained failures into safe recovery decisions without touching cloud resources.

What you learned:

• Built a deterministic health detector that keeps one-off failures quiet. Sustained breaches become routed alarm events.
• Exposed duplicate remediation by replaying one incident three times. Added persistent duplicate suppression. Added cooldown enforcement. Added bounded attempts. Protected critical workloads through escalation. Rejected unsafe policy overrides through fail-closed validation.
• Verified the safety contract with eight deterministic pytest tests. Generated JSON evidence from SQLite audit history. Produced matching Markdown reports. Produced JSONL audit records.
• Secret Mission: Replayed a 24-delivery recovery storm across one controller process restart. Suppressed 16 duplicate deliveries. Limited recovery execution to two restart actions.

Ready to quiz yourself?