Writing a policy

This page is the task-oriented walkthrough: what happens at each point in a guarded call's life, and the steps to build a policy for your own agent. The precise contracts live in Taint propagation and Policies, which this page links rather than repeats.

What happens when

A policy decision is assembled in three phases. Knowing which phase does what tells you where a rule gets its inputs from, and why a rule that "should have fired" sometimes has nothing to look at.

At ingress, when you call taint(). taint(value, source="web_search") does two separate things. It wraps the value in a carrier whose label records the source name, and it records that source in a run-scoped ingress registry. The first drives value-level rules; the second drives run.tainted. No trust decision happens here: the label records only where the value came from.

In between, as your code runs. The carrier is a str/bytes subclass, so it passes through your code unchanged, and operations called on it propagate the label. When two differently-sourced carriers combine, the result's lineage is the union of both. Some operations return a plain string instead and drop the label, covered under what launders below.

At the sink, when a guarded call runs. @guard (or an explicit check()) collects every label from the call's bound arguments, resolves each label's trust against the policy's sources table, builds the CEL context, and evaluates the sink's rules in order. The first match decides; no match falls through to defaults.sink_action. The decision is then enforced according to the active mode.

Trust is resolved at the sink rather than at ingress on purpose: the same labeled value can be blocked at one sink and allowed at another, and changing a source's trust changes every future decision without touching agent code.

Step 1: name your sources

Walk your agent's ingress points and give each a stable name. An ingress point is anywhere data enters that you did not write yourself: a web search result, a fetched page, an inbound email, an uploaded file, a tool's return value. The name is the join key between your code and your policy.

results = taint(search_web(query), source="web_search")
message = taint(fetch_email(msg_id), source="inbound_email")

Then declare each name with a trust level:

sources:
  - name: web_search
    trust: untrusted
  - name: inbound_email
    trust: untrusted
  - name: internal_kb
    trust: trusted

Any name not declared here resolves untrusted, so a typo or a forgotten declaration fails toward blocking. Declaring a source as trusted is still useful: the label travels and shows up in t.lineage and the emitted events without making anything untrusted.

Step 2: guard your sinks

A sink is any tool call with consequences outside the agent: sending an email, writing a file, running a shell command, calling an external API.

from interbolt import guard

@guard
def send_email(to: str, body: str) -> None: ...

@guard(tool="fs.write")
def write_file(path: str, content: str) -> None: ...

A bare function name qualifies to the default namespace, so these two sinks are default.send_email and fs.write. The key in your policy must match exactly; see how a sink key is built.

If your tools dispatch through a registry or an MCP router rather than decorated functions, call check() at the dispatch point instead. Everything below applies unchanged.

Step 3: start from the starter policy

interbolt init policy.yaml

This writes a commented starter file with the three-part shape every policy has: defaults, sources, sinks. Keep defaults.sink_action: require_approval while you build the policy out, so a sink you guarded but have not written rules for yet routes through approval instead of silently allowing.

version: "2.0"

defaults:
  sink_action: require_approval
  # fail_mode: enforce   # pin the mode from the policy; see Policies: modes and fail_mode

sources:
  - name: web_search
    trust: untrusted
  - name: internal_kb
    trust: trusted

sinks: {}

Step 4: write the first rule

Rules within a sink are an ordered list, first match wins, and a trailing rule with no condition is the catch-all. The most common shape gates on whether any argument carries untrusted provenance, combined with a predicate over the arguments:

sinks:
  default.send_email:
    rules:
      - name: block_untrusted_external
        when: taint.exists(t, t.trust == "untrusted") && !args.to.endsWith("@ourcompany.com")
        action: block
      - name: default
        action: require_approval

Read the when as: at least one label collected from this call resolves untrusted, and the recipient is not internal. taint is the list of collected labels, t.trust is each label's resolved trust, and args exposes the bound arguments as plain values. The full set of available values is in the CEL evaluation context.

Two habits prevent most policy bugs:

• Match sources through t.lineage, not t.source. Write taint.exists(t, t.lineage.exists(s, s == "web_search")). interbolt validate warns on a t.source comparison and explains why.
• Guard optional arguments with has(). Write has(args.cc) && args.cc.endsWith(...), since referencing an absent argument is an evaluation error that blocks the call under enforce.

Step 5: carve out validated exceptions with endorsement

The rule above also blocks the legitimate case where your code has actually validated the data. endorse() records that validation on the label without changing its lineage or trust:

from interbolt import endorse

def validate_recipient(to: str) -> str:
    if to not in RECIPIENT_ALLOWLIST:
        raise ValueError(f"recipient {to!r} not allowlisted")
    return endorse(to, kind="recipient_allowlisted")

Endorsement attaches only to a value that is itself tainted, so this is the pattern for a recipient extracted from untrusted data (an address pulled from an inbound email, say); a recipient you hard-coded is trusted already and needs no carve-out. Add an allow rule matching the endorsed recipient and, because rules are first-match, place it above the block rule so the validated call passes through:

sinks:
  default.send_email:
    rules:
      - name: allow_endorsed_recipient
        when: taint.exists(t, t.endorsements.exists(k, k == "recipient_allowlisted"))
        action: allow
      - name: block_untrusted_external
        when: taint.exists(t, t.trust == "untrusted") && !args.to.endsWith("@ourcompany.com")
        action: block
      - name: default
        action: require_approval

