Metrics World and Read Tools

Simulate services, SLOs, logs, and read-only tools that never mutate world state.

The environment is a mutable dict the mutate tools will change later. Read tools must not change it. That split is a test: call get_metrics a thousand times, hashes of WORLD stay equal. This part builds the world, SLOs, log lines, and read APIs.

World schema

Each service:

error_rate float 0–1
p95_ms int
cpu float 0–1
deploy_id, prev_deploy_id
replicas int
slo_error, slo_p95_ms
logs list of strings (ring buffer)

A healthy checkout sits at 0.4% errors, p95 220ms. An incident snapshot starts unhealthy so the agent has work. You can also store SCENARIO = "bad_deploy" to pick log lines that mention NullPointer after deploy d44 — diagnosis should point at rollback, not scale.

Read tools

get_metrics(service) → copy of numeric fields (not necessarily all logs).
get_logs(service, n=20) → last n lines.
list_services() → names.
get_slo(service) → slo fields.

Return {"error": "unknown_service"} for typos. Copies via dict so the agent cannot mutate WORLD by accident through a returned object (in Python the numbers are immutable; lists of logs need list(...)).

Live Pythonpython

import copy
import json

def make_world(incident=True):
    logs_ok = ["ok GET /pay 200 80ms", "ok GET /pay 200 90ms"]
    logs_bad = [
        "deploy d44 started",
        "error NullPointer in charge.py last_good=d43",
        "error payment 500 12%",
        "warn p95 1800ms",
    ]
    return {
        "checkout": {
            "error_rate": 0.12 if incident else 0.004,
            "p95_ms": 1800 if incident else 220,
            "cpu": 0.81 if incident else 0.22,
            "deploy_id": "d44",
            "prev_deploy_id": "d43",
            "replicas": 3,
            "slo_error": 0.01,
            "slo_p95_ms": 400,
            "logs": logs_bad if incident else logs_ok,
        },
        "search": {
            "error_rate": 0.002,
            "p95_ms": 90,
            "cpu": 0.18,
            "deploy_id": "s9",
            "prev_deploy_id": "s8",
            "replicas": 2,
            "slo_error": 0.02,
            "slo_p95_ms": 200,
            "logs": ["ok GET /q 200"],
        },
    }

WORLD = make_world(True)

def get_metrics(service):
    row = WORLD.get(service)
    if not row:
        return {"error": "unknown_service", "service": service}
    return {
        k: row[k]
        for k in ("error_rate", "p95_ms", "cpu", "deploy_id", "prev_deploy_id", "replicas")
    }

def get_logs(service, n=20):
    row = WORLD.get(service)
    if not row:
        return {"error": "unknown_service"}
    n = int(n)
    n = max(1, min(n, 50))
    return {"lines": list(row["logs"][-n:])}

def get_slo(service):
    row = WORLD.get(service)
    if not row:
        return {"error": "unknown_service"}
    return {"slo_error": row["slo_error"], "slo_p95_ms": row["slo_p95_ms"]}

def list_services():
    return {"services": sorted(WORLD)}

def burning(service):
    m, s = get_metrics(service), get_slo(service)
    if "error" in m:
        return m
    return {
        "error_burn": m["error_rate"] > s["slo_error"],
        "latency_burn": m["p95_ms"] > s["slo_p95_ms"],
    }

READ = {
    "get_metrics": lambda **kw: get_metrics(kw["service"]),
    "get_logs": lambda **kw: get_logs(kw["service"], kw.get("n", 20)),
    "get_slo": lambda **kw: get_slo(kw["service"]),
    "list_services": lambda **kw: list_services(),
}

snap = copy.deepcopy(WORLD)
print(json.dumps(READ["list_services"](), indent=2))
print(json.dumps(READ["get_metrics"](service="checkout"), indent=2))
print(json.dumps(READ["get_logs"](service="checkout", n=3), indent=2))
print("burn", burning("checkout"))
print("read tools mutated world?", WORLD != snap)
print("search healthy burn", burning("search"))

Output

Run to execute this in your browser. Nothing is sent to a server.

Step-by-step environment rules

Deep-copy in tests when you need a snapshot; for the inequality check above, copy.deepcopy before reads.
Cap log n. Models will request 10 million lines.
Do not put secrets in logs. Fake logs have no API keys. Add a redaction function if a line matches sk-.
Multiple services so diagnosis cannot be "always checkout".

SLO burn as a boolean

A tiny helper burning(service) is not a mutate tool. It is allowed in supervisor code. You can also expose it as check_slo(service) read tool so the model does not do inequality wrong. Prefer tools that compute the predicate for ops — arithmetic in LLMs is sloppy.

Note:Metrics are the observations. Diagnosis is a policy. Keep them separate so you can test burn math without the model.

Cardinality and time

Real metrics systems have labels (region, version, customer). Your WORLD can grow a labels dict later; do not let the model pass arbitrary PromQL. If you add a query_metrics(expr) tool, you have built SQL-injection-for-ops. Stick to get_metrics(service) until you have a parser.

Time is a list of snapshots if you need it: HISTORY[service] = [snapshot, ...]. The agent may read the last two to say "error_rate jumped after d44." Do not generate 10,000 fake points in the Try it box. Two snapshots already teach change, which is what diagnosis needs.

Read tools should be idempotent: same arguments, same result, until a mutate happens. If get_logs pops the ring buffer, you have built a destructive read. Tests will flake and diagnosis will lose evidence. Copy the tail; do not consume it. Metrics snapshots should be JSON-serializable so you can hash them in the audit log.

Exercise

Add a payments service that is healthy. Confirm list_services includes it. Add a log line containing sk-demo-not-real and a redact(lines) that replaces sk- tokens with sk-***. Logs tools must use it.

Check your understanding

Why must get_metrics return a copy of fields rather than the live nested dict with logs?

JSON cannot encode dicts
So the policy cannot mutate production state through a leftover list/dict reference
Copies are required for cosine similarity
Ops agents cannot read logs

Overview and Architecture

Up next

Diagnose and Propose Actions