Invoice controls: exact destinations, private data, and quarantine

For child isolation, contract checks, and Argus crossing evidence in the dashboard, open the Airlock walkthrough.

Mix.install([{:kino, "~> 0.19.0"}, {:jason, "~> 1.4"}])

1. Connect

Start Dixie with a cookie

iex --name dixie@127.0.0.1 --cookie "testcookie" -S mix phx.server

repository_input = Kino.Input.text("Absolute Dixie repository path", default: File.cwd!())

dixie_node = :"dixie@127.0.0.1"
cookie = "testcookie"
Node.set_cookie(dixie_node, String.to_atom(cookie))
unless Node.connect(dixie_node), do: raise("Cannot connect to the dedicated Dixie node")

rpc = fn module, function, arguments ->
  :erpc.call(dixie_node, module, function, arguments, 30_000)
end

for {key, expected} <- [
      oracle_authorizer: Dixie.Warden.Oracle.Authorizer.Cedar,
      warden_policy_impl: Dixie.Warden.Policy.Live,
      warden_kernel_impl: Dixie.Warden.Kernel.Argus
    ] do
  unless rpc.(Application, :get_env, [:dixie, key]) == expected,
    do: raise("Enable #{key}=#{inspect(expected)} in the dedicated node's boot configuration")
end

alias Dixie.{Agents, Authz, Catalog, Warden}
alias Dixie.Agents.{Reconciler, Tools}
alias Dixie.Warden.{Chain, Invocation, Operator, Projection}
alias Dsxir.Primitives.Tool

root = Path.join(Kino.Input.read(repository_input), "examples/invoice_fraud")
python = System.find_executable("python3") || raise "Python 3.10+ is required"
script = Path.join(root, "invoice_world.py")
definitions = root |> Path.join("tools.json") |> File.read!() |> Jason.decode!()
policy_template = File.read!(Path.join(root, "advanced_policy.cedar"))

python_json = fn arguments ->
  {json, status} = System.cmd(python, [script | arguments], stderr_to_stdout: true)
  unless status in [0, 1], do: raise("Simulator failed: #{json}")
  Jason.decode!(json)
end

2. Choose a control and create a fresh world

• Destinations and least privilege: attempt a bank change, a different path, a different origin, and the wrong method; then make one permitted payment.
• Private-data exfiltration: permit a partner request before reading private purchase-order data, deny it afterwards, and still permit an internal lookup.
• Quarantine: commit a trusted payment but omit its durable receipt, then run the production reconciliation path. This creates genuine kernel quarantine.

These controls intentionally do not ingest raw invoices before making a payment: that would hit the original example's integrity denial before reaching the control being demonstrated. This is not a claim that raw invoices become trusted.

control_input = Kino.Input.select("Control", [
  quarantine: "Payment committed, receipt lost: quarantine",
  destinations: "Exact endpoint and least-privilege rules",
  exfiltration: "Private data blocks external verification"
])

control = Kino.Input.read(control_input)
run_id = rpc.(Ash.UUID, :generate, [])
directory = Path.join(root, "runs/advanced-#{run_id}")
File.mkdir_p!(directory)
db = Path.join(directory, "world.sqlite3")
identity = python_json.(["seed", "--db", db, "--scenario", "clean"])

{:ok, world_owner} = Kino.start_child({Agent, fn ->
  port = Port.open({:spawn_executable, python}, [
    :binary, :exit_status, {:line, 65536},
    args: [script, "serve", "--db", db, "--port", "0", "--shutdown-on-stdin-close"]
  ])

  receive do
    {^port, {:data, {:eol, line}}} -> %{port: port, ready: Jason.decode!(line), ran?: false}
    {^port, {:exit_status, status}} -> raise "Simulator exited: #{status}"
  after
    4000 -> raise "Simulator did not announce readiness"
  end
end})

ready = Agent.get(world_owner, & &1.ready)
true = ready["world_id"] == identity["world_id"]
origin = ready["url"]
Kino.Tree.new(%{control: control, origin: origin, database: db})

3. Inspect and install the policy

resource.attributes is a generic set of string name/value pairs. The Cedar schema does not know HTTP or invoices. Trusted tool configuration supplies invoice.operation; the HTTP adapter derives http.method, http.origin, and http.path from the same frozen realization its function executes. Caller arguments cannot supply or override them. HTTP redirects are not followed.

Rules use exact comparisons, not path prefixes. Origin includes scheme, host, and non-default port. Query parameters are not covered by the path comparison. These string attributes support membership/equality rules, not numeric amount comparisons. Capability and confidentiality conditions remain typed kernel data.

cedar = String.replace(policy_template, "__INVOICE_ORIGIN__", origin)
Kino.Text.new(cedar)

%{id: tenant} = rpc.(Dixie.Accounts, :get_tenant_by_slug!, ["dev-local", [authorize?: false]])
policy = root |> Path.join("warden_policy.json") |> File.read!() |> Jason.decode!()
policy = policy |> Map.put("owner_tenant_id", tenant) |> Map.put("guard_mode", "enforce")
rpc.(Authz, :upsert_warden_policy!, [policy])

