Human-in-the-loop approval gate guarding an irreversible action

soma.toml

[package]
name = "agents-corpus"
version = "0.1.0"
description = "Verified AI-agent example cells (training corpus)"

# Agent LLM config. `mock = "echo"` keeps every example in this directory
# hermetic: `soma test` / `soma run` need no network and no API key, and
# results are deterministic. For a live run, delete the `mock` line (or
# override with SOMA_LLM_MOCK) and provide the key via env.
[agent]
provider = "anthropic"
model    = "claude-opus-4-6"
key      = "${ANTHROPIC_API_KEY}"
mock     = "echo"

# Termination proof for verified_termination.cell's `workflow` machine:
# every path eventually reaches a terminal, no deadlocks.
# `cells` scopes these properties to Solver — the other agents in this
# directory have their own state names and must not inherit them.
[verify]
cells = ["Solver"]
deadlock_free = true
eventually    = ["done", "failed"]

[verify.after.understanding]
eventually = ["solving", "failed"]

[verify.after.solving]
eventually = ["verifying", "failed"]

[verify.after.verifying]
eventually = ["done", "failed"]

approval_gate.cell

// Human-in-the-loop approval gate guarding an irreversible action.
// Demonstrates: approve() as a lifecycle gate, a state machine where the
// "executing" path is only reachable past the gate, and a rejected path
// that lands in a distinct terminal.
//
// approve() never answers on its own: a test scripts it with
// `mock approve true|false`; `soma run` asks the person at the terminal;
// under `soma serve` (nobody attached) it raises kind "approval_required"
// unless SOMA_APPROVE=always|never states an explicit policy.
//
// Run:  soma run approval_gate.cell deploy "v2.1.0"          (asks y/N)
//       SOMA_APPROVE=never soma run approval_gate.cell deploy "v2.1.0"
// Test: soma test approval_gate.cell

cell agent Deployer {
    face {
        signal deploy(version: String) -> Map
    }

    memory {
        history: List<String> [persistent]
    }

    state pipeline {
        initial: idle
        idle      -> planning
        planning  -> awaiting
        awaiting  -> executing
        awaiting  -> rejected
        executing -> done
        *         -> failed except [done, rejected]
    }

    on deploy(version: String) {
        set_budget(1500)
        let id = to_string(next_id())
        transition(id, "planning")
        transition(id, "awaiting")

        let approved = approve("Deploy {version} to production?")
        if !approved {
            transition(id, "rejected")
            return map("status", "rejected", "version", version)
        }

        transition(id, "executing")
        history.push(version)
        transition(id, "done")
        map("status", "done", "version", version, "count", len(history))
    }
}

cell test DeployTests {
    rules {
        mock approve true
        assert deploy("v1").status == "done"

        mock approve false
        assert deploy("v2").status == "rejected"

        // nobody answers: the gate fails closed and the handler rolls back
        assert_fails deploy("v3") matching "approval_required"

        // one answer per call: only v1 and v4 were ever deployed
        mock approve true
        let d4 = deploy("v4")
        assert d4.count == 2
    }
}