===============
LIBRARY RULES
===============
From library maintainers:
- Prefer the documented Drawbore public APIs shown in the guides and examples.
- Use pipeline.test_mode for local tests that mock LLMs, tools, MCP servers, databases, and external APIs.
- Use explicit From(...) bindings when a step consumes fields from earlier steps.
- Treat pipeline JSON as workflow shape and policy, not as runtime credentials or live objects.
- Assert on RunResult and audit_trace when testing pipeline behavior.

### Install and Test Tiny Pipeline

Source: https://github.com/daviesayo/drawbore/blob/main/examples/tiny_pipeline/README.md

Installs the Drawbore library and runs the tests for the tiny pipeline example. This pipeline is fully deterministic and does not require mocks when run with `pipeline.test_mode()`.

```bash
pip install drawbore
pytest examples/tiny_pipeline/test_tiny_pipeline.py
```

--------------------------------

### Install Drawbore Core and Optional Packages

Source: https://github.com/daviesayo/drawbore/blob/main/README.md

Install the core Drawbore library or include optional dependencies for MCP server tools or OTLP span export.

```bash
pip install drawbore            # core
pip install "drawbore[mcp]"     # + MCP server tools
pip install "drawbore[otlp]"    # + OTLP span export
```

--------------------------------

### Install Drawbore

Source: https://github.com/daviesayo/drawbore/blob/main/docs/quickstart.mdx

Install the Drawbore package using pip. This is the first step to using Drawbore.

```bash
pip install drawbore
```

--------------------------------

### Run Risk Branch Pipeline Example

Source: https://github.com/daviesayo/drawbore/blob/main/examples/risk_branch/README.md

Execute the pytest command to run the risk branch pipeline example. No provider or API key is required as all agents are deterministic.

```bash
pytest examples/risk_branch -v
```

--------------------------------

### Run Live OpenAI Example

Source: https://github.com/daviesayo/drawbore/blob/main/docs/examples/live-openai-smoke.mdx

Executes the Python script to run the live OpenAI smoke test. Ensure the PYTHONPATH is set correctly and the OpenAI API key is provided as an environment variable.

```bash
PYTHONPATH=src OPENAI_API_KEY=sk-... python3.11 examples/live_openai_smoke/pipeline.py
```

--------------------------------

### Register MCP Server with Fake Client

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/mcp-tools.mdx

Register an MCP server with a fake client for testing. This example shows how to define the tools the fake server advertises and specify which tools are allowed for an agent.

```python
from drawbore.tools import ToolRegistry
from drawbore.mcp import register_mcp_server, FakeMCPClient, MCPToolSpec

registry = ToolRegistry()

# In production you supply a LiveMCPClient (see below); FakeMCPClient is the
# dependency-free test double — give it the tools the fake server advertises.
client = FakeMCPClient(tools={
    "send_message": MCPToolSpec(name="send_message", input_schema={}),
    "delete_channel": MCPToolSpec(name="delete_channel", input_schema={}),
})

await register_mcp_server(
    registry,
    name="slack",
    url="https://slack.com/mcp",
    allowed_tools=["send_message"],          # one tool, not all of Slack
    client=client,                           # the transport you supply
)
```

--------------------------------

### Install and Run Remittance Validation Test

Source: https://github.com/daviesayo/drawbore/blob/main/examples/remittance_validation/README.md

Install the Drawbore library and run the pytest suite for the remittance validation example. This command executes the tests defined in the specified Python file.

```bash
pip install drawbore
pytest examples/remittance_validation/test_remittance_validation.py
```

--------------------------------

### Register Live MCP Server

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/mcp-tools.mdx

Registers a live MCP server with the specified registry. Ensure you have installed the 'drawbore[mcp]' extra for LiveMCPClient.

```python
from drawbore.mcp import LiveMCPClient, register_mcp_server

await register_mcp_server(
    registry,
    name="slack",
    url="https://slack.com/mcp",
    allowed_tools=["send_message"],
    client=LiveMCPClient(),                   # requires `pip install drawbore[mcp]`
)
```

--------------------------------

### Configure Agent for Evidence Retrieval

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/evidence-compression.mdx

To enable an agent to retrieve original evidence, declare `EVIDENCE_TOOL_REF` in its `tools` and pass the `evidence_store` to `Pipeline.run`. This single argument handles both compression storage and retrieval setup.