case rpc.(Authz, :active_policy_for!, [tenant, [authorize?: false]]) do
  nil ->
    rpc.(Authz, :create_policy!, [%{owner_tenant_id: tenant, source: cedar, active: true}])

  existing ->
    rpc.(Authz, :update_policy!, [existing, %{source: cedar, active: true}])
end

private_order =
  definitions
  |> Enum.find(&(&1["name"] == "get_purchase_order"))
  |> Map.merge(%{"name" => "read_private_order", "output_conf" => "restricted"})

partner =
  definitions
  |> Enum.find(&(&1["name"] == "get_vendor"))
  |> Map.merge(%{"name" => "external_verification", "egress" => ["network_external"]})
  |> put_in(["realization", "authorization_attributes"], %{"invoice.operation" => "verification"})

entries = Map.new(definitions ++ [private_order, partner], fn definition ->
  logical_name = definition["name"]
  name = logical_name <> "-advanced"
  operation = if logical_name == "submit_payment", do: "payment", else: "read"

  definition =
    definition
    |> Map.put("name", name)
    |> Map.put("owner_tenant_id", tenant)
    |> update_in(["realization", "url"], &(origin <> URI.parse(&1).path))
    |> update_in(["realization", "authorization_attributes"], fn existing ->
      existing || %{"invoice.operation" => operation}
    end)

  entry = case rpc.(Catalog, :resolve_tool, [tenant, name]) do
    nil ->
      rpc.(Catalog, :create_entry!, [definition])

    existing ->
      updates = Map.drop(definition, ["owner_tenant_id", "name", "type"])
      rpc.(Catalog, :update_entry!, [existing, updates])
  end

  {logical_name, entry}
end)

specs = Map.new(entries, fn {name, entry} ->
  {name, rpc.(Catalog.Entry, :to_tool_spec, [entry, tenant])}
end)

agent = rpc.(Agents, :create_agent!, [%{
  tenant_id: tenant, name: "Invoice control #{run_id}",
  brain_type: :dsxir, brain_module: Dixie.Reasoning.Programs.LookupAgent
}])

rpc.(Agents, :spawn_session!, [%{
  tenant_id: tenant, agent_id: agent.id, brain_type: :dsxir, brain_dump: %{}
}])

# The projection registers a model identity but never invokes a model.
:ok = rpc.(Projection, :project, [tenant, agent.id, Map.values(specs), "openai:gpt-5.4-mini"])
deps = %{tenant_id: tenant, agent_id: agent.id}

execute = fn spec, args ->
  tool = rpc.(Tools, :gate, [spec, deps])
  rpc.(Tool, :execute, [tool, args])
end

payment = %{"invoice_id" => "invoice-1", "bank_account" => "ACME-001", "amount_cents" => 48000}

4. Execute once

The quarantine control intentionally separates the normal invocation bracket: write the pending ledger row, begin authorization, execute the HTTP effect, then stop before recording a result or settling. That reproduces the durable state of a crash after commit. It does not kill Dixie or pretend an HTTP error string causes quarantine. Reconciliation below is the real production implementation.

already_ran = Agent.get_and_update(world_owner, fn state -> {state.ran?, %{state | ran?: true}} end)
if already_ran, do: raise("Create a fresh world and agent before running again")

observations = case control do
  :destinations ->
    bank = execute.(specs["change_vendor_bank_account"], %{
      "vendor_id" => "vendor-1", "bank_account" => "ATTACKER-999"
    })

    variants = [
      {"different path", origin <> "/tools/change_vendor_bank_account", "POST"},
      {"different origin", String.replace(origin, "127.0.0.1", "localhost") <> "/tools/submit_payment", "POST"},
      {"wrong method", origin <> "/tools/submit_payment", "GET"}
    ]

    refused = Enum.map(variants, fn {label, url, method} ->
      entry = entries["submit_payment"]
      entry = %{entry | realization: Map.merge(entry.realization, %{"url" => url, "method" => method})}
      spec = rpc.(Catalog.Entry, :to_tool_spec, [entry, tenant])
      %{attempt: label, result: inspect(execute.(spec, payment))}
    end)

    [%{attempt: "administrative bank change", result: inspect(bank)}] ++ refused ++
      [%{attempt: "exact payment endpoint", result: inspect(execute.(specs["submit_payment"], payment))}]

  :exfiltration ->
    Enum.map([
      {"partner before private read", "external_verification", %{"vendor_id" => "vendor-1"}},
      {"private purchase order", "read_private_order", %{"purchase_order_id" => "po-1"}},
      {"partner after private read", "external_verification", %{"vendor_id" => "vendor-1"}},
      {"internal lookup still allowed", "get_vendor", %{"vendor_id" => "vendor-1"}}
    ], fn {label, name, args} -> %{attempt: label, result: inspect(execute.(specs[name], args))} end)

  :quarantine ->
    inv = rpc.(Ash.UUID, :generate, [])
    session = rpc.(Agents, :get_session!, [agent.id])
    row = %{"inv_id" => inv, "tool" => specs["submit_payment"].name, "args" => payment,
      "status" => "pending", "result" => nil, "retry_safe" => false}
    session = rpc.(Agents, :put_ledger!, [session, %{ledger: [row]}])

    {:ok, :allow, _context} = rpc.(Invocation, :begin, [tenant, %{
      agent: agent.id, tool: specs["submit_payment"].name, entry_or_snapshot: specs["submit_payment"],
      call_args: payment, inv: inv
    }])

    receipt = rpc.(:erlang, :apply, [specs["submit_payment"].function, [payment]]) |> Jason.decode!()
    true = receipt["invoice_id"] == "invoice-1"

    {envelopes, _anchor} = rpc.(Chain, :load, [tenant])
    {:ok, state} = rpc.(Warden, :project, [tenant])
    decisions = rpc.(Reconciler, :reconcile, [session.ledger, {envelopes, state.pending}])
    :ok = rpc.(Reconciler, :settle, [decisions, deps, session])
    [%{attempt: "commit without durable receipt, then reconcile", result: inspect(decisions)}]
