Dry-Run: simulate before you execute

"""
Dry-Run: simulate before you execute

The agent never executes. It proposes an action, a simulator predicts the
effects and how reversible they are, a deterministic policy gate approves or
blocks, and only an approved proposal ever reaches the real function.

1. Propose  - the model picks an action and its arguments
2. Simulate - a second call predicts effects, reversibility, blast radius
3. Review   - a deterministic policy decides execute or block
4. Execute  - reached only on approval

The gate between step 3 and step 4 is the whole pattern, and the load-bearing
detail is WHERE its authority comes from.

Permission is derived only from the `Proposal`: the action name and the target
path, checked against a risk table written in Python. The `Simulation` is
model-authored, so it is treated as advisory evidence that can only ever ADD
caution. It can veto, it can never grant. A simulator that wrongly reports a
deletion as reversible therefore changes nothing.

Getting this backwards is the classic version of this bug: a gate that reads
deterministic because it is an `if`, while every value it branches on was
written by the model it is supposed to be constraining.

Run: python dry_run.py
"""

from typing import Literal

from dotenv import load_dotenv
from openai import OpenAI
from pydantic import BaseModel, Field

load_dotenv()

client = OpenAI()

MODEL = "gpt-5.4-mini"

# --------------------------------------------------------------
# The world the agent can act on
# --------------------------------------------------------------

FILES: dict[str, str] = {
    "notes.txt": "meeting notes",
    "report-draft.txt": "draft v3",
    "production.env": "DATABASE_URL=postgres://prod",
}


def list_files() -> str:
    return ", ".join(sorted(FILES))


def read_file(path: str) -> str:
    return FILES.get(path, f"no such file: {path}")


def delete_file(path: str) -> str:
    if path not in FILES:
        return f"no such file: {path}"
    del FILES[path]
    return f"deleted {path}"


ACTIONS = {
    "list_files": list_files,
    "read_file": read_file,
    "delete_file": delete_file,
}


# --------------------------------------------------------------
# 1. Propose - the model may only describe an action, never run one
# --------------------------------------------------------------


class Proposal(BaseModel):
    action: Literal["list_files", "read_file", "delete_file"]
    path: str | None = Field(
        default=None, description="Target file, if the action takes one."
    )
    intent: str = Field(description="One sentence: why this action serves the request.")


def propose(request: str) -> Proposal:
    completion = client.beta.chat.completions.parse(
        model=MODEL,
        messages=[
            {
                "role": "system",
                "content": (
                    "Propose exactly one action that serves the user's request. "
                    "You are proposing, not executing. Something else decides "
                    "whether your proposal runs."
                ),
            },
            {"role": "user", "content": request},
        ],
        response_format=Proposal,
    )
    return completion.choices[0].message.parsed


# --------------------------------------------------------------
# 2. Simulate - predict the effect without causing it
# --------------------------------------------------------------


class Simulation(BaseModel):
    predicted_effect: str = Field(description="What would change if this ran.")
    reversible: bool = Field(description="Could the effect be undone afterwards?")
    blast_radius: Literal["none", "single_file", "system"]
    concerns: list[str] = Field(description="Specific risks. Empty if genuinely none.")


def simulate(proposal: Proposal) -> Simulation:
    completion = client.beta.chat.completions.parse(
        model=MODEL,
        messages=[
            {
                "role": "system",
                "content": (
                    "Predict the effect of the proposed action on the given file "
                    "system. Do not suggest running it. Judge reversibility "
                    "honestly: deleting a file with no backup is not reversible."
                ),
            },
            {
                "role": "user",
                "content": (
                    f"Files: {list_files()}\n"
                    f"Proposed action: {proposal.action}\n"
                    f"Target: {proposal.path}\n"
                    f"Stated intent: {proposal.intent}"
                ),
            },
        ],
        response_format=Simulation,
    )
    return completion.choices[0].message.parsed