```python
from drawbore.evidence import EVIDENCE_TOOL_REF, EvidencePolicy, InMemoryEvidenceStore

store = InMemoryEvidenceStore()

@agent(input=Transactions, output=Verdict, model="m", tools=[EVIDENCE_TOOL_REF])
async def screen(v, tools):
    hits = await tools.call(EVIDENCE_TOOL_REF, {"handle_id": h, "mode": "search", "query": "high_risk"})

p = Pipeline(name="aml")
p.add(screen, evidence=EvidencePolicy(name="aml", enabled=True, min_tokens=800))

result = await p.run(transactions, engine=engine, evidence_store=store)
```

--------------------------------

### Profile-Bound Pipeline Safety Test

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/safety-gauntlet.mdx

When agents bind models by profile, forward the runtime configuration to resolve the profile correctly during safety testing. This example demonstrates setting up LLMRuntimeConfig with profiles and providers, and then using it with assert_contained for a schema violation test.

```python
from drawbore.llm import LLMRuntimeConfig, ModelProfile, ModelTarget, ProviderConfig
from drawbore.testing import StaticCredentialChecker, assert_contained, schema_violation

llm_config = LLMRuntimeConfig(
    profiles={"judgment": ModelProfile(targets=(
        ModelTarget(provider="openrouter", model="anthropic/claude-3-5-sonnet"),
    ))},
    providers={"openrouter": ProviderConfig(credential_env="OPENROUTER_API_KEY")},
)

case = schema_violation("risk_scorer", {"missing": "required fields"})
await assert_contained(
    pipeline, case, initial=my_input,
    llm_config=llm_config,
    credential_checker=StaticCredentialChecker(available=True),
)
```

--------------------------------

### Run Pipeline Test

Source: https://github.com/daviesayo/drawbore/blob/main/docs/examples/risk-branch.mdx

Executes the pytest suite for the risk branch pipeline example. This command is used to verify the functionality of the pipeline defined in the Python code.

```bash
pytest examples/risk_branch/test_risk_branch.py
```

--------------------------------

### Get Schema Relaxation Diff

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/authority-regression.mdx

Use this function to get the raw schema relaxation diff without raising an error. The `diff.ok` attribute indicates if any relaxation was found.

```python
from drawbore.config import schema_relaxation_diff

diff = schema_relaxation_diff(old, new)
if not diff.ok:
    print(diff.legible())   # same certificate the error carries
```

--------------------------------

### Register Agents and Export/Import Pipeline Manifest

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/json-config.mdx

This snippet demonstrates how to use `AgentCatalog` to register agent functions and then export a live pipeline to a JSON manifest using `to_json`. It also shows how to rebuild a pipeline from a reviewed manifest using `from_json`, requiring the catalog and a registry.

```python
from drawbore.config import AgentCatalog, to_json, from_json

catalog = AgentCatalog()
catalog.register("examples.remittance:screen_transaction", screen_transaction)
catalog.register("examples.remittance:fetch_transaction", fetch_transaction)

# Export a live pipeline to a reviewable manifest:
manifest = to_json(pipeline, agents=catalog)

# Rebuild a pipeline from a reviewed manifest:
pipeline = from_json(manifest, agents=catalog, registry=registry)
```

--------------------------------

### Configure OTLP Export for Spans

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/observability-audit.mdx

Use this function to export OpenTelemetry spans to an OTLP backend. Ensure you have installed the optional `otlp` extra for Drawbore.

```python
from drawbore.observability import configure_otlp_export

configure_otlp_export("https://otlp.your-drain.example", headers={"authorization": "..."})
# requires `pip install drawbore[otlp]`
```

--------------------------------

### Run Pipeline and Verify Confinement Receipt

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/confinement-receipt.mdx

This snippet demonstrates how to run a pipeline, obtain its confinement receipt, and then verify the receipt's status. It's useful for initial checks after a pipeline run.

```python
import asyncio
from drawbore.confinement import verify

result = await pipeline.run(initial)

receipt = result.confinement_receipt
verdict = verify(receipt)
print(verdict.status)          # "confined" | "breached" | "unverifiable"
print(receipt.legible())       # regulator-readable block
```

--------------------------------

### Define and Run an Agent with Tool Declarations

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/agentic-tool-loop.mdx

