Build an AI-Assisted Self-Healing System

Wire anomaly detection, runbook search, an LLM and Ansible into a guarded remediation loop with approval tiers, audit logs and verification.

Architecture Overview & Problem Statement

This project builds a control loop that watches one Kubernetes workload, detects an anomaly with a machine learning model, asks an LLM (with runbook search) what to do, checks the proposal against hard safety rules, runs an approved fix with Ansible, and then verifies the result. Low-risk actions run on their own, medium-risk ones run and then notify a human, and high-risk ones wait for approval. Every step is written to a tamper-evident audit log. You finish with measured results, such as time to recovery and how often the guardrails stopped a bad proposal, and a repository you can show in a portfolio.

"Self-healing" here does not mean unsupervised. The system handles a short, human-chosen list of reversible actions on one workload, and you decide what it is allowed to do. Treat any claim that it fixes incidents without human involvement as something to test and measure, not assume.

This project builds on two earlier ones. Build an Anomaly Detection Pipeline gives you the trained model, and Build an AI-Powered Incident Response Agent gives you the runbook search and context collection. Here you add the part that acts. Deploy a Self-Healing Microservices Platform with Istio Service Mesh takes a different approach, where resilience is built into the service mesh. This project is an external control loop that observes and repairs from outside.

What you build:

A detect, diagnose, validate, execute, verify pipeline
A safety layer that bounds every action and never lets the LLM pick names or numbers freely
Five Ansible playbooks and a decision engine with three autonomy tiers
An approval service, a hash-chained audit log, and a dry-run mode and kill switch
Least-privilege Kubernetes permissions, scoped to one namespace

flowchart LR P[Prometheus
metrics] --> D[Detect
Isolation Forest] D --> L[Diagnose
runbooks and LLM] L --> V[Validate
bounds and evidence] V --> T{Decide
tier} T --> E[Execute
Ansible] E --> F[Verify
recovery] F --> A[(Audit log)] classDef src fill:#475569,color:#fff classDef ml fill:#8B5CF6,color:#fff classDef guard fill:#F97316,color:#fff classDef act fill:#326CE5,color:#fff classDef ok fill:#10B981,color:#fff class P src class D,L ml class V,T guard class E act class F,A ok

The three tiers of autonomy

Not every action carries the same risk, so the system sorts them.

flowchart TD X[Proposed action] --> G{Safety checks
pass?} G -->|no| B[Blocked] G -->|yes| C{Tier and
confidence} C -->|Tier 1| T1[Run now] C -->|Tier 2| T2[Run, then notify] C -->|Tier 3 or low| T3[Ask a human] classDef bad fill:#EF4444,color:#fff classDef go fill:#10B981,color:#fff classDef mid fill:#F97316,color:#fff classDef ask fill:#326CE5,color:#fff classDef gate fill:#475569,color:#fff class B bad class T1 go class T2 mid class T3 ask class X,G,C gate

Tier 1, run immediately: restart one pod, scale up. Easy to reverse, and common.
Tier 2, run then notify: raise a memory limit, rolling restart. Higher impact, still reversible.
Tier 3, ask first: roll back a deployment. A wrong call can cause real damage.

Design rules that keep it safe

The LLM chooses an action, never targets. The namespace, deployment, and container come from your configuration. A model that invents a name cannot touch anything else.
Every number is bounded. Replica counts and memory limits are clamped by code, whatever the model says.
Evidence before action. For example, a pod restart is refused when fewer than two replicas are ready, because it could cause the outage it was meant to fix.
Confidence is a weak signal. A model's own confidence score is not calibrated, so it is one input among several and never the only gate.
Dry run by default. The system logs what it would do for a period before it is allowed to act.

Security
Anything that can change production needs a kill switch, an audit trail, and the smallest possible permissions. Build those first, and add actions to the allowed list later, one at a time.

Milestone 1: Set Up the Project and Prerequisites

This milestone creates the project and brings in what you need from the two earlier projects.

You need: a Kubernetes cluster with Prometheus, the trained model file from the detection project (a joblib file holding model, scaler, and feature_columns), and the runbook search and context collection code from the incident project. Install the same scikit-learn and pandas versions you trained with, or the model may not load.

Bash

mkdir aiops-self-healing && cd aiops-self-healing
mkdir -p src ansible/playbooks ansible/inventory audit models k8s tests
touch src/__init__.py
 
# Bring in the model and the two helpers from the earlier projects
cp ../aiops-anomaly-detection/models/isolation_forest.joblib models/
cp ../aiops-incident-agent/src/rag_retriever.py src/
cp ../aiops-incident-agent/src/context_collector.py src/
 
pip3 install ansible-core ansible-runner kubernetes fastapi uvicorn requests joblib pandas scikit-learn pytest
ansible-galaxy collection install kubernetes.core

Copy the two helper files into this project instead of importing them across folders. Both projects have a package called src, and importing one from the other causes name collisions that are painful to debug.

src/adapters.py is the only file that touches those helpers, so if their function names differ in your versions, this is the one place to change:

Python

import logging
 
logger = logging.getLogger("selfheal")
 
try:
    from src.rag_retriever import search_runbooks, format_runbooks_for_prompt
    from src.context_collector import collect_service_context, format_context_for_prompt
    HAVE_AGENT = True
except ImportError:
    HAVE_AGENT = False
    logger.warning("Runbook search and context collection are not available")
 
 
def get_runbook_text(query):
    if not HAVE_AGENT:
        return "No runbooks available."
    return format_runbooks_for_prompt(search_runbooks(query))
 
 
def get_context_text(namespace, deployment):
    if not HAVE_AGENT:
        return "No live context available."
    context = collect_service_context(namespace, deployment, "AnomalyDetected")
    return format_context_for_prompt(context)

