The Thermostat — continuous rightsizing inside a policy envelope, so cost cleanup stops being a quarterly fire drill

mkdir thermostat-lab && cd thermostat-lab
python3 -m venv .venv && source .venv/bin/activate
pip install anthropic pydantic pyyaml
export ANTHROPIC_API_KEY=sk-ant-...   # your key
mkdir usage policy recommendations

ollama pull llama3.1          # general reasoning; qwen2.5-coder for code-heavy swaps

"""Generates 14 days of per-service utilization. Deterministic on purpose."""
import json
from pathlib import Path

DAYS = [f"2026-07-{d:02d}" for d in range(6, 20)]           # 14 days
WOBBLE = [0, 1, -1, 2, 0, 1, -2, 0, 1, 0, -1, 2, 0, 1]      # day-to-day noise


def days(avg_cpu, peak_cpu, avg_mem, peak_mem):
    return [
        {
            "date": d,
            "cpu_avg_pct": avg_cpu + w,
            "cpu_peak_pct": peak_cpu + w,
            "mem_avg_pct": avg_mem + w,
            "mem_peak_pct": peak_mem + w,
        }
        for d, w in zip(DAYS, WOBBLE)
    ]


services = [
    {   # steady and heavily overprovisioned — the easy win
        "name": "catalog-api", "instance_type": "m5.2xlarge", "replicas": 6,
        "tags": [], "hourly_cost_usd_per_replica": 0.384,
        "daily": days(12, 20, 22, 30),
    },
    {   # also overprovisioned
        "name": "image-resizer", "instance_type": "c5.xlarge", "replicas": 4,
        "tags": [], "hourly_cost_usd_per_replica": 0.17,
        "daily": days(8, 13, 18, 24),
    },
    {   # THE TRAP: low average, hard nightly peak (finance batch at 02:00)
        "name": "report-runner", "instance_type": "m5.xlarge", "replicas": 3,
        "tags": [], "hourly_cost_usd_per_replica": 0.192,
        "daily": days(16, 85, 30, 74),
    },
    {   # healthy utilization — nothing to do here
        "name": "session-store", "instance_type": "r5.large", "replicas": 3,
        "tags": [], "hourly_cost_usd_per_replica": 0.126,
        "daily": days(38, 55, 52, 68),
    },
]

Path("usage/metrics.json").write_text(
    json.dumps({"window_days": 14, "unit": "percent_of_allocated",
                "services": services}, indent=2)
)
print("wrote usage/metrics.json")

# The policy envelope. The agent may only act inside this boundary.
# Enforcement lives in thermostat.py's policy check — in code, not in the prompt.
allowed_actions:
  - rightsize_within_family_one_step   # m5.2xlarge -> m5.xlarge OK; m5 -> c6g never
  - scale_replicas_plus_minus_one
forbidden_tags:
  - stateful            # services carrying this tag are untouchable
min_headroom_pct: 40    # after any change, projected PEAK utilization must stay
                        # below (100 - min_headroom_pct) percent
auto_execute: false     # recommend tier: a human merges every change

"""The Thermostat — closed-loop rightsizing inside a policy envelope."""
import json
from pathlib import Path

import anthropic
import yaml
from pydantic import BaseModel

VCPUS = {"large": 2, "xlarge": 4, "2xlarge": 8, "4xlarge": 16}
LADDER = ["large", "xlarge", "2xlarge", "4xlarge"]


class ResourceState(BaseModel):
    instance_type: str
    replicas: int


class Recommendation(BaseModel):
    service: str
    current: ResourceState
    proposed: ResourceState
    projected_saving_usd: float
    risk_note: str
    evidence: str


class RecommendationSet(BaseModel):
    recommendations: list[Recommendation]


SYSTEM = """You are a capacity analyst. Given 14 days of utilization data and a
policy envelope, propose rightsizing actions ONLY where the data supports them.

Rules:
- Peaks matter more than averages. A service with a low average but high daily
  peaks needs its headroom: explain that in a risk_note and propose nothing, or
  explicitly propose nothing for it.
- At most one action per service. Prefer one-step instance rightsizing within
  the same family. Only propose actions the envelope allows.
- evidence must cite actual numbers from the data (avg, peak, window length).
- projected_saving_usd is monthly, derived from the hourly cost figures.
- If nothing should change for a service, omit it entirely.
Your output is a proposal. A separate policy check in code validates every
recommendation — do not assume you are the last line of defense."""