Define an agent function that declares its tools and model. Then, set up an LLM runtime and run the agent within a Drawbore Pipeline using the ADKEngine.

```python
from drawbore import Pipeline, agent
from drawbore.orchestration import ADKEngine
from drawbore.llm import LLMRuntime, LLMRuntimeConfig, ModelProfile, ModelTarget, ProviderConfig

@agent(name="screen", input=Transactions, output=Verdict, model="gpt-4o", tools=["risk_lookup"])
async def screen(txns, tools):
    ...  # the function body is not run on the model path

runtime = LLMRuntime(config=LLMRuntimeConfig(
    profiles={"default": ModelProfile(targets=(
        ModelTarget(provider="openai", model="gpt-4o"),
    ))},
    providers={"openai": ProviderConfig(credential_env="OPENAI_API_KEY")},
))

p = Pipeline(name="aml", registry=registry)
p.add(screen)
result = await p.run(transactions, engine=ADKEngine(llm_runtime=runtime))
```

--------------------------------

### Compose and Run a Pipeline

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/pipelines.mdx

Create a Drawbore pipeline, add agents to it, and run the pipeline with initial input data. Asserts the completion status and output value.

```python
from drawbore import Pipeline, From

pipeline = Pipeline(name="example")
pipeline.add(double)

result = await pipeline.run(Seed(value=5))
assert result.status == "completed"
assert result.outputs["double"].doubled == 10
```

--------------------------------

### Run Remittance Validation Tests

Source: https://github.com/daviesayo/drawbore/blob/main/docs/examples/remittance-validation.mdx

This command executes the tests for the remittance validation pipeline using pytest. Ensure you have pytest installed and the test file is in the correct location.

```bash
pytest examples/remittance_validation/test_remittance_validation.py
```

--------------------------------

### Get Authority Difference

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/authority-regression.mdx

Obtain a diff object representing authority changes between two pipeline manifests without raising an exception. The diff object contains a human-readable and JSON certificate.

```python
from drawbore.config import authority_diff

diff = authority_diff(old, new)
if not diff.ok:
    print(diff.certificate())   # same human-readable + JSON payload the error carries
```

--------------------------------

### Fan-in Topology - Corrected Binding

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/pipelines.mdx

Demonstrates the correct approach for fan-in topologies by declaring explicit From(...) bindings on every branch to ensure data is sourced from the intended agent.

```python
pipeline.add(router)
pipeline.add(branch_a, inputs={"x": From("router.x")})
pipeline.add(branch_b, inputs={"x": From("router.x")})  # explicit — reads from router
pipeline.add(Join("merged", sources=["branch_a", "branch_b"],
                  policy="all_present", output=MergedResult,
                  inputs={"a_out": From("branch_a.result"),
                          "b_out": From("branch_b.result")}))
```

--------------------------------

### Manage Agent Deployment with Rollback

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/identity-versioning.mdx

Initiate a deployment with a defined rollout plan and classify the change kind. Rollback to the last stable version is available if a gate fails.

```python
from drawbore.versioning import Deployment, RolloutPlan

dep = Deployment(stable_version="1.0.0")
dep.begin(RolloutPlan(change_kind=kind, new_version="1.1.0"))
# ... each stage's gate must pass before advancing; a failed gate ...
dep.rollback()   # ... routes back to the last stable version, atomically
```

--------------------------------

### Define and Run a Simple Agent Pipeline

Source: https://github.com/daviesayo/drawbore/blob/main/README.md

Demonstrates defining a Pydantic model, an agent function, and a pipeline. The pipeline is then run in test mode to process payment data and print an audit trace.

```python
from pydantic import BaseModel
from drawbore import agent, Pipeline

class Payment(BaseModel):
    cents: int

class Normalized(BaseModel):
    dollars: float

@agent(name="normalize", input=Payment, output=Normalized)
async def normalize(p: Payment) -> Normalized:
    return Normalized(dollars=p.cents / 100)

pipeline = Pipeline(name="payments", version="1.0.0")
pipeline.add(normalize)

async def main():
    async with pipeline.test_mode() as tp:
        result = await tp.run(Payment(cents=2500))
    assert result.status == "completed"
    print(result.audit_trace.legible())   # a regulator-readable run record
```

--------------------------------