Milestone 2: Configure the Autonomy Limits

This milestone defines exactly what the system may do. The configuration is the blast radius: every limit here is enforced by code in later milestones.

src/config.py:

Python

import os
 
PROMETHEUS_URL = os.getenv("PROMETHEUS_URL", "http://localhost:9090")
OLLAMA_URL = os.getenv("OLLAMA_URL", "http://localhost:11434")
LLM_MODEL = os.getenv("LLM_MODEL", "llama3.1")
MODEL_PATH = os.getenv("MODEL_PATH", "models/isolation_forest.joblib")
 
# Scope: this system manages exactly one workload
TARGET_NAMESPACE = os.getenv("TARGET_NAMESPACE", "selfheal-demo")
TARGET_DEPLOYMENT = os.getenv("TARGET_DEPLOYMENT", "demo-api")
TARGET_CONTAINER = os.getenv("TARGET_CONTAINER", "demo-api")
 
# Safety switches
DRY_RUN = os.getenv("DRY_RUN", "true").lower() == "true"
KILL_SWITCH_FILE = os.getenv("KILL_SWITCH_FILE", "/tmp/selfheal.disabled")
 
DETECTION_INTERVAL_SECONDS = 60
ANOMALY_CONSECUTIVE_THRESHOLD = 2
 
# Only actions that have a playbook belong here
AUTONOMOUS_ACTIONS = ["restart_pod", "scale_up_deployment"]
NOTIFY_AFTER_ACTIONS = ["increase_memory_limit", "rolling_restart_deployment"]
APPROVAL_REQUIRED_ACTIONS = ["rollback_deployment"]
ALL_ACTIONS = AUTONOMOUS_ACTIONS + NOTIFY_AFTER_ACTIONS + APPROVAL_REQUIRED_ACTIONS
 
CONFIDENCE_THRESHOLD = 0.75
MAX_ACTIONS_PER_HOUR = 3
MAX_REPLICAS = 10
MAX_MEMORY_LIMIT_MI = 1024
MIN_READY_REPLICAS_FOR_RESTART = 2
 
APPROVAL_TOKEN = os.getenv("APPROVAL_TOKEN", "")
APPROVAL_HOST = os.getenv("APPROVAL_HOST", "127.0.0.1")
APPROVAL_PORT = int(os.getenv("APPROVAL_PORT", "8081"))
APPROVAL_TTL_SECONDS = 600
 
SLACK_WEBHOOK_URL = os.getenv("SLACK_WEBHOOK_URL", "")
VERIFICATION_WAIT_SECONDS = int(os.getenv("VERIFICATION_WAIT_SECONDS", "90"))
 
AUDIT_LOG_PATH = os.getenv("AUDIT_LOG_PATH", "audit/actions.jsonl")
ANSIBLE_PLAYBOOK_DIR = "ansible/playbooks"
ANSIBLE_INVENTORY = "ansible/inventory/hosts.yaml"

Start the autonomous list small, with only restart_pod, and widen it as the system proves itself. Never put an action in the autonomous list if you cannot undo it within a minute.

Common Mistake
Listing actions in the config that have no playbook. The system then proposes something it cannot run, and fails in the middle of an incident. Every action here has a playbook in the next milestone.

Milestone 3: Write the Ansible Playbooks

This milestone writes one playbook per action. Ansible gives you idempotent modules, structured results, and a saved record of each run, which a bare subprocess call to kubectl does not. The playbooks use the Kubernetes modules and take only the variables that the safety layer has already validated.

ansible/inventory/hosts.yaml:

YAML

all:
  hosts:
    localhost:
      ansible_connection: local
      ansible_python_interpreter: "{{ ansible_playbook_python }}"

ansible/playbooks/wait_rollout.yaml is a shared task file that waits until every replica is updated and ready:

YAML

- name: Wait for the rollout to finish
  kubernetes.core.k8s_info:
    api_version: apps/v1
    kind: Deployment
    namespace: "{{ namespace }}"
    name: "{{ deployment_name }}"
  register: dep
  until: >-
    dep.resources[0].status.updatedReplicas | default(0) == dep.resources[0].spec.replicas
    and dep.resources[0].status.readyReplicas | default(0) == dep.resources[0].spec.replicas
  retries: 18
  delay: 10

ansible/playbooks/restart_pod.yaml:

YAML

- name: Restart one pod; the Deployment recreates it
  hosts: localhost
  connection: local
  gather_facts: false
  tasks:
    - name: Delete the pod
      kubernetes.core.k8s:
        state: absent
        api_version: v1
        kind: Pod
        namespace: "{{ namespace }}"
        name: "{{ pod_name }}"
        wait: true
        wait_timeout: 60
 
    - name: Wait until all replicas are ready again
      ansible.builtin.include_tasks: wait_rollout.yaml

ansible/playbooks/scale_up_deployment.yaml:

YAML

- name: Scale a deployment up
  hosts: localhost
  connection: local
  gather_facts: false
  tasks:
    - name: Set the new replica count
      kubernetes.core.k8s_scale:
        api_version: apps/v1
        kind: Deployment
        namespace: "{{ namespace }}"
        name: "{{ deployment_name }}"
        replicas: "{{ new_replicas | int }}"
        wait: true
        wait_timeout: 180

ansible/playbooks/increase_memory_limit.yaml:

YAML

- name: Raise the memory limit of one container
  hosts: localhost
  connection: local
  gather_facts: false
  tasks:
    - name: Patch the limit
      kubernetes.core.k8s:
        state: patched
        api_version: apps/v1
        kind: Deployment
        namespace: "{{ namespace }}"
        name: "{{ deployment_name }}"
        definition:
          spec:
            template:
              spec:
                containers:
                  - name: "{{ container_name }}"
                    resources:
                      limits:
                        memory: "{{ new_memory_limit }}"
 
    - name: Wait for the rollout
      ansible.builtin.include_tasks: wait_rollout.yaml

