Contractual runtime
for AI agents

Installation

Requirements

Install

bash
# Install from PyPI
pip install deed-runtime

# From source
git clone https://github.com/Deadly-Reiter/deed
cd deed && pip install -e .

Quickstart

From zero to a running contractual agent in under 2 minutes.

bash
pip install deed-runtime

deeddeeds/agents/score_agent.dd
agent score_agent
  capabilities ["score_company"]

  contract score_contract
    pre  company_name
    post score_generated

  policy
    cap   budget_tokens <= 5000
    allow score_company if company_name

deeddeeds/pipelines/my_pipeline.dd
pipeline my_pipeline
  stage score
    agent score_agent
    -> score_company()
    checkpoint after
    on_error retry

jsoninput/state.json
{
  "company_name": "Acme AI",
  "budget_tokens": 3000
}

pythonrun.py
from deed import DeedRuntime

def score_company(ctx: dict) -> dict:
    ctx["score"]           = 0.82
    ctx["score_generated"] = True
    return ctx

runtime = DeedRuntime(".", initial_credits=100)
runtime.register("score_company", score_company)

result = runtime.run("my_pipeline", "input/state.json")
print(result.get("score"))       # 0.82
print(result.get("run_status")) # completed

Core Concepts

The Contract Principle

DEED doesn't trust LLMs to self-regulate. Every deed (atomic action) must satisfy a contract before and after execution. If either check fails, the runtime rejects the action — before it hits your API, database, or wallet.

Execution Lifecycle

Key Primitives

Use Case — Sales Intelligence Agent

A three-stage pipeline that enriches a company profile, scores ICP fit, and generates a personalized outreach brief. Each stage runs a dedicated specialist agent with its own contract.

deedpipelines/sales_intelligence.dd
pipeline sales_intelligence
  description "End-to-end sales intelligence workflow"
  input company_profile

  stage enrich
    agent data_agent
    -> enrich_company()
    checkpoint after
    on_error retry

  stage score
    agent score_agent
    -> score_company()
    checkpoint after
    on_error retry

  stage brief
    agent brief_agent
    -> generate_brief()
    -> persist_result()
    on_error deadletter

  observe
    trace true
    latency warn

deedagents/data_agent.dd
agent data_agent
  description "Enriches company profile with public data"
  capabilities ["enrich_company"]

  policy
    cap budget_tokens <= 5000
    deny enrich_company if region == "restricted"

  contract enrich_contract
    pre  company_name_present
    post enriched

agent score_agent
  description "Scores ICP fit 0.0 – 1.0"
  capabilities ["score_company"]

  policy
    cap budget_tokens <= 5000
    allow score_company if enriched

  contract score_contract
    pre  enriched
    post score_generated

agent brief_agent
  description "Generates personalized outreach brief"
  capabilities ["generate_brief", "persist_result"]

  policy
    cap budget_tokens <= 4000
    allow generate_brief if scored

  contract brief_contract
    pre  scored
    post brief_generated

pythonrun.py
from deed import DeedRuntime
from executors.sales_ops import enrich_company, score_company, generate_brief, persist_result

runtime = DeedRuntime(".", initial_credits=100)
(runtime
    .register("enrich_company",  enrich_company,  charge=1)
    .register("score_company",   score_company,   charge=1)
    .register("generate_brief",  generate_brief,  charge=1)
    .register("persist_result",  persist_result,  charge=1)
)

result = runtime.run("sales_intelligence", "input/company.json")

print(f"Score:   {result['score']}")
print(f"Brief:   {result['brief']}")
print(f"Credits: {runtime.x402.spent} used")

output
[DEED] stage: enrich
[DEED] charged: 1 credit | balance: 99
[DEED] checkpoint: enrich::enrich_company

[DEED] stage: score
[DEED] charged: 1 credit | balance: 98
[DEED] checkpoint: score::score_company

[DEED] stage: brief
[DEED] charged: 1 credit | balance: 97
[DEED] checkpoint: brief::generate_brief
[DEED] checkpoint: brief::persist_result

Score:   1.0
Priority: high
Brief:   Acme AI GmbH is a B2B SaaS company (Series B).
         ICP score: 1.0 (high priority). Tech stack: Salesforce, HubSpot, Slack.
         Recommended next action: research decision makers.
Credits: 4 used

Use Case — Data Pipeline Agent

An extraction-transformation-load pipeline where each stage has explicit data contracts. If the schema contract fails at any stage, the pipeline halts before bad data propagates downstream.

deedpipelines/data_etl.dd
pipeline data_etl
  description "Schema-safe extraction and transformation"
  input raw_source

  stage extract
    agent extractor_agent
    -> fetch_records()
    checkpoint after
    on_error retry

  stage validate
    agent validator_agent
    -> check_schema()
    checkpoint after
    on_error deadletter  # halt — don't load bad data

  stage transform
    agent transform_agent
    -> normalize_fields()
    -> enrich_metadata()
    checkpoint after
    on_error retry

  stage load
    agent loader_agent
    -> write_to_warehouse()
    checkpoint after
    on_error deadletter