### Configure LLMRuntime for OpenAI

Source: https://github.com/daviesayo/drawbore/blob/main/docs/examples/live-openai-smoke.mdx

Sets up the LLMRuntime with a specific OpenAI model profile and configures the OpenAI provider with an environment variable for credentials. Use this to define your LLM backend.

```python
runtime = LLMRuntime(config=LLMRuntimeConfig(
    profiles={
        "judgment": ModelProfile(targets=(
            ModelTarget(provider="openai", model="gpt-4o-mini"),
        )),
    },
    providers={
        "openai": ProviderConfig(credential_env="OPENAI_API_KEY"),
    },
))
result = await pipeline.run(input, engine=ADKEngine(llm_runtime=runtime))
```

--------------------------------

### Resume Halted Pipeline Run with Checkpoints

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/reliability.mdx

Initialize a checkpoint store and run a pipeline. The first run may halt, and the second run with the same ID and store will resume from the point of failure.

```python
from drawbore.state import InMemoryCheckpointStore

store = InMemoryCheckpointStore()
first = await pipeline.run(order, run_id="order-A-1", checkpoints=store)   # halts at step 3
again = await pipeline.run(order, run_id="order-A-1", checkpoints=store)   # resumes at step 3
```

--------------------------------

### Register an Agent with Identity

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/identity-versioning.mdx

Register a new agent with its sponsor, purpose, and a time-to-live. The sponsor is always required.

```python
from drawbore.identity import IdentityRegistry

registry = IdentityRegistry()
registry.register(
    scorer.spec,
    sponsor="alice@bank.example",   # always required — no agent without an owner
    purpose="score remittance risk",
    ttl_seconds=86_400,
)
```

--------------------------------

### Bind Data Between Steps with From

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/pipelines.mdx

Explicitly bind a field from a previous step's output to the input of a subsequent step using the From helper.

```python
pipeline.add(decide, inputs={"doubled": From("double.doubled")})
```

--------------------------------

### Initializing and Using Effect Ledger with Checkpoints

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/durable-resume.mdx

Integrate the effect ledger with a checkpoint store when running a pipeline. This ensures that effectful calls are recorded and replayed, not re-fired, upon resuming.

```python
from drawbore.state import FileCheckpointStore, InMemoryEffectLedger

store = FileCheckpointStore("/var/lib/myapp/checkpoints")
ledger = InMemoryEffectLedger()

first = await pipeline.run(order, run_id="order-A-1", checkpoints=store, effect_ledger=ledger)
# process crashes partway...
again = await pipeline.run(order, run_id="order-A-1", checkpoints=store, effect_ledger=ledger)
# effectful calls already recorded are replayed, never re-fired
```

--------------------------------

### Phase 2: Approve as-is with Decision

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/human-approval.mdx

Submit an `ApprovalDecision` with the verdict 'approved' to continue the pipeline. The output is checkpointed, and the run proceeds from the next step.

```python
from drawbore.escalation import ApprovalDecision

# ---- approve as-is ----
decision = ApprovalDecision(
    request_id=request.request_id,
    verdict="approved",
    reviewer_id="alice@example.com",
)

result = await pipeline.run(
    initial,
    run_id="order-A-1",
    checkpoints=store,
    approval=decision,
)
# result.status == "completed" (if no later step halts)
```

--------------------------------

### Run an agent with an LLM runtime

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/model-backed-agents.mdx

Configure and use an LLMRuntime to execute a pipeline containing model-backed agents. The runtime handles provider resolution, credential checks, and fallback attempts. The framework validates the model's output against the specified Pydantic model.

```python
from drawbore import Pipeline
from drawbore.orchestration import ADKEngine
from drawbore.llm import LLMRuntime, LLMRuntimeConfig, ModelProfile, ModelTarget, ProviderConfig

pipeline = Pipeline(name="screening")
pipeline.add(score)

runtime = LLMRuntime(config=LLMRuntimeConfig(
    profiles={
        "judgment": ModelProfile(targets=(
            ModelTarget(provider="openai", model="gpt-4o-mini"),
        )),
    },
    providers={
        "openai": ProviderConfig(credential_env="OPENAI_API_KEY"),
    },
))

result = await pipeline.run(tx, engine=ADKEngine(llm_runtime=runtime))
```

--------------------------------

