deed-runtime 0.1.0

Contractual runtime for AI agents — pre/post contracts, policy enforcement, credits metering, checkpoint & replay

DEED

Contractual runtime for AI agents.

pip install deed-runtime

Zero dependencies. Python 3.10+.

---

The problem

Your AI agent ran. Something unexpected happened. No pre-conditions checked. No audit trail. No rollback.

The answer

Every action passes through a contract:

agent score_agent
  contract score_contract
    pre  company_name_present
    post score_generated

  policy
    cap  budget_tokens <= 5000
    deny score_company if input.region == "restricted"
    allow score_company if company_name_present

• Pre-condition fails → action rejected, no side effects, no charge
• Post-condition fails → DLQ entry written, credits refunded
• Policy violation → blocked before execution

Five lines

from deed import DeedRuntime

runtime = DeedRuntime(".")
runtime.register("score_company", my_scorer, charge=1)
result = runtime.run("my_pipeline", "input/state.json")
print(result["score"])  # 0.82

What you get

• Pre/post contracts on every agent action
• Policy engine (cap / allow / deny) enforced before execution
• Credits metering — charge per deed, refund on violation
• Checkpoint & replay — resume from last known good state
• DLQ — failed stages captured with full context

Three working examples

# Clone and run:
git clone https://github.com/syntriq/deed
cd deed/examples/mushroom_safety
python run.py           # happy path
python run.py --fail    # ContractViolation → DLQ

• examples/mushroom_safety/ — safety-critical triage
• examples/orchid_rescue/ — conservation scoring
• examples/sales_agent/ — ICP scoring pipeline

How register() works

runtime = DeedRuntime(
    project_root=".",           # must contain deeds/ folder with .dd files
    initial_credits=100,        # starting credit balance
    named_predicates_fn=my_fn,  # optional: callable(context) -> dict[str, bool]
)

# Chain register calls
(runtime
    .register("normalize",   normalize_fn,  charge=1)
    .register("score",       score_fn,      charge=1, retry=2)
    .register("send_alert",  alert_fn,      charge=1, idempotency_key="alert_id")
)

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

Writing .dd files

agent my_agent
  description "Does something useful"
  capabilities ["action_a", "action_b"]

  contract my_contract
    pre  input_present
    post result_generated

  policy
    cap  budget_tokens <= 3000
    deny action_b if region == "restricted"
    allow action_a if input_present
    allow action_b if input_present and not region == "restricted"

pipeline my_pipeline
  input my_input

  stage step_one
    agent my_agent
    -> action_a()
    checkpoint after
    on_error retry(2)

  stage step_two
    agent my_agent
    -> action_b()
    on_error deadletter

Error types

from deed import ContractViolation, PolicyViolation, DeedError

try:
    result = runtime.run("my_pipeline", "input/state.json")
except ContractViolation as e:
    # pre or post condition failed
    # action was NOT executed (pre) or ran but outcome invalid (post)
    print(e)
except PolicyViolation as e:
    # cap / deny / allow rule blocked the action
    # action was NOT executed, no credits charged
    print(e)

Replay

# After a failure — resume from last checkpoint
runtime = DeedRuntime(".", replay=True, initial_credits=100, ...)
runtime.register(...)  # same registrations
result = runtime.run("my_pipeline", "input/state.json")
# Completed steps are skipped, credits not re-charged

Docs

docs.deed.dev

License

MIT — Syntriq