deeddeeds/extract.dd
agent extractor_agent
  capabilities ["fetch_records"]

  contract extract_contract
    pre  source_url_present
    post records_fetched

  policy
    cap record_count <= 100000
    deny fetch_records if source_type == "pii_unmasked"

agent validator_agent
  capabilities ["check_schema"]

  contract validate_contract
    pre  records_fetched
    post schema_valid

agent loader_agent
  capabilities ["write_to_warehouse"]

  deed write_to_warehouse
    idempotency_key batch_id  # safe to retry
    retry 3

  contract load_contract
    pre  schema_valid and records_transformed
    post data_persisted

pythonrun.py
from deed import DeedRuntime
from executors.etl_ops import fetch_records, check_schema, normalize_fields, enrich_metadata, write_to_warehouse

runtime = DeedRuntime(".", initial_credits=500)
(runtime
    .register("fetch_records",    fetch_records,    charge=1, retry=3)
    .register("check_schema",     check_schema,     charge=1)
    .register("normalize_fields", normalize_fields, charge=1, retry=2)
    .register("enrich_metadata",  enrich_metadata,  charge=1, retry=2)
    .register("write_to_warehouse", write_to_warehouse, charge=1,
              idempotency_key="batch_id")
)

# First run
result = runtime.run("data_etl", "input/source.json")

# Failure mid-run? Resume from last checkpoint
runtime2 = DeedRuntime(".", replay=True, initial_credits=500)
runtime2.register("fetch_records", fetch_records)
# ... same registrations ...
result = runtime2.run("data_etl", "input/source.json")
# → Skips fetch, validate, transform
# → Resumes at load — no duplicates

Use Case — API Orchestrator

Coordinates multiple third-party API calls with rate limiting, budget caps per provider, and automatic retry. Useful for agents that fan out to multiple services (search, enrichment, notification).

deedagents/api_orchestrator.dd
agent api_orchestrator
  description "Coordinates external API calls with budget control"
  capabilities ["search_web", "enrich_linkedin", "send_slack_alert", "write_crm"]

  policy
    cap api_calls       <= 50
    cap budget_tokens   <= 5000
    deny send_slack_alert if score < 0.6
    deny write_crm        if not data_validated

  contract orchestrator_contract
    pre  query_present
    post result_written or alert_sent

  observe
    trace   true
    explain true

deeddeeds/api_calls.dd
deed search_web
  retry           3
  idempotency_key query_hash

deed enrich_linkedin
  retry           2
  idempotency_key profile_url

deed send_slack_alert
  retry           1
  idempotency_key alert_id
  # blocked by policy if score < 0.6

deed write_crm
  retry           3
  idempotency_key company_id

pythonrun.py
from deed import DeedRuntime, ContractViolation, PolicyViolation
from executors.api_ops import search_web, enrich_linkedin, send_slack_alert, write_crm

runtime = DeedRuntime(".", initial_credits=200)
(runtime
    .register("search_web",       search_web,       charge=1, retry=3)
    .register("enrich_linkedin",  enrich_linkedin,  charge=2, retry=2)
    .register("send_slack_alert", send_slack_alert, charge=1)
    .register("write_crm",        write_crm,        charge=1, retry=3)
)

try:
    result = runtime.run("api_orchestrator", "input/query.json")
except PolicyViolation as e:
    print(f"Blocked by policy: {e}")
    # e.g. "score 0.4 — send_slack_alert denied"
except ContractViolation as e:
    print(f"Contract failed: {e}")
    # Checkpoint saved — replay when ready

Syntax Reference — agent

An agent block declares a named agent with its capabilities, policy constraints, and contracts.

deed
agent <name>
  description "..."                # optional, human-readable
  capabilities ["action1", "action2"]   # actions this agent can perform

  policy
    ...                              # enforcement rules

  contract <name>
    ...                              # pre/post conditions

  observe
    ...                              # tracing / explain

Fields

observe block

deed
observe
  trace   true    # emit checkpoint events
  explain true    # LLM explains each decision (verbose)
  latency warn    # warn if stage exceeds threshold

Syntax Reference — pipeline

A pipeline defines an ordered sequence of stages. Each stage maps to a specific agent and a set of actions. DEED-Runtime executes stages in declaration order.

deed
pipeline <name>
  description "..."
  input <context_key>

  stage <name>
    agent     <agent_name>
    -> action1()
    -> action2()
    checkpoint after
    on_error   retry | deadletter

  observe
    trace true

Stage fields

Syntax Reference — deed