### LLM Runtime Configuration

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/production-llm-gateway.mdx

Configure the LLMRuntime with model profiles and provider settings. This includes defining available model targets for each profile and specifying credentials for providers.

```python
from drawbore.llm import LLMRuntime, LLMRuntimeConfig, ModelProfile, ModelTarget, ProviderConfig
from drawbore.orchestration import ADKEngine

runtime = LLMRuntime(config=LLMRuntimeConfig(
    profiles={
        "judgment": ModelProfile(targets=(
            ModelTarget(provider="openrouter", model="anthropic/claude-3-5-sonnet"),
            ModelTarget(provider="openai", model="gpt-4o"),
        )),
        "judgment_backup": ModelProfile(targets=(
            ModelTarget(provider="openai", model="gpt-4o"),
        )),
    },
    providers={
        "openrouter": ProviderConfig(credential_env="OPENROUTER_API_KEY"),
        "openai": ProviderConfig(credential_env="OPENAI_API_KEY"),
    },
))
result = await pipeline.run(input, engine=ADKEngine(llm_runtime=runtime))
```

--------------------------------

### Enable Evidence Compression for a Pipeline Step

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/evidence-compression.mdx

Enable evidence compression for a specific step in a Drawbore pipeline. Set `enabled=True` and define `min_tokens` to trigger compression when the evidence exceeds this threshold. The original evidence is stored separately.

```python
from drawbore import Pipeline
from drawbore.evidence import EvidencePolicy, InMemoryEvidenceStore

p = Pipeline(name="aml")
p.add(
    screen,                                   # a model-backed agent (@agent(model=...))
    evidence=EvidencePolicy(name="aml", enabled=True, min_tokens=800),
)

store = InMemoryEvidenceStore()
result = await p.run(transactions, engine=engine, evidence_store=store)
```

--------------------------------

### Resume a Pipeline Run with FileCheckpointStore

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/durable-resume.mdx

Use `FileCheckpointStore` to persist checkpoints to a directory, enabling resumes across process restarts. The first process runs and persists state, while a subsequent process using the same directory restores and resumes.

```python
from drawbore.state import FileCheckpointStore

# First process: runs until it halts, persisting each completed step.
store = FileCheckpointStore("/var/lib/myapp/checkpoints")
first = await pipeline.run(order, run_id="order-A-1", checkpoints=store)

# A later process (a restart): a brand-new store over the same directory
# restores the completed steps and resumes — no re-execution, no spurious drift.
store = FileCheckpointStore("/var/lib/myapp/checkpoints")
again = await pipeline.run(order, run_id="order-A-1", checkpoints=store)
```

--------------------------------

### Admitting a Candidate Pipeline Manifest

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/safety-ratchet.mdx

This snippet demonstrates the core workflow of admitting a candidate pipeline manifest using the `admit()` function. It covers observing the baseline, seeding the regression corpus, and gating the proposed manifest. Use `FileRegressionCorpus` for production to ensure the corpus survives process restarts.

```python
import asyncio

from drawbore.config import AgentCatalog, to_config
from drawbore.ratchet import (
    FileRegressionCorpus, InMemoryRatchetSink,
    admit, derive_cases,
)

# 1. Observe the baseline: run your pipeline once in test mode and keep the
#    serializable mock subset you used (tools + one-shot model responses).
config = to_config(pipeline, agents=catalog)
async with pipeline.test_mode(**full_mocks) as tp:
    baseline = await tp.run(initial)

# 2. Seed the corpus mechanically — cases are derived, never hand-written.
#    Use FileRegressionCorpus so the corpus survives a process restart.
corpus = FileRegressionCorpus("/var/lib/myapp/ratchet-corpus")
for case in derive_cases(config, pipeline, baseline, initial=initial,
                         baseline_mocks=serializable_mocks,
                         derived_at="2026-06-13T00:00:00Z", predecessor=None):
    corpus.append(case, sponsor="a.reviewer")

# 3. Gate every proposed manifest.  A fresh instance over the same directory
#    reloads the persisted corpus on the first read — no explicit load call needed.
corpus = FileRegressionCorpus("/var/lib/myapp/ratchet-corpus")
verdict = await admit(candidate_json, agents=catalog, corpus=corpus,
                      sponsor="a.reviewer", baseline_config=config,
                      initial_input=initial, baseline_mocks=serializable_mocks,
                      derived_at="2026-06-13T01:00:00Z", registry=registry,
                      sink=InMemoryRatchetSink())
if verdict.admitted:
    active_pipeline = verdict.pipeline   # adoption = replacing your reference
else:
    print(verdict.legible())             # which layer refused, and why
```