ansible/playbooks/rolling_restart_deployment.yaml:

YAML

- name: Rolling restart of a deployment
  hosts: localhost
  connection: local
  gather_facts: false
  tasks:
    - name: Change a pod template annotation to trigger a rollout
      kubernetes.core.k8s:
        state: patched
        api_version: apps/v1
        kind: Deployment
        namespace: "{{ namespace }}"
        name: "{{ deployment_name }}"
        definition:
          spec:
            template:
              metadata:
                annotations:
                  kubectl.kubernetes.io/restartedAt: "{{ lookup('pipe', 'date -u +%Y-%m-%dT%H:%M:%SZ') }}"
 
    - name: Wait for the rollout
      ansible.builtin.include_tasks: wait_rollout.yaml

ansible/playbooks/rollback_deployment.yaml:

YAML

- name: Roll a deployment back to its previous revision
  hosts: localhost
  connection: local
  gather_facts: false
  tasks:
    - name: Roll back
      kubernetes.core.k8s_rollback:
        api_version: apps/v1
        kind: Deployment
        namespace: "{{ namespace }}"
        name: "{{ deployment_name }}"
 
    - name: Wait for the rollout
      ansible.builtin.include_tasks: wait_rollout.yaml

Bash

for p in ansible/playbooks/*.yaml; do
  case "$p" in *wait_rollout.yaml) continue ;; esac
  ansible-playbook --syntax-check -i ansible/inventory/hosts.yaml "$p"
done

Common Mistake
Building shell commands from model output, such as kubectl rollout restart deployment/{{ name }}. A model that returns a crafted name can run arbitrary commands. These playbooks use Kubernetes modules, and the names they receive come from configuration, not from the model.

Milestone 4: Build the Safety Layer

This milestone builds the code that stands between a proposal and an action. It turns an LLM suggestion into a set of validated variables, or refuses it with a reason.

src/validators.py:

Python

import json
import math
import re
import subprocess
 
from src import config
 
NAME_RE = re.compile(r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?$")
 
 
def kubectl_json(*args):
    out = subprocess.run(
        ["kubectl", *args, "-o", "json"],
        capture_output=True, text=True, timeout=20,
    )
    if out.returncode != 0:
        raise RuntimeError(out.stderr.strip())
    return json.loads(out.stdout)
 
 
def _to_mi(quantity):
    for unit, factor in (("Ki", 1 / 1024), ("Mi", 1), ("Gi", 1024)):
        if quantity.endswith(unit):
            return float(quantity[:-2]) * factor
    return float(quantity) / (1024 * 1024)  # plain bytes
 
 
def _selector(deployment):
    labels = deployment["spec"]["selector"]["matchLabels"]
    return ",".join(f"{k}={v}" for k, v in labels.items())
 
 
def _pick_pod(namespace, selector):
    pods = kubectl_json("get", "pods", "-n", namespace, "-l", selector)["items"]
    if not pods:
        return None
 
    def sort_key(pod):
        statuses = pod["status"].get("containerStatuses", [])
        ready = bool(statuses) and all(s.get("ready") for s in statuses)
        restarts = sum(s.get("restartCount", 0) for s in statuses)
        return (ready, -restarts)  # unhealthy and most restarted first
 
    return sorted(pods, key=sort_key)[0]["metadata"]["name"]
 
 
def build_action_vars(action, params):
    """Return (variables, None) if the action is safe, or (None, reason) if not."""
    namespace = config.TARGET_NAMESPACE
    name = config.TARGET_DEPLOYMENT
    container = config.TARGET_CONTAINER
 
    for value in (namespace, name, container):
        if not NAME_RE.match(value):
            return None, f"Invalid name in configuration: {value}"
 
    deployment = kubectl_json("get", "deployment", name, "-n", namespace)
    replicas = deployment["spec"].get("replicas", 1)
    ready = deployment["status"].get("readyReplicas", 0)
    base = {"namespace": namespace, "deployment_name": name, "container_name": container}
 
    if action == "restart_pod":
        if ready < config.MIN_READY_REPLICAS_FOR_RESTART:
            return None, (
                f"Only {ready} replica(s) ready. Restarting a pod now could remove "
                f"the remaining capacity, so it is refused."
            )
        pod = _pick_pod(namespace, _selector(deployment))
        if not pod:
            return None, "No pod found for the target deployment"
        return {**base, "pod_name": pod}, None
 
    if action == "scale_up_deployment":
        new_replicas = min(math.ceil(replicas * 1.5), config.MAX_REPLICAS)
        if new_replicas <= replicas:
            return None, f"Already at the replica ceiling ({config.MAX_REPLICAS})"
        return {**base, "new_replicas": new_replicas}, None
 
    if action == "increase_memory_limit":
        spec = next(
            (c for c in deployment["spec"]["template"]["spec"]["containers"]
             if c["name"] == container),
            None,
        )
        limit = (spec or {}).get("resources", {}).get("limits", {}).get("memory")
        if not limit:
            return None, "The container has no memory limit to raise"
        current = _to_mi(limit)
        wanted = params.get("new_memory_limit_mi")
        new_limit = int(wanted) if isinstance(wanted, (int, float)) else math.ceil(current * 1.5)
        new_limit = min(max(new_limit, math.ceil(current * 1.1)), config.MAX_MEMORY_LIMIT_MI)
        if new_limit <= current:
            return None, f"Already at the memory ceiling ({config.MAX_MEMORY_LIMIT_MI}Mi)"
        return {**base, "new_memory_limit": f"{new_limit}Mi"}, None
 
    if action == "rolling_restart_deployment":
        return base, None
 
    if action == "rollback_deployment":
        annotations = deployment["metadata"].get("annotations", {})
        revision = int(annotations.get("deployment.kubernetes.io/revision", "1"))
        if revision < 2:
            return None, "There is no earlier revision to roll back to"
        return base, None
 
    return None, f"Unknown action: {action}"

src/audit_logger.py writes each event with a hash of the previous one. If anyone edits or deletes an entry later, the chain no longer verifies, which makes the log tamper-evident. A plain file of JSON lines is not.

Python

import hashlib
import json
import logging
import os
from datetime import datetime, timedelta, timezone
 
from src import config
 
logger = logging.getLogger("selfheal")
 
 
def _last_hash():
    if not os.path.exists(config.AUDIT_LOG_PATH):
        return "0" * 64
    last = None
    with open(config.AUDIT_LOG_PATH) as f:
        for line in f:
            if line.strip():
                last = line
    return json.loads(last)["hash"] if last else "0" * 64
 
 
def _digest(prev_hash, entry):
    body = json.dumps(entry, sort_keys=True)
    return hashlib.sha256((prev_hash + body).encode()).hexdigest()
 
 
def log_event(event_type, service, action, details, outcome=None):
    os.makedirs(os.path.dirname(config.AUDIT_LOG_PATH) or ".", exist_ok=True)
    prev_hash = _last_hash()
    entry = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "event_type": event_type,
        "service": service,
        "action": action,
        "outcome": outcome,
        "details": details,
        "prev_hash": prev_hash,
    }
    entry["hash"] = _digest(prev_hash, {k: v for k, v in entry.items() if k != "hash"})
    with open(config.AUDIT_LOG_PATH, "a") as f:
        f.write(json.dumps(entry) + "\n")
    logger.info("audit: %s | %s | %s | %s", event_type, service, action or "-", outcome or "-")
 
 
def verify_chain():
    prev_hash = "0" * 64
    if not os.path.exists(config.AUDIT_LOG_PATH):
        return True
    with open(config.AUDIT_LOG_PATH) as f:
        for line in f:
            if not line.strip():
                continue
            entry = json.loads(line)
            stored = entry.pop("hash")
            if entry["prev_hash"] != prev_hash or _digest(prev_hash, entry) != stored:
                return False
            prev_hash = stored
    return True
 
 
def recent_events(service=None, hours=24):
    cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)
    events = []
    if not os.path.exists(config.AUDIT_LOG_PATH):
        return events
    with open(config.AUDIT_LOG_PATH) as f:
        for line in f:
            if not line.strip():
                continue
            event = json.loads(line)
            if datetime.fromisoformat(event["timestamp"]) < cutoff:
                continue
            if service and event["service"] != service:
                continue
            events.append(event)
    return events

src/decision_engine.py decides how much freedom a proposal gets. It counts recent executions from the audit log, so the rate limit survives a restart.

Python

from src import config
from src.audit_logger import recent_events
 
 
def decide(action, confidence, service):
    if action not in config.ALL_ACTIONS:
        return {"decision": "blocked", "can_execute": False,
                "reason": f"'{action}' is not on the allowed action list"}
 
    executed = [
        e for e in recent_events(service, hours=1)
        if e["event_type"] == "action_executed"
    ]
    if len(executed) >= config.MAX_ACTIONS_PER_HOUR:
        return {"decision": "blocked", "can_execute": False,
                "reason": (f"{len(executed)} actions already ran in the last hour "
                           f"(limit {config.MAX_ACTIONS_PER_HOUR}). This looks like a "
                           f"loop, so a human should investigate.")}
 
    if action in config.APPROVAL_REQUIRED_ACTIONS:
        return {"decision": "approval_required", "can_execute": False,
                "reason": f"'{action}' is a Tier 3 action and needs approval"}
 
    if confidence < config.CONFIDENCE_THRESHOLD:
        return {"decision": "approval_required", "can_execute": False,
                "reason": f"Confidence {confidence:.0%} is below {config.CONFIDENCE_THRESHOLD:.0%}"}
 
    if action in config.AUTONOMOUS_ACTIONS:
        return {"decision": "autonomous", "can_execute": True,
                "reason": f"'{action}' is a Tier 1 action"}
 
    return {"decision": "notify_after", "can_execute": True,
            "reason": f"'{action}' is a Tier 2 action; humans are told afterwards"}

Common Mistake
Trusting the LLM's "confidence 0.95". Language models are not calibrated, so that number can be high when the diagnosis is wrong. This is why the hard bounds and the evidence checks above are the real protection, and confidence is only one input.

Milestone 5: Build the Executor and the Verifier

This milestone runs the playbook and then checks whether the fix actually worked. An action that ran without errors has not necessarily solved anything.

src/ansible_executor.py:

Python

import os
from datetime import datetime, timezone
 
import ansible_runner
 
from src import config
 
ACTION_PLAYBOOKS = {
    "restart_pod": "restart_pod.yaml",
    "scale_up_deployment": "scale_up_deployment.yaml",
    "increase_memory_limit": "increase_memory_limit.yaml",
    "rolling_restart_deployment": "rolling_restart_deployment.yaml",
    "rollback_deployment": "rollback_deployment.yaml",
}
 
 
def run_action(action, action_vars):
    playbook = ACTION_PLAYBOOKS.get(action)
    if not playbook:
        return {"success": False, "error": f"No playbook for {action}"}
 
    if config.DRY_RUN:
        return {"success": True, "dry_run": True,
                "output": f"DRY RUN: would run {playbook} with {action_vars}"}
 
    run_dir = os.path.abspath(os.path.join(
        "ansible", "artifacts", datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S")))
    os.makedirs(run_dir, exist_ok=True)
 
    result = ansible_runner.run(
        private_data_dir=run_dir,
        playbook=os.path.abspath(os.path.join(config.ANSIBLE_PLAYBOOK_DIR, playbook)),
        inventory=os.path.abspath(config.ANSIBLE_INVENTORY),
        extravars=action_vars,
        quiet=True,
    )
 
    messages = []
    for event in result.events:
        res = event.get("event_data", {}).get("res", {})
        if event.get("event") == "runner_on_failed" and isinstance(res.get("msg"), str):
            messages.append(res["msg"])
 
    return {
        "success": result.status == "successful",
        "status": result.status,
        "rc": result.rc,
        "artifacts": run_dir,
        "output": "\n".join(messages),
    }

src/verifier.py waits, checks that the deployment is fully ready, and runs the detection model again. A fix counts as successful only if the workload is healthy and the anomaly has cleared. A deployment with zero pods must not count as healthy.

Python

import asyncio
import logging
 
from src import config
from src.validators import kubectl_json
 
logger = logging.getLogger("selfheal")
 
 
async def verify_remediation(detect_fn):
    logger.info("Waiting %ss before verifying", config.VERIFICATION_WAIT_SECONDS)
    # asyncio.sleep, not time.sleep, so the approval service keeps responding
    await asyncio.sleep(config.VERIFICATION_WAIT_SECONDS)
 
    deployment = kubectl_json(
        "get", "deployment", config.TARGET_DEPLOYMENT, "-n", config.TARGET_NAMESPACE)
    desired = deployment["spec"].get("replicas", 0)
    ready = deployment["status"].get("readyReplicas", 0)
    deployment_ok = desired > 0 and ready == desired
 
    still_anomalous = await asyncio.to_thread(detect_fn)
    resolved = deployment_ok and not still_anomalous
 
    return {
        "resolved": resolved,
        "ready_replicas": ready,
        "desired_replicas": desired,
        "still_anomalous": still_anomalous,
    }

Milestone 6: Build the Approval Service

This milestone lets a human approve or reject a Tier 3 action. The service issues a random, single-use request ID, expires it after ten minutes, and requires a secret token. An approval endpoint with no authentication would let anyone approve a rollback.

src/approvals.py:

Python

import asyncio
import logging
import secrets
import time
 
import requests
import uvicorn
from fastapi import FastAPI, Header, HTTPException
 
from src import config
 
logger = logging.getLogger("selfheal")
app = FastAPI()
_pending = {}
 
 
def notify(message):
    if not config.SLACK_WEBHOOK_URL:
        logger.info("[slack not configured] %s", message)
        return
    try:
        requests.post(config.SLACK_WEBHOOK_URL, json={"text": message}, timeout=10)
    except requests.RequestException as exc:
        logger.error("Slack notification failed: %s", exc)
 
 
@app.post("/approvals/{request_id}")
async def respond(request_id: str, approved: bool, x_approval_token: str = Header(default="")):
    if not config.APPROVAL_TOKEN or not secrets.compare_digest(
            x_approval_token, config.APPROVAL_TOKEN):
        raise HTTPException(status_code=401, detail="Invalid token")
    item = _pending.pop(request_id, None)  # single use
    if not item or item["expires"] < time.time():
        raise HTTPException(status_code=404, detail="Unknown or expired request")
    item["future"].set_result(approved)
    return {"recorded": approved}
 
 
async def request_approval(action, service, summary):
    request_id = secrets.token_urlsafe(16)
    future = asyncio.get_running_loop().create_future()
    _pending[request_id] = {"future": future, "expires": time.time() + config.APPROVAL_TTL_SECONDS}
 
    notify(
        f"Approval needed for `{action}` on `{service}`.\n{summary}\n"
        f"Request ID: {request_id} (expires in {config.APPROVAL_TTL_SECONDS // 60} minutes)"
    )
    try:
        return await asyncio.wait_for(future, config.APPROVAL_TTL_SECONDS)
    except asyncio.TimeoutError:
        _pending.pop(request_id, None)
        return False  # no answer means no action
 
 
async def serve_approvals():
    server = uvicorn.Server(uvicorn.Config(
        app, host=config.APPROVAL_HOST, port=config.APPROVAL_PORT, log_level="warning"))
    await server.serve()

A person approves with the request ID from the message:

Bash

curl -X POST "http://127.0.0.1:8081/approvals/REQUEST_ID?approved=true" \
  -H "X-Approval-Token: $APPROVAL_TOKEN"

This is a simple design for a lab. In production, use Slack interactive buttons, verify Slack's request signature, and never expose this endpoint to the internet.

Milestone 7: Build the Orchestrator

This milestone connects everything into the main loop. It detects, and when an anomaly persists, it diagnoses, validates, decides, acts, and verifies. It also supports a --simulate mode that skips detection and the live LLM, so you can test the pipeline repeatably.

src/orchestrator.py:

Python

import argparse
import asyncio
import json
import logging
import math
import os
 
import joblib
import pandas as pd
import requests
 
from src import config
from src.adapters import get_context_text, get_runbook_text
from src.ansible_executor import run_action
from src.approvals import notify, request_approval, serve_approvals
from src.audit_logger import log_event
from src.decision_engine import decide
from src.validators import build_action_vars
from src.verifier import verify_remediation
 
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("selfheal")
SERVICE = config.TARGET_DEPLOYMENT
 
 
def metric_queries():
    ns, dep = config.TARGET_NAMESPACE, config.TARGET_DEPLOYMENT
    pods = f'namespace="{ns}",pod=~"{dep}-.*"'
    return {
        "cpu_usage": f'sum(rate(container_cpu_usage_seconds_total{{{pods},container!=""}}[5m]))',
        "memory_usage": f'sum(container_memory_working_set_bytes{{{pods},container!=""}})',
        "http_request_rate": f'sum(rate(http_requests_total{{namespace="{ns}"}}[5m]))',
        "http_error_rate": (
            f'sum(rate(http_requests_total{{namespace="{ns}",status=~"5.."}}[5m]))'
            f' / sum(rate(http_requests_total{{namespace="{ns}"}}[5m]))'
        ),
    }
 
 
def load_model():
    package = joblib.load(config.MODEL_PATH)
    return package["model"], package["scaler"], package["feature_columns"]
 
 
def collect_metrics(features):
    queries = metric_queries()
    metrics = {}
    for name in features:
        value = 0.0
        if name in queries:
            try:
                r = requests.get(f"{config.PROMETHEUS_URL}/api/v1/query",
                                 params={"query": queries[name]}, timeout=10).json()
                if r["status"] == "success" and r["data"]["result"]:
                    value = float(r["data"]["result"][0]["value"][1])
            except (requests.RequestException, ValueError, KeyError):
                pass
        else:
            logger.warning("No query for feature '%s'; using 0", name)
        metrics[name] = 0.0 if math.isnan(value) else value
    return metrics
 
 
def detect(model, scaler, features):
    metrics = collect_metrics(features)
    row = pd.DataFrame([metrics])[features]
    scaled = scaler.transform(row)
    score = float(model.score_samples(scaled)[0])
    return model.predict(scaled)[0] == -1, score, metrics
 
 
def call_llm(context_text, runbook_text, metrics):
    prompt = f"""You are an SRE. An anomaly was detected on one workload.
 
LIVE STATE:
{context_text}
 
RUNBOOKS:
{runbook_text}
 
METRICS: {json.dumps(metrics)}
 
Choose ONE action from: {", ".join(config.ALL_ACTIONS)}.
Reply with JSON only, using these keys:
diagnosis (two sentences), action, confidence (0 to 1), reasoning,
params (optional, may include new_memory_limit_mi as a number).
Do not include namespaces, deployment names or pod names."""
    try:
        r = requests.post(
            f"{config.OLLAMA_URL}/api/generate",
            json={"model": config.LLM_MODEL, "prompt": prompt, "stream": False,
                  "format": "json", "options": {"temperature": 0.05}},
            timeout=120,
        )
        proposal = json.loads(r.json()["response"])
        if proposal.get("action") in config.ALL_ACTIONS:
            return proposal
        logger.error("LLM proposed an unknown action: %s", proposal.get("action"))
    except (requests.RequestException, ValueError, KeyError) as exc:
        logger.error("LLM call failed: %s", exc)
    return None
 
 
def make_stub_llm(action):
    def stub(_context, _runbooks, _metrics):
        return {"diagnosis": "Simulated diagnosis", "action": action,
                "confidence": 0.9, "reasoning": "stub", "params": {}}
    return stub
 
 
async def handle_anomaly(metrics, score, detect_fn, llm_fn):
    logger.info("Anomaly confirmed (score %.4f). Starting the pipeline.", score)
    log_event("anomaly_detected", SERVICE, None, {"score": score, "metrics": metrics})
 
    context_text = get_context_text(config.TARGET_NAMESPACE, config.TARGET_DEPLOYMENT)
    runbook_text = get_runbook_text(f"anomaly on {SERVICE}: {json.dumps(metrics)}")
    proposal = await asyncio.to_thread(llm_fn, context_text, runbook_text, metrics)
    if not proposal:
        log_event("diagnosis_complete", SERVICE, None, {"error": "no valid proposal"}, "failure")
        return
 
    action = proposal["action"]
    confidence = float(proposal.get("confidence", 0))
    log_event("diagnosis_complete", SERVICE, action, {
        "diagnosis": proposal.get("diagnosis"), "confidence": confidence,
        "reasoning": proposal.get("reasoning"),
    })
 
    # Safety layer: bounds and evidence. The model never supplies targets.
    action_vars, problem = await asyncio.to_thread(
        build_action_vars, action, proposal.get("params", {}) or {})
    if problem:
        log_event("action_blocked", SERVICE, action, {"reason": problem}, "blocked")
        notify(f"Action `{action}` on `{SERVICE}` was refused by the safety layer: {problem}")
        return
 
    decision = decide(action, confidence, SERVICE)
    log_event("action_decided", SERVICE, action, decision)
    logger.info("Decision: %s (%s)", decision["decision"], decision["reason"])
 
    if decision["decision"] == "blocked":
        notify(f"Action `{action}` on `{SERVICE}` was blocked: {decision['reason']}")
        return
 
    if not decision["can_execute"]:
        approved = await request_approval(
            action, SERVICE, f"{proposal.get('diagnosis')}\nWhy approval: {decision['reason']}")
        log_event("approval_result", SERVICE, action, {"approved": approved})
        if not approved:
            return
 
    result = await asyncio.to_thread(run_action, action, action_vars)
    if result.get("dry_run"):
        log_event("action_dry_run", SERVICE, action, result, "success")
        notify(f"DRY RUN: would run `{action}` on `{SERVICE}` with {action_vars}")
        return
 
    log_event("action_executed", SERVICE, action, result,
              "success" if result["success"] else "failure")
    if decision["decision"] == "notify_after":
        notify(f"Ran `{action}` on `{SERVICE}` (Tier 2). Result: {result['success']}")
    if not result["success"]:
        notify(f"`{action}` FAILED on `{SERVICE}`. A human should investigate.")
        return
 
    verdict = await verify_remediation(detect_fn)
    log_event("verification_passed" if verdict["resolved"] else "verification_failed",
              SERVICE, action, verdict)
    notify(("Resolved: " if verdict["resolved"] else "NOT resolved: ")
           + f"`{action}` on `{SERVICE}`. {verdict}")
 
 
async def main(simulate, simulated_action):
    approvals = asyncio.create_task(serve_approvals())
 
    if simulate:
        stub = make_stub_llm(simulated_action)
        await handle_anomaly({"memory_usage": 1.5e8}, -0.4, lambda: False, stub)
        approvals.cancel()
        return
 
    model, scaler, features = load_model()
    detect_fn = lambda: detect(model, scaler, features)[0]
    consecutive = 0
    logger.info("Started. dry_run=%s interval=%ss", config.DRY_RUN, config.DETECTION_INTERVAL_SECONDS)
 
    while True:
        if os.path.exists(config.KILL_SWITCH_FILE):
            logger.warning("Kill switch present. Not acting.")
        else:
            is_anomaly, score, metrics = await asyncio.to_thread(detect, model, scaler, features)
            logger.info("score %.4f %s", score, "ANOMALY" if is_anomaly else "normal")
            consecutive = consecutive + 1 if is_anomaly else 0
            if consecutive >= config.ANOMALY_CONSECUTIVE_THRESHOLD:
                await handle_anomaly(metrics, score, detect_fn, call_llm)
                consecutive = 0
        await asyncio.sleep(config.DETECTION_INTERVAL_SECONDS)
 
 
if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--simulate", action="store_true",
                        help="Skip detection and the live LLM; run once with a canned proposal")
    parser.add_argument("--action", default="increase_memory_limit")
    args = parser.parse_args()
    asyncio.run(main(args.simulate, args.action))

Common Mistake
Running two copies of the orchestrator. Both detect the same anomaly and both act, so you get double scale-ups and duplicate messages. Run exactly one, and treat the single-writer rule as part of the design.

Milestone 8: Deploy with Least-Privilege Permissions

This milestone gives the system the smallest set of Kubernetes permissions it needs: a Role in the one namespace it manages. A cluster-wide role would let a bug, or a hijacked model output, touch every workload in the cluster.

First, create the demo workload. It runs three pods that need more memory than their limit allows, so they are killed for running out of memory, which gives the system a real fault to fix.

k8s/demo.yaml:

YAML

apiVersion: v1
kind: Namespace
metadata:
  name: selfheal-demo
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: demo-api
  namespace: selfheal-demo
spec:
  replicas: 3
  selector:
    matchLabels:
      app: demo-api
  template:
    metadata:
      labels:
        app: demo-api
    spec:
      containers:
        - name: demo-api
          image: polinux/stress
          command: ["stress"]
          args: ["--vm", "1", "--vm-bytes", "120M", "--vm-hang", "0"]
          resources:
            requests:
              memory: 50Mi
            limits:
              memory: 100Mi

k8s/rbac.yaml:

YAML

apiVersion: v1
kind: ServiceAccount
metadata:
  name: aiops-self-healing
  namespace: monitoring
---
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: aiops-self-healing
  namespace: selfheal-demo
rules:
  - apiGroups: ["apps"]
    resources: ["deployments", "replicasets"]
    verbs: ["get", "list", "patch", "update"]
  - apiGroups: ["apps"]
    resources: ["deployments/scale"]
    verbs: ["get", "patch", "update"]
  - apiGroups: [""]
    resources: ["pods", "events"]
    verbs: ["get", "list"]
  - apiGroups: [""]
    resources: ["pods"]
    verbs: ["delete"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: aiops-self-healing
  namespace: selfheal-demo
subjects:
  - kind: ServiceAccount
    name: aiops-self-healing
    namespace: monitoring
roleRef:
  kind: Role
  name: aiops-self-healing
  apiGroup: rbac.authorization.k8s.io

Bash

kubectl apply -f k8s/demo.yaml -f k8s/rbac.yaml
 
kubectl auth can-i delete pods -n selfheal-demo \
  --as=system:serviceaccount:monitoring:aiops-self-healing
# Expected: yes
 
kubectl auth can-i delete pods -n kube-system \
  --as=system:serviceaccount:monitoring:aiops-self-healing
# Expected: no

Run it locally first. When you move it into the cluster, build an image with Python, kubectl, your requirements, the kubernetes.core collection, src/, ansible/ and models/, then run it as a single replica with a Recreate strategy. Mount a volume for audit/ so the log survives restarts, and store the Slack webhook and approval token as Secrets.

Security
The approval service should be reachable only inside the cluster or through a port-forward. Anyone who can reach it and holds the token can approve a rollback.

Milestone 9: Run the Experiments and Measure

This milestone proves each guardrail works, and records the numbers. Run them in order, starting with the safest.

Experiment 1: unit checks for the safety rules. Create tests/test_safety.py and run pytest:

Python

import os
import tempfile
 
os.environ["AUDIT_LOG_PATH"] = os.path.join(tempfile.mkdtemp(), "audit.jsonl")
 
from src import config
from src.audit_logger import log_event, verify_chain
from src.decision_engine import decide
 
 
def test_tiers():
    assert decide("restart_pod", 0.9, "svc")["decision"] == "autonomous"
    assert decide("increase_memory_limit", 0.9, "svc")["decision"] == "notify_after"
    assert decide("rollback_deployment", 0.99, "svc")["decision"] == "approval_required"
    assert decide("restart_pod", 0.5, "svc")["decision"] == "approval_required"
    assert decide("delete_everything", 0.99, "svc")["decision"] == "blocked"
 
 
def test_rate_limit():
    for _ in range(config.MAX_ACTIONS_PER_HOUR):
        log_event("action_executed", "svc2", "restart_pod", {}, "success")
    assert decide("restart_pod", 0.95, "svc2")["decision"] == "blocked"
 
 
def test_audit_chain_detects_tampering():
    assert verify_chain() is True
    with open(config.AUDIT_LOG_PATH) as f:
        text = f.read()
    with open(config.AUDIT_LOG_PATH, "w") as f:
        f.write(text.replace("success", "failure", 1))
    assert verify_chain() is False

Experiment 2: a dry run. The system proposes and logs, and changes nothing. Start the demo workload, then:

Bash

export APPROVAL_TOKEN=$(openssl rand -hex 16)
DRY_RUN=true python3 -m src.orchestrator --simulate
grep -E 'dry_run|action_decided' audit/actions.jsonl

Experiment 3: a real fix. Allow actions and run the simulated proposal against the failing demo pods. The pods are killed for running out of memory, so the fix is a higher limit.

Bash

kubectl get pods -n selfheal-demo   # expect OOMKilled or CrashLoopBackOff
DRY_RUN=false python3 -m src.orchestrator --simulate --action increase_memory_limit
kubectl get pods -n selfheal-demo   # expect all Running
kubectl get deployment demo-api -n selfheal-demo \
  -o jsonpath='{.spec.template.spec.containers[0].resources.limits.memory}'

Experiment 4: the safety layer refuses a bad action. With the pods crashing, no replicas are ready, so a restart must be refused:

Bash

kubectl apply -f k8s/demo.yaml     # reset the demo to the failing state
DRY_RUN=false python3 -m src.orchestrator --simulate --action restart_pod
grep action_blocked audit/actions.jsonl | tail -1

Experiment 5: the rate limit. Run Experiment 3 four times in a row. The fourth must be blocked with the loop message.

Experiment 6: approval. Create a second revision, then propose a rollback. The request waits for you:

Bash

kubectl set env deployment/demo-api -n selfheal-demo CHANGE=1
DRY_RUN=false python3 -m src.orchestrator --simulate --action rollback_deployment
# In another terminal, approve with the request ID shown in the log
curl -X POST "http://127.0.0.1:8081/approvals/REQUEST_ID?approved=true" \
  -H "X-Approval-Token: $APPROVAL_TOKEN"

Also try rejecting it, and letting it expire, and confirm no change was made.

Experiment 7, optional: the live loop. Run without --simulate, with DRY_RUN=true first. If your detection model was trained on features that match the queries in metric_queries, it should flag the failing demo and the LLM should propose an action. Run in dry-run mode for a while, and read what it would have done before you ever enable real actions.

Then calculate your own results from the audit log:

Python

import json
from datetime import datetime
 
events = [json.loads(line) for line in open("audit/actions.jsonl")]
start = None
for e in events:
    t = datetime.fromisoformat(e["timestamp"])
    if e["event_type"] == "anomaly_detected":
        start = t
    elif e["event_type"] in ("verification_passed", "verification_failed") and start:
        print(e["event_type"], f"{(t - start).total_seconds():.0f}s after detection")
        start = None

Time yourself fixing the same fault by hand, from noticing it to confirming recovery, and record that as your baseline. Fill in a results table in your README:

Measure	Target	Measured
Detection to verified recovery, autonomous fix	Lower than your manual baseline
Manual baseline for the same fault	Recorded honestly
Proposals blocked by the safety layer	Recorded, with reasons
Actions that needed human approval	Recorded
Verification failures (fix ran, problem stayed)	Recorded and explained
Anomalies flagged when nothing was wrong	Recorded

Report the cases where the system was wrong or refused to act as clearly as the ones where it worked. That honesty is what makes the result believable.

Validation & Testing

Confirm each safety property before you call the project finished.

Bash

# 1. Playbooks are valid
for p in ansible/playbooks/*.yaml; do
  case "$p" in *wait_rollout.yaml) continue ;; esac
  ansible-playbook --syntax-check -i ansible/inventory/hosts.yaml "$p"
done
 
# 2. Safety rules pass
python3 -m pytest tests -q
 
# 3. The audit log is intact
python3 -c "from src.audit_logger import verify_chain; print('chain intact:', verify_chain())"
 
# 4. Permissions are scoped to one namespace
kubectl auth can-i delete pods -n kube-system \
  --as=system:serviceaccount:monitoring:aiops-self-healing
# Expected: no
 
# 5. The kill switch works
touch /tmp/selfheal.disabled
# The live loop should log that it is not acting. Remove the file to resume.
rm /tmp/selfheal.disabled

The project is complete when:

The system runs in dry-run mode by default and acts only when you turn that off
A real fault on the demo workload is fixed by a Tier 2 action, and verification confirms it
The safety layer refuses a restart when too few replicas are ready
The rate limit stops a repeating action, and the approval flow works for approved, rejected, and expired requests
The audit log verifies, and tampering with it is detected
Your results table contains numbers you measured yourself

In your portfolio, include the two diagrams, the audit log excerpt for one full incident, and the results table. Add a short section on what would have to be true before you let this act on a real production service, such as weeks in dry-run, a reviewed action list, and an owner on call.

Clean up when you finish:

Bash

kubectl delete -f k8s/demo.yaml
kubectl delete -f k8s/rbac.yaml

Found a mistake, an outdated step or a broken link?

Build an AI-Assisted Self-Healing System

Blueprint Walkthrough

Architecture Overview & Problem Statement

The three tiers of autonomy

Design rules that keep it safe

Milestone 1: Set Up the Project and Prerequisites

Milestone 2: Configure the Autonomy Limits

Milestone 3: Write the Ansible Playbooks

Milestone 4: Build the Safety Layer

Milestone 5: Build the Executor and the Verifier

Milestone 6: Build the Approval Service

Milestone 7: Build the Orchestrator

Milestone 8: Deploy with Least-Privilege Permissions

Milestone 9: Run the Experiments and Measure

Validation & Testing