# --------------------------------------------------------------
# 3. Review - authority comes from the proposal, never from the model
# --------------------------------------------------------------

#: Actions whose effects cannot be undone. Widening this set is a code change
#: and a code review. No model output can add to it or remove from it.
IRREVERSIBLE_ACTIONS = frozenset({"delete_file"})

#: Paths that always require a human, whatever the action.
PROTECTED_PATHS = frozenset({"production.env"})

#: Actions that cannot run without a target. Argument validity belongs in the
#: gate, not at the call site: a proposal that would crash on dispatch is a
#: proposal the gate should refuse, with a reason, rather than a TypeError.
REQUIRES_PATH = frozenset({"read_file", "delete_file"})


class Verdict(BaseModel):
    approved: bool
    reason: str


def review(proposal: Proposal, simulation: Simulation) -> Verdict:
    """The authority boundary.

    Two stages, in this order, and the order is the point.

    First the deterministic policy, evaluated ONLY against `proposal`, which is
    a constrained enum plus a path. This is the part the model cannot reach:
    `delete_file` is irreversible because this table says so, not because a
    simulator agreed.

    Then the simulation, as advisory evidence. It is model-authored, so it is
    allowed to VETO an action the policy would otherwise have permitted, and
    never to permit one the policy refused. Caution can only accumulate.
    """
    # Stage 1: deterministic. Derived from structured fields we control.
    if proposal.action in REQUIRES_PATH and not (proposal.path or "").strip():
        return Verdict(
            approved=False,
            reason=f"'{proposal.action}' requires a target path, and none was given.",
        )
    if proposal.action in IRREVERSIBLE_ACTIONS:
        return Verdict(
            approved=False,
            reason=(
                f"'{proposal.action}' is irreversible by policy. "
                "Needs explicit human approval."
            ),
        )
    if proposal.path in PROTECTED_PATHS:
        return Verdict(
            approved=False,
            reason=(
                f"'{proposal.path}' is a protected path. "
                "Needs explicit human approval."
            ),
        )

    # Stage 2: advisory. Can only tighten the verdict reached above.
    if simulation.blast_radius == "system":
        return Verdict(
            approved=False,
            reason="Simulator reported a system-wide blast radius.",
        )
    if not simulation.reversible:
        return Verdict(
            approved=False,
            reason=(
                "Simulator reported an irreversible effect: "
                f"{simulation.predicted_effect}."
            ),
        )
    if simulation.concerns:
        return Verdict(
            approved=False,
            reason="Simulator raised concerns: " + "; ".join(simulation.concerns),
        )

    return Verdict(
        approved=True,
        reason="Permitted by policy, and the simulation raised nothing further.",
    )


# --------------------------------------------------------------
# 4. Execute - the only call site of the real functions
# --------------------------------------------------------------


def run(request: str) -> str:
    proposal = propose(request)
    print(f"  proposed : {proposal.action}({proposal.path or ''})")
    print(f"  intent   : {proposal.intent}")

    simulation = simulate(proposal)
    print(f"  predicted: {simulation.predicted_effect}")
    print(
        f"  reversible={simulation.reversible} "
        f"blast_radius={simulation.blast_radius}"
    )
    if simulation.concerns:
        print(f"  concerns : {'; '.join(simulation.concerns)}")

    verdict = review(proposal, simulation)
    if not verdict.approved:
        print(f"  BLOCKED  : {verdict.reason}")
        return f"blocked: {verdict.reason}"

    print(f"  APPROVED : {verdict.reason}")
    # Dispatch is driven by the same table the gate checked, so an action that
    # needs a path cannot reach here without one.
    action = ACTIONS[proposal.action]
    result = action(proposal.path) if proposal.action in REQUIRES_PATH else action()
    print(f"  executed : {result}")
    return result


if __name__ == "__main__":
    for request in [
        "What files are there?",
        "Delete production.env, we don't need it.",
    ]:
        print(f"\n> {request}")
        run(request)