--------------------------------

### Register Pipeline with Registry

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/tools.mdx

Create a Pipeline and associate it with a ToolRegistry. The pipeline will reject agents that declare tools not present in the registry.

```python
from drawbore import Pipeline

pipeline = Pipeline(name="confirm", registry=registry)
pipeline.add(retrieve)   # rejected at this line if 'retrieve' declares a tool
                         # that isn't registered
```

--------------------------------

### Importing the verify function

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/confinement-receipt.mdx

Import the `verify` function from the `drawbore.confinement` module to perform offline verification of confinement receipts.

```python
from drawbore.confinement import verify
```

--------------------------------

### Verify Confinement Receipt with Expected Footprint Fingerprint

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/confinement-receipt.mdx

This snippet demonstrates how to verify a confinement receipt against an independently computed footprint fingerprint. This is the strongest defense against inflated declared facts, ensuring the receipt's footprint matches the manifest's.

```python
from drawbore.config.authority import effective_authority

config = pipeline.to_config()
expected_fp = effective_authority(config).fingerprint()
verdict = verify(receipt, expected_footprint_fingerprint=expected_fp)
```

--------------------------------

### Configure Pipeline with Confidence Threshold and Human Approval

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/escalation.mdx

Define a Pydantic model with confidence tracking for risk assessment. Set a pipeline-wide confidence threshold and mark an agent to require human approval.

```python
from pydantic import BaseModel
from drawbore import Pipeline, agent
from drawbore.escalation import EscalationPolicy, HasConfidence

class RiskScore(BaseModel, HasConfidence):
    score: int
    confidence: float          # 0..1 — the framework never invents this

@agent(input=Order, output=RiskScore, requires_human_approval=True)
async def assess(order: Order) -> RiskScore:
    ... # implementation omitted

# the threshold lives on the pipeline; a RiskScore below it escalates
pipeline = Pipeline(
    name="payments",
    on_failure=EscalationPolicy(channel="slack", target="compliance_queue"),
    confidence_threshold=0.80,
)
```

--------------------------------

### Run a Pipeline and Handle Escalations

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/escalation.mdx

Execute a pipeline and check the result status. If escalated, print the human-readable escalation details.

```python
result = await pipeline.run(order)
if result.status == "escalated":
    print(result.escalations[-1].legible())
```

--------------------------------

### Phase 1: Run Pipeline Until Human Approval Gate

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/human-approval.mdx

Execute the pipeline up to a gated step. The run halts, checkpointing completed steps, and produces an `ApprovalRequest` object.

```python
result = await pipeline.run(
    initial,
    run_id="order-A-1",
    checkpoints=store,
)
# result.status == "halted"
# result.halt_code == "requires_human_approval"
# result.approval_request is the typed request to hand to a reviewer
request = result.approval_request
print(request.question)         # readable description of what is being asked
print(request.package_legible)  # the full escalation package, as legible text
```

--------------------------------

### Attach an Escalation Policy to a Pipeline

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/escalation.mdx

Configure a pipeline to escalate to Slack synchronously when a failure occurs. Use 'sync' mode for irreversible actions.

```python
from drawbore import Pipeline
from drawbore.escalation import EscalationPolicy

pipeline = Pipeline(
    name="payments",
    on_failure=EscalationPolicy(channel="slack", target="compliance_queue", mode="sync"),
)
```

--------------------------------

### Initializing Pipeline with Untrusted Data

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/taint-and-trust.mdx

Set the initial trust level of pipeline inputs to UNTRUSTED when dealing with external data sources like documents or emails.

```python
from drawbore.tools import TrustLabel

await pipeline.run(initial, initial_trust=TrustLabel.UNTRUSTED)
```

--------------------------------

### Use InMemoryAuditSink for Audit Trails

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/observability-audit.mdx

Demonstrates how to use the `InMemoryAuditSink` to capture and print the audit trace of a pipeline run. This sink is the default for tests and local runs.