def split_type(itype: str):
    family, _, size = itype.partition(".")
    return family, size


def capacity_factor(cur: ResourceState, prop: ResourceState) -> float:
    _, cs = split_type(cur.instance_type)
    _, ps = split_type(prop.instance_type)
    return (VCPUS[ps] / VCPUS[cs]) * (prop.replicas / cur.replicas)


def policy_check(rec: Recommendation, svc: dict, envelope: dict) -> list[str]:
    """Return a list of violations. Empty list means the recommendation passes."""
    violations = []
    if any(t in envelope.get("forbidden_tags", []) for t in svc["tags"]):
        violations.append(
            f"service is tagged {svc['tags']} — forbidden by envelope")
    cf, cs = split_type(rec.current.instance_type)
    pf, ps = split_type(rec.proposed.instance_type)
    if cf != pf:
        violations.append(f"cross-family move {cf} -> {pf} is forbidden")
    elif cs != ps and abs(LADDER.index(cs) - LADDER.index(ps)) != 1:
        violations.append(f"{cs} -> {ps} is more than one step within the family")
    if abs(rec.proposed.replicas - rec.current.replicas) > 1:
        violations.append("replica change larger than +/-1")
    if (rec.current.instance_type != svc["instance_type"]
            or rec.current.replicas != svc["replicas"]):
        violations.append(
            "'current' does not match the metrics file — hallucinated state")
    observed_peak = max(d["cpu_peak_pct"] for d in svc["daily"])
    projected_peak = observed_peak / capacity_factor(rec.current, rec.proposed)
    limit = 100 - envelope["min_headroom_pct"]
    if projected_peak > limit:
        violations.append(
            f"headroom: observed peak {observed_peak}% -> projected "
            f"{projected_peak:.0f}% exceeds limit {limit}% "
            f"(min_headroom_pct={envelope['min_headroom_pct']})")
    return violations


def write_diff(rec: Recommendation, note: str) -> None:
    diff = f"""--- a/deploy/{rec.service}.yaml
+++ b/deploy/{rec.service}.yaml
@@ resources @@
-  instance_type: {rec.current.instance_type}
-  replicas: {rec.current.replicas}
+  instance_type: {rec.proposed.instance_type}
+  replicas: {rec.proposed.replicas}

# projected saving : ${rec.projected_saving_usd:,.0f}/month (model estimate — verify against billing)
# evidence         : {rec.evidence}
# risk note        : {rec.risk_note}
# policy check     : PASSED — {note}
"""
    Path(f"recommendations/{rec.service}.diff").write_text(diff)


def main() -> None:
    metrics = json.loads(Path("usage/metrics.json").read_text())
    envelope = yaml.safe_load(Path("policy/envelope.yaml").read_text())
    by_name = {s["name"]: s for s in metrics["services"]}

    client = anthropic.Anthropic()
    response = client.messages.parse(
        model="claude-opus-4-8",
        max_tokens=4096,
        thinking={"type": "adaptive"},
        system=SYSTEM,
        messages=[{
            "role": "user",
            "content": (
                "POLICY ENVELOPE:\n" + Path("policy/envelope.yaml").read_text()
                + "\n\nUTILIZATION DATA:\n" + json.dumps(metrics, indent=2)
            ),
        }],
        output_format=RecommendationSet,
    )

    for rec in response.parsed_output.recommendations:
        svc = by_name.get(rec.service)
        if svc is None:
            print(f"REJECT {rec.service}: unknown service (not in metrics)")
            continue
        violations = policy_check(rec, svc, envelope)
        if violations:
            print(f"REJECT {rec.service}: " + "; ".join(violations))
            continue
        peak = max(d["cpu_peak_pct"] for d in svc["daily"])
        projected = peak / capacity_factor(rec.current, rec.proposed)
        note = (f"peak {peak}% -> projected {projected:.0f}% "
                f"<= {100 - envelope['min_headroom_pct']}%")
        write_diff(rec, note)
        print(f"PASS   {rec.service}: {rec.current.instance_type} x{rec.current.replicas}"
              f" -> {rec.proposed.instance_type} x{rec.proposed.replicas}"
              f" (~${rec.projected_saving_usd:,.0f}/mo) — {note}")


if __name__ == "__main__":
    main()

python thermostat.py

python usage/make_metrics.py && python thermostat.py