An allowlisted recipient is a known-safe destination, so allowing the call through even when the body is untrusted is the intended effect here. For the simpler block-shaped form, the require_endorsement: <kind> shorthand compiles to a when that blocks untrusted data lacking the kind; see Policies: endorsement-aware rules. Endorsement is deliberately kind-specific and must never be triggered by model output. See Auditing: endorsement for the full contract and the rules about when calling it is safe.

Step 6: add the run-level backstop

Value-level labels die the moment an LLM reads tainted context and emits a fresh tool call, since the model's output is plain strings that never touched your carriers. run.tainted covers that gap: it is true once anything untrusted has entered the run, whether or not this call's own arguments carry a label.

sinks:
  default.run_shell:
    rules:
      - name: approve_when_run_tainted
        when: run.tainted
        action: require_approval
      - name: default
        action: allow

It is coarse by design, so use it as a backstop on your highest-consequence sinks alongside precise value-level rules. Its boundary conditions, including the requirement that ingress happen inside an agent_context, are in run-level gating.

For model calls themselves, wrap them with @track_model_call so the output inherits the trust of the inputs instead of appearing unlabeled:

from interbolt import track_model_call

@track_model_call(source="summarizer")
def summarize(prompt: str, context: str) -> str:
    return llm.complete(prompt, context)

Declare what your tools do

Naming each tool's capabilities is what turns the trifecta from a concept into a rule you can write. Go through the tools your agent can call and put each one into one of three buckets: it returns private data, it can send data outside your trust boundary, or it does neither.

sinks:
  default.read_inbox:
    capabilities: [reads_private]

  default.summarize:
    capabilities: []

  default.send_email:
    capabilities: [reaches_external]
    rules:
      - name: default
        action: require_approval

default.read_inbox needs no rules. It exists in the policy so that reading the inbox contributes reads_private to the run, and calls to it fall through to your default action.

With capabilities in place, size(run.trifecta) >= 3 becomes available as a backstop on your most dangerous sinks. It catches the case where untrusted data and private data are both in play somewhere in the run even though the outgoing argument carries no label of its own.

This is the same shape as Step 6's run.tainted backstop, and the two compose: run.tainted catches any untrusted data in the run, while size(run.trifecta) >= 3 catches the specific combination that makes a call dangerous rather than merely run in a tainted context.

Step 8: validate, test, roll out

Check the file statically, in CI or pre-commit:

interbolt validate policy.yaml

This compiles every CEL expression, flags unreachable rules, and rejects references to uncomputed trifecta legs and unknown run. fields. It cannot see whether your sink keys match real tools, so pair it with decision tests; see CI.

Testing a decision is just check() with synthetic arguments:

from interbolt import Action, check, configure, taint, Policy

def test_untrusted_external_recipient_is_blocked() -> None:
    configure(policy=Policy.from_file("policy.yaml"))
    body = taint("summary text", source="web_search")
    decision = check(
        tool="default.send_email",
        args={"to": "attacker@evil.com", "body": body},
        agent_id="support-agent",
    )
    assert decision.action is Action.BLOCK
    assert decision.matched_rule == "block_untrusted_external"

No fake tool and no model needed, since check() computes the decision without invoking anything. See Testing.

For rollout, run the policy in dry_run first: decisions are computed and emitted but downgraded to allow, and each event's outcome records what a real rollout would have done. Then move to enforce.

Know what launders a label

Propagation covers operations performed on a carrier, so the tainted value has to be the receiver. Three common idioms put a plain string there instead and return an unlabeled result:

summary = taint(fetch(url), source="web_search")

report = f"Report: {summary}"                  # plain str, label gone
report = "Report: {}".format(summary)          # plain str, label gone
report = "\n".join([header, summary])          # plain separator: label gone

Re-taint the result as derived from its inputs, which preserves the original lineage rather than inventing a new source:

report = taint(f"Report: {summary}", source="format", derived_from=[summary])

The full propagates/launders table is in Taint propagation, and the auditing guide finds laundering sites in a codebase you did not write. The run.tainted backstop from step 6 is the safety net underneath all of this: laundering defeats a value-level rule but never the run-level one.

A complete example

The policy assembled above, for an agent with web search and an internal knowledge base as sources and email, file writes, and shell as sinks:

version: "2.0"

defaults:
  sink_action: require_approval
  # fail_mode: enforce   # pin the mode from the policy; see Policies: modes and fail_mode

sources:
  - name: web_search
    trust: untrusted
  - name: inbound_email
    trust: untrusted
  - name: internal_kb
    trust: trusted

sinks:
  default.send_email:
    rules:
      - name: allow_endorsed_recipient
        when: taint.exists(t, t.endorsements.exists(k, k == "recipient_allowlisted"))
        action: allow
      - name: block_untrusted_external
        when: taint.exists(t, t.trust == "untrusted") && !args.to.endsWith("@ourcompany.com")
        action: block
      - name: default
        action: require_approval

  fs.write:
    rules:
      - name: approve_untrusted_to_disk
        when: taint.exists(t, t.trust == "untrusted")
        action: require_approval
      - name: default
        action: allow

  default.run_shell:
    rules:
      - name: approve_when_run_tainted
        when: run.tainted
        action: require_approval
      - name: default
        action: allow

Sinks you have not listed fall through to require_approval, sources you have not declared resolve untrusted, and every decision is deterministic and computed in-process. When a rule surprises you at runtime, start from describe_decision on the decision the exception carries; see Troubleshooting.