```python
from drawbore.audit import InMemoryAuditSink

my_sink = InMemoryAuditSink()
result = await pipeline.run(transaction, audit=my_sink, tenant_id="acme")

print(result.audit_trace.legible())
# Run 'run-1' of pipeline 'remittance' v1.0.0 (tenant acme): completed.
# Steps completed: 4.
# Escalations: 0.
# Schema violations: 0.
# - step 0 'validate' v1.0.0: ok
# ...
```

--------------------------------

### Register a Tool

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/tools.mdx

Register a tool with the ToolRegistry. This makes the tool available for agents to use. Ensure the tool function is defined before registration.

```python
from drawbore.tools import ToolRegistry

registry = ToolRegistry()

async def read_transaction(args):
    return {"id": args["id"], "amount": 100.0, "currency": "USD"}

registry.register_tool("transactions_db.read_transaction", read_transaction)
```

--------------------------------

### Run a Drawbore pipeline in test mode

Source: https://github.com/daviesayo/drawbore/blob/main/docs/agents/testing-drawbore.mdx

Use `pipeline.test_mode` to mock models and tools for local pipeline testing. Assert on the result status and audit trace.

```python
async with pipeline.test_mode(
    mock_model_responses={"scorer": {"risk": "low"}},
    mock_tools={"transactions_db.read_transaction": {"amount": 100}},
) as tp:
    result = await tp.run(Initial(...))
    assert result.status == "completed"
    assert "low" in result.audit_trace.legible()
```

--------------------------------

### Fan-in Topology - Problematic Binding

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/pipelines.mdx

Illustrates a potential issue in fan-in topologies where omitting explicit bindings can lead to predecessor-output mode delivering incorrect data, causing schema violations.

```python
pipeline.add(router)
pipeline.add(branch_a, inputs={"x": From("router.x")})
pipeline.add(branch_b)          # no inputs= — looks like it reads from router
pipeline.add(Join("merged", sources=["branch_a", "branch_b"],
                  policy="all_present", output=MergedResult,
                  inputs={"a_out": From("branch_a.result"),
                          "b_out": From("branch_b.result")}))
```

--------------------------------

### Registering Tools with Taint Information

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/taint-and-trust.mdx

Declare tools with their exfiltration capability and source trust level. 'untrusted-source' is the default for tools not explicitly marked.

```python
from drawbore.tools import ToolRegistry

reg = ToolRegistry()
reg.register_mcp_tool("docs:fetch", handler)                           # untrusted-source by default
reg.register_tool("http_post", handler, exfil_capable=True)            # a sink
```

--------------------------------

### Define a Tiny Pipeline with Pydantic Models

Source: https://github.com/daviesayo/drawbore/blob/main/docs/quickstart.mdx

Define input and output models using Pydantic, then create a Drawbore agent and pipeline. This sets up a basic pipeline with a single agent for data normalization.

```python
from pydantic import BaseModel
from drawbore import agent, Pipeline

class Payment(BaseModel):
    cents: int

class Normalized(BaseModel):
    dollars: float

@agent(name="normalize", input=Payment, output=Normalized)
async def normalize(p: Payment) -> Normalized:
    return Normalized(dollars=p.cents / 100)

pipeline = Pipeline(name="payments", version="1.0.0")
pipeline.add(normalize)
```

--------------------------------

### Enable Native Structured Output

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/production-llm-gateway.mdx

Configure a provider to use its native JSON schema decoding capabilities. This can significantly reduce reprompts by ensuring the model returns schema-valid JSON directly. Set `native_structured_output=True` on the `ProviderConfig`.

```python
from drawbore.llm import ProviderConfig

ProviderConfig(
    credential_env="OPENAI_API_KEY",
    native_structured_output=True,   # auto-wires each agent's output schema
)
```

--------------------------------

### Run Live OpenAI Smoke Test

Source: https://github.com/daviesayo/drawbore/blob/main/examples/live_openai_smoke/README.md

Execute the pytest for the live OpenAI smoke test to verify the provider wiring. This test checks the integration without necessarily validating the model's output content.

```bash
python3.11 -m pytest examples/live_openai_smoke/test_live_openai_smoke.py
```

--------------------------------

### Verify Confinement Receipt with Proxy Log

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/confinement-receipt.mdx