A deed is the atomic unit of action. It wraps a single operation with retry semantics and idempotency guarantees.

deed
deed <name>
  retry           3                   # attempts before DLQ
  idempotency_key <context_key>       # safe to re-run

Fields

Syntax Reference — contract

Contracts define what must be true before and after a deed or agent execution. DEED-Kernel enforces both. Violations halt execution before any side effects occur.

deed
contract <name>
  pre  <expression>     # checked before execution
  post <expression>     # checked after execution

Pre-condition

Evaluated against the current context before the action runs. If false: ContractViolation is raised, no credits are charged, no side effects occur.

Post-condition

Evaluated against the updated context after the action runs. If false: ContractViolation, action is rolled back, credits refunded.

Examples

deed
// Named predicate
contract c1
  pre  company_name_present
  post score_generated

# Comparison expression
contract c2
  pre  input.balance > 0
  post output.status == "success"

# Compound condition
contract c3
  pre  enriched and score_generated
  post brief_generated or result_persisted

Built-in named predicates

Syntax Reference — policy

Policy rules are evaluated before the LLM receives control. This makes jailbreak and unauthorized tool use structurally impossible — the gate closes before any generation happens.

deed
policy
  cap   <field>  <= | >= | ==  <value>          # resource limit
  allow <action> if <condition>                  # allow if condition true
  deny  <action> if <condition>                  # deny if condition true

Rule types

Evaluation order

cap rules → deny rules → allow rules. An action is allowed if no deny matches and at least one allow matches (or no allow rules are defined for that action).

Examples

deed
policy
  # Hard resource caps
  cap budget_tokens <= 5000
  cap api_calls     <= 50

  # Conditional deny — geography restriction
  deny enrich_company if region == "restricted"

  # Conditional allow — require prerequisite
  allow generate_brief if scored
  allow write_crm       if data_validated and score >= 0.7

  # Unconditional deny
  deny delete_all if true

Syntax Reference — Expressions

Expressions are used in pre:, post:, and if clauses. They are evaluated against the runtime context dictionary.

Comparison operators

Logical operators

deed
// and — both must be true
pre: enriched and company_name_present

// or — at least one must be true
post: result_persisted or alert_sent

// not — negation
pre: not data_validated

Context path resolution

Dotted paths like input.region or output.status resolve against the context dict. The input prefix is optional and stripped automatically.

deed
// These are equivalent:
pre: input.balance > 0
pre: balance > 0

Runtime — Execution Lifecycle

Every deed in DEED passes through a deterministic five-phase cycle. No phase can be skipped.

State machine guarantees

Runtime — DEED Credits

DEED treats compute cost as a first-class object. Every deed execution costs 1 DEED Credit by default (configurable via charge in register()). Budget enforcement happens at the policy layer — the agent is stopped before it overspends.

How charging works

python
from deed import DeedRuntime

runtime = DeedRuntime(
    project_root=".",
    initial_credits=1000,  # starting balance
)

runtime.register("score_company", score_fn, charge=1)
runtime.register("send_alert",    alert_fn, charge=2)  # costs 2 credits

result = runtime.run("my_pipeline", "input/state.json")

print(runtime.x402.credits)     # remaining balance
print(runtime.x402.spent)       # total spent this run

Runtime — Checkpoint & Replay

Every successful deed writes a checkpoint: the step identifier and full context state. On failure, pass replay=True to resume from the last checkpoint. Completed steps are skipped — no duplicate charges, no duplicate writes.

Checkpoint format

jsonoutput/checkpoint.json
{
  "completed_steps": [
    "enrich::enrich_company",
    "score::score_company"
  ],
  "context": {
    "company_name": "Acme AI GmbH",
    "score": 0.82,
    "score_generated": true,
    "run_status": "running"
  }
}

Replay

python
# Normal run
runtime = DeedRuntime(".", initial_credits=100)
runtime.register("score_company", score_fn, charge=1)
result = runtime.run("my_pipeline", "input/state.json")

# Failure at stage 3? Resume:
runtime2 = DeedRuntime(".", replay=True, initial_credits=100)
runtime2.register("score_company", score_fn, charge=1)
result = runtime2.run("my_pipeline", "input/state.json")
# → [DEED] replay skip: enrich::enrich_company
# → [DEED] replay skip: score::score_company
# → [DEED] stage: brief  ← continues here

Runtime — Error Handling & DLQ

When a deed fails after exhausting retries, it enters the Dead Letter Queue. DEED-Runtime can either halt execution or continue with subsequent stages depending on the on_error setting.

on_error modes

DLQ entry format

jsonoutput/deed.dlq
{
  "step":    "validate::check_schema",
  "ts":      "2026-05-05T04:22:11Z",
  "error":   "ContractViolation: post: schema_valid not satisfied",
  "context": { "...": "last known state" },
  "retries": 3
}

Recovery flow