Projects/Autonomous Ops Agent/Part 4

HUMAN_APPROVAL Gate

Implement execute() so every mutate tool checks an approval map, records attempts, and never changes the world on deny or missing approval.

The gate is the load-bearing wall. If it is a comment in the prompt ("please don't rollback without asking"), it will fail. This part implements execute(proposal, approval_map) in Python. The invariant: WORLD does not change unless approval_map.get(proposal_id) == "allow" and the tool is in the mutate allowlist and args match the proposal (no bait-and-switch).

Bait-and-switch

A hostile policy proposes rollback to d43, gets approval, then executes rollback to d0-experimental. Prevent this by approving a hash of the proposal (tool + canonical JSON args), not a bare id. Store:

HUMAN_APPROVAL[action_id] = {"decision": "allow", "digest": sha256(canonical)}

Execute recomputes digest; mismatch → deny.

execute outcomes

Case	Result	World
read tool	run	unchanged (reads)
mutate, no key	`needs_approval`	unchanged
mutate, deny	`denied`	unchanged
mutate, allow, digest match	`applied`	changed
unknown tool	`unknown_tool`	unchanged
digest mismatch	`tamper`	unchanged

Simulate rollback effect

On approved rollback, set error_rate to 0.004, p95_ms to 220, deploy_id to to, append a log line rollback applied. That lets the loop detect resolution.

Live Pythonpython

import hashlib
import json

WORLD = {
    "checkout": {
        "error_rate": 0.12,
        "p95_ms": 1800,
        "deploy_id": "d44",
        "prev_deploy_id": "d43",
        "replicas": 3,
        "logs": ["bad"],
    }
}
MUTATE = {"rollback_deploy", "scale_replicas", "restart_service"}
AUDIT = []

def canonical(proposal):
    payload = {"tool": proposal["tool"], "args": proposal["args"], "id": proposal["id"]}
    return json.dumps(payload, sort_keys=True, separators=(",", ":"))

def digest(proposal):
    return hashlib.sha256(canonical(proposal).encode()).hexdigest()

def rollback_deploy(service, to):
    row = WORLD[service]
    row["deploy_id"] = to
    row["error_rate"] = 0.004
    row["p95_ms"] = 220
    row["logs"].append(f"rollback applied to {to}")
    return {"applied": True, "deploy_id": to}

IMPL = {"rollback_deploy": lambda **kw: rollback_deploy(kw["service"], kw["to"])}

def execute(proposal, approval):
    tool = proposal["tool"]
    if tool not in MUTATE:
        return {"status": "unknown_tool"}
    d = digest(proposal)
    rec = approval.get(proposal["id"])
    if rec is None:
        AUDIT.append({"id": proposal["id"], "status": "needs_approval", "digest": d})
        return {"status": "needs_approval", "digest": d, "proposal": proposal}
    if rec.get("decision") != "allow":
        AUDIT.append({"id": proposal["id"], "status": "denied"})
        return {"status": "denied"}
    if rec.get("digest") != d:
        AUDIT.append({"id": proposal["id"], "status": "tamper"})
        return {"status": "tamper"}
    result = IMPL[tool](**proposal["args"])
    AUDIT.append({"id": proposal["id"], "status": "applied", "digest": d})
    return {"status": "applied", "result": result}

prop = {
    "id": "act-01",
    "tool": "rollback_deploy",
    "args": {"service": "checkout", "to": "d43"},
}
print("before", WORLD["checkout"]["error_rate"], WORLD["checkout"]["deploy_id"])
print("no approval", execute(prop, {}))
print("still", WORLD["checkout"]["error_rate"])

approval = {"act-01": {"decision": "allow", "digest": digest(prop)}}
print("allow", execute(prop, approval))
print("after", WORLD["checkout"]["error_rate"], WORLD["checkout"]["deploy_id"])

# tamper: same id, different args
WORLD["checkout"]["error_rate"] = 0.12
WORLD["checkout"]["deploy_id"] = "d44"
evil = dict(prop)
evil["args"] = {"service": "checkout", "to": "d0-experimental"}
print("tamper", execute(evil, approval))
print("tamper world", WORLD["checkout"]["deploy_id"])
print("audit", json.dumps(AUDIT, indent=2))

Output

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

Step-by-step gate wiring in the agent loop

Diagnose + propose (part 3).
Call execute. If needs_approval, finish with status awaiting_approval and include digest + proposal in the incident draft. Do not retry execute in a for-loop.
A second run (new process or next user click) loads the same proposal and an updated HUMAN_APPROVAL.
On applied, re-read metrics; if not burning, go to report.
On denied / tamper, report and stop.

In the Try it box you can run both phases in one script: first execute empty approval, then set approval, execute again.

page_oncall

If you add it, decide: is it mutate? It pages a human — noisy, not destructive. You can allow it without digest, but rate-limit to 1 per incident. Still audit.

Watch out:Logging approval maps must not become a way to smuggle decision: allow from the model. The model must not write HUMAN_APPROVAL. Only the supervisor / UI writes it.

Dual control and expiry

Production gates add expiry: an approval older than 15 minutes is dead. Incidents drift; a rollback approved for d44 must not apply after someone already shipped d45. In this project you can store approved_at_step and reject if current_step - approved_at_step > 3. Also bind deploy_id at proposal time: if WORLD's current deploy changed, digest still matches the old args but you should re-check args["to"] == prev_deploy_id now. Stale approvals are a real outage class.

Dual control (two humans) is a product flag. Your tests should still pass with one fake human. The code path is the same: more keys in the approval record, all(decisions == allow).

Never implement "auto-approve if sev1." Severity is a paging signal, not a sudoers file. The eval that forbids mutate without approval must not have a severity exception, or you will ship the exception.

Exercise

Write a test: approval allow for act-01, execute act-01 with extra args key force: true → digest mismatch → tamper → world unchanged. Add restart_service impl that requires approval and sets a log line without changing deploy_id.

Check your understanding

What does the gate approve — the action id string alone?

Yes, ids are enough
The id plus a digest of tool and args so the payload cannot change after approval
The model's thought
Any mutate after sev1

Diagnose and Propose Actions

Up next

Incident Report and Hardening