end

Kino.DataTable.new(observations)

5. Inspect receipts and quarantine

Expected results:

• Destinations: four authorizer_denied ledger rows, one successful payment, and only submit_payment reaching Python.
• Exfiltration: the second partner request is denied, the internal lookup succeeds, and no payment occurs. This shows provenance containment, not a model detecting a malicious document.
• Quarantine: Python has one payment, the session is recovering, and the kernel retains a quarantined pending invocation. No payment retry is performed. Hosted AgentServer turns refuse a session with unresolved quarantine.

platform = python_json.(["report", "--db", db])
session = rpc.(Agents, :get_session!, [agent.id])
{:ok, state} = rpc.(Warden, :project, [tenant])
{envelopes, anchor} = rpc.(Chain, :load, [tenant])

pending = for {inv, invocation} <- state.pending, invocation.agent == agent.id do
  %{inv: inv, tool: invocation.policy.tool, quarantined: invocation.quarantined}
end

inv_ids = MapSet.new(session.ledger, & &1["inv_id"])
kernel_events = for envelope <- envelopes,
                    MapSet.member?(inv_ids, Map.get(envelope.action, :inv)) do
  envelope.action |> Map.from_struct() |> Map.take([:inv, :tool, :verdict, :outcome, :disposition])
end

report = %{control: control, tenant_id: tenant, agent_id: agent.id, platform: platform,
  ledger: session.ledger, status: session.status, quarantine: session.quarantine,
  pending: pending, kernel_events: kernel_events, chain_length: anchor.length}

Kino.Layout.tabs([
  {"Payment platform", Kino.Tree.new(platform)},
  {"Dixie ledger", Kino.Tree.new(session.ledger)},
  {"Quarantine", Kino.Tree.new(%{status: session.status, session: session.quarantine, kernel: pending})},
  {"Kernel events", Kino.Tree.new(kernel_events)}
])

File.write!(Path.join(directory, "report.json"), Jason.encode!(report, pretty: true), [:exclusive])
Kino.Text.new("Saved evidence in #{directory}")

6. Resolve the committed payment, without replaying it

Quarantine control only. Inspect the Python ledger first. A lost receipt is not proof of failure: this payment actually committed. The local operator attests success, completes the original ledger row, and clears the session's recovery state. The original effect is never re-executed.

The actor below represents the trusted operator of the dedicated dev-local demo. Real operator endpoints must derive actor identity and role from authentication; never accept a caller's self-asserted admin map.

verified_input = Kino.Input.checkbox("I checked the payment ledger and verified the committed payment", default: false)

unless control == :quarantine and Kino.Input.read(verified_input),
  do: raise("Select the quarantine control and verify the platform receipt first")

current = python_json.(["report", "--db", db])
[%{"invoice_id" => "invoice-1", "bank_account" => "ACME-001", "amount_cents" => 48000}] = current["payments"]
session = rpc.(Agents, :get_session!, [agent.id])
[%{"inv_id" => inv}] = session.quarantine
actor = %{role: :admin, tenant_id: tenant}
{:ok, _resolution} = rpc.(Operator, :resolve_quarantine, [tenant, inv, :success, [actor: actor]])

ledger = Enum.map(session.ledger, fn row ->
  if row["inv_id"] == inv, do: Map.put(row, "status", "completed"), else: row
end)
session = rpc.(Agents, :put_ledger!, [session, %{ledger: ledger}])
session = rpc.(Agents, :resolve_quarantine!, [session, %{quarantine: [], status: :idle}])
{:ok, resolved_state} = rpc.(Warden, :project, [tenant])
false = Enum.any?(resolved_state.pending, fn {pending_inv, _} -> pending_inv == inv end)

Kino.Tree.new(%{status: session.status, quarantine: session.quarantine,
  payments: python_json.(["report", "--db", db])["payments"]})

7. Stop the simulator

This closes the local server, not the retained audit records or SQLite database. The quarantine evidence remains until explicitly resolved. To run another control, reevaluate from section 2 to create a fresh agent and world under dev-local.

Kino.terminate_child(world_owner)
Kino.Text.new("Simulator stopped; evidence retained in #{directory}")