This snippet shows how to verify a confinement receipt by providing the proxy log. This binds the receipt to the actual execution log, ensuring that the log hasn't been tampered with and matches the receipt's run ID.

```python
verdict = verify(receipt, result.proxy_log)
print(verdict.status)          # re-derives the verdict from the live log
```

--------------------------------

### Accessing Evidence Compression Details

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/evidence-compression.mdx

After a pipeline run with evidence compression enabled, you can inspect the `evidence` attribute of the step records to see details about the compression, including token counts before and after, and the handle ID.

```python
print(result.audit_trace.step_records[0].evidence)
# evidence compressed (compressed for model); via json_rows; 4200->950 tokens; handle 8f3c...
```

--------------------------------

### Run Pipeline Locally with Mocked Dependencies

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/local-testing.mdx

Use `pipeline.test_mode` to provide mocked responses for tools, models, and loop scripts. The `async with` block ensures mocks are cleaned up after the test. The `result.audit_trace` provides production-shaped audit data for assertions.

```python
from drawbore.testing import call, final

async with pipeline.test_mode(
    mock_tools={"transactions_db.read_transaction": {"amount": 100}},
    mock_model_responses={"risk_scorer": {"risk": "low", "confidence": 0.94}},
    mock_loop_scripts={"resolver": [call("lookup", {"id": "t1"}), final({"answer": "ok"})]}
) as test:
    result = await test.run(TransactionInput(transaction_id="test_123"))

assert result.status == "completed"
audit = result.audit_trace            # production-shaped audit, your primary assertion surface
```

--------------------------------

### Registering MCP Server with Taint Information

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/taint-and-trust.mdx

Configure MCP servers to protect against untrusted data exfiltration by declaring exfil_capable and source_trust facts for their tools.

```python
from drawbore.tools import TrustLabel

await register_mcp_server(
    reg, name="slack", url="https://slack.com/mcp",
    allowed_tools=["send_message", "read_status"], client=client,
    exfil_capable={"send_message": True},              # an MCP sink
    source_trust={"read_status": TrustLabel.TRUSTED},  # an internal read you trust
)
```

--------------------------------

### Resume a Pipeline Run with InMemoryCheckpointStore

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/durable-resume.mdx

Use `InMemoryCheckpointStore` to resume a pipeline within the same process. The first run halts at step 3, and the second run with the same `run_id` resumes from that point.

```python
from drawbore.state import InMemoryCheckpointStore

store = InMemoryCheckpointStore()
first = await pipeline.run(order, run_id="order-A-1", checkpoints=store)   # halts at step 3
again = await pipeline.run(order, run_id="order-A-1", checkpoints=store)  # resumes at step 3
```

--------------------------------

### Run Number Chain Pipeline

Source: https://github.com/daviesayo/drawbore/blob/main/docs/examples/number-chain.mdx

Executes the number chain pipeline using Python 3.11, setting the PYTHONPATH and providing an OpenAI API key.

```bash
PYTHONPATH=src OPENAI_API_KEY=sk-... python3.11 examples/number_chain/pipeline.py
```

--------------------------------

### Accessing Metrics from Pipeline Run

Source: https://github.com/daviesayo/drawbore/blob/main/docs/guide/observability-audit.mdx

Shows how to access and print step durations, token usage, costs, and tool call details from the `result.metrics` object after a pipeline run. The metrics can also be exported as a dictionary.

```python
result = await pipeline.run(transaction)

for step in result.metrics.steps:
    print(step.agent, step.duration_seconds, step.tokens, step.cost)
    # 'risk_scorer' 0.42 TokenUsage(input_tokens=812, output_tokens=64, total_tokens=876) 0.0013

for call in result.metrics.tool_calls:
    print(call.tool, call.operation, call.duration_seconds, call.result)
    # 'mcp://identity/verify' invoke 0.08 ok

blob = result.metrics.to_dict()   # JSON-safe primitives for export
```

--------------------------------

### Define Pipeline Steps

Source: https://github.com/daviesayo/drawbore/blob/main/docs/examples/number-chain.mdx

Adds three sequential steps to a Drawbore pipeline: draw_number, inspect_number, and explain_result.

```python
pipeline = Pipeline(name="number_chain", version="1.0.0")
pipeline.add(draw_number)
pipeline.add(inspect_number)
pipeline.add(explain_result)
```