mycelium-runtime 1.3.4

Runtime guards for AI agents: stale context, bad tool calls, duplicate side effects on retry

Mycelium runtime

Current package: mycelium-runtime v1.3.4 (transition envelope).

One painful bug → a few lines of config

LangGraph Cloud redispatches a long tool call while the first is still running. Both complete. You pay twice. Side effects run twice. langgraph#7417

pip install mycelium-runtime   # Python 3.10+
mycelium init                  # on-ramp scaffold (transition + one ledgered tool)
mycelium init --full           # reference scaffold (all guards; fill TODOs)
mycelium demo                  # see the bug and the fix (no LangGraph required)

from mycelium import load_config

config = load_config("mycelium.yaml")  # includes transition: + side_effect_class

@config.apply
def subagent_task(task: str) -> dict:
    return run_slow_subagent(task)

# Pass tool_call_id from LangGraph; redispatch resolves the existing transition
subagent_task(task="analyze_market", tool_call_id=call["id"])

What else it does

Problem	What Mycelium does
Stale or broken context	TTL cache, message repair, history limits; agent sees fresh, valid data
Bad or unauthorized tool calls	Validate inputs/outputs, allowlists, scoped paths; block before execution
Duplicate side effects on retry	v1.3 transition envelope (side-effect class, poll / hard-block), ledgers, state flush, signed receipts

Framework-agnostic. Raw message lists and plain Python functions (LangGraph, CrewAI, OpenAI tool loops, etc.).

Install

Requires Python 3.10+ (3.11+ recommended).

pip install mycelium-runtime
mycelium init              # on-ramp: duplicate-tool fix → ./mycelium.yaml
mycelium init --full       # reference: every guard section (not the default)
mycelium init --minimal    # smaller multi-guard scaffold
mycelium demo              # terminal demo of langgraph#7417

Quickstart: stale context & broken transcripts

from mycelium import protect, Session

@protect(entity_param="customer_id", ttl=60)
async def fetch_customer(customer_id: str) -> dict:
    return await db.get(customer_id)

async def handle_request(customer_id: str):
    async with Session():
        return await fetch_customer(customer_id=customer_id)

Sync tools (CrewAI, Smolagents):

from mycelium import protect_sync, Session

@protect_sync(entity_param="customer_id", ttl=60)
def fetch_customer(customer_id: str) -> dict:
    return db.get(customer_id)

with Session():
    customer = fetch_customer(customer_id="c1")

What @protect / protect_sync / Session do

• @protect / protect_sync: TTL cache with per-entity keys; auto-refetch when stale; clear on error
• Session: one cache per agent run; use in production to prevent cross-request leakage

MessageValidator

Run before each LLM call to catch broken transcripts:

from mycelium import MessageValidator

messages = MessageValidator().repair(messages)  # auto-fix what it can
# or
messages = MessageValidator().validate(messages)  # raise on first issue

Catches orphan tool results, duplicate tool-call IDs, invalid roles, and related serialization bugs.

HistoryGuard

Run before each LLM call to catch oversized or corrupted history:

from mycelium import HistoryGuard

guard = HistoryGuard(max_tokens=100_000)
messages = guard.validate(messages)
guard.check_for_drops(processed_messages)  # after framework trimming

Raises on token overflow, message count limits, duplicate turns, and silent message drops.

Quickstart: tool boundaries

from mycelium import bounded, ToolRegistry, ToolRunner

FETCH_CUSTOMER_SCHEMA = {
    "customer_id": {"type": "string", "required": True, "pattern": r"^c\d+$"},
}

CUSTOMER_RECORD_SCHEMA = {
    "customer_id": {"type": "string", "required": True},
    "name": {"type": "string", "required": True},
}

registry = ToolRegistry(allowed=["fetch_customer"])

@registry.register
@bounded(
    schema=FETCH_CUSTOMER_SCHEMA,
    output_schema=CUSTOMER_RECORD_SCHEMA,
    allowed_paths=["/workspace/src/"],
)
async def fetch_customer(customer_id: str) -> dict:
    return await db.get(customer_id)

runner = ToolRunner(registry=registry)
result = await runner.call(fetch_customer, customer_id="c1")

Sync tools:

from mycelium import bounded_sync

@bounded_sync(schema=FETCH_CUSTOMER_SCHEMA)
def fetch_customer(customer_id: str) -> dict:
    return db.get(customer_id)

Field spec keys: type (string, integer, number, boolean), required, pattern, min_length, max_length. You pass plain dicts; Mycelium validates internally; no Pydantic imports in your code.

What @bounded / bounded_sync do

• @bounded / bounded_sync: validate tool args against your field spec before the function runs
• output_schema: validate the return value after the function runs; bad results are not propagated
• allowed_paths / entity_pattern: user-defined scope gates (path prefixes, entity ID format)
• On failure, raises ToolBoundaryError with llm_message for the agent loop; does not retry by itself

ToolRegistry

Run before dispatch to enforce which tools this agent may call:

from mycelium import ToolRegistry

registry = ToolRegistry(allowed=["search_docs", "summarize"])
registry.validate_call("fetch_customer")  # raises ToolBoundaryError

Blocks calls to tools outside the developer-defined allowlist.

ToolRunner

Run around @bounded tools when you want automatic retries:

from mycelium import ToolRunner

runner = ToolRunner(registry=registry, max_llm_retries=2, max_tool_retries=3)

result, messages = await runner.run_with_llm_retry(
    fetch_customer,
    messages=messages,
    tool_call_id="call_1",
    kwargs={"customer_id": "c1"},
    invoke_llm=llm.ainvoke,
    parse_tool_kwargs=extract_tool_args,
)

• Input, allowlist, and scope failures → append tool error to messages → LLM retry
• Output failures → retry the tool up to max_tool_retries → then LLM retry
• Raises ToolBoundaryExhaustedError when retries are used up

Quickstart: idempotency & audit receipts (v1.3 transition envelope)

Stop duplicate payments, emails, and API calls when the framework retries. v1.3.3 uses five effect-semantic side_effect_class values: reads poll in-flight duplicates; mutating tools hard-block ambiguous states instead of blind re-execute.

Tool-level idempotency

from mycelium import ledger_sync
from mycelium.transition import SideEffectClass, ToolTransitionBinding

binding = ToolTransitionBinding.for_tool(
    agent_id="payment-agent",
    policy_version="2026.07.1",
    side_effect_class=SideEffectClass.KEYED_MUTATE,
)

@ledger_sync(transition_binding=binding)
def send_payment(amount: float, recipient: str) -> dict:
    return gateway.charge(amount, recipient)

# Same logical call executes only once.
send_payment(amount=100.0, recipient="acct_123", tool_call_id="call_abc")
send_payment(amount=100.0, recipient="acct_123", tool_call_id="call_abc")

Or wire from YAML (recommended):

transition:
  agent_id: payment-agent
  policy_version: "2026.07.1"
  lease_ttl: 3600

action_ledger:
  storage: file
  path: ./mycelium-ledger.json
  tools: [send_payment, search_docs]

tools:
  send_payment:
    side_effect_class: keyed_mutate
    retry_permission: manual_reconciliation_required
  search_docs:
    side_effect_class: read

Async tools:

from mycelium import ledger

@ledger()
async def send_payment(amount: float, recipient: str) -> dict:
    return await gateway.charge(amount, recipient)

What @ledger / ledger_sync do

• Record every tool invocation in a durable ActionLedger
• Deduplicate retries and redispatches via a rich transition key (scope + tool + args + side_effect_class + policy), not only tool_call_id
• read tools: poll in-flight, reclaim expired leases, retry failed-before-effect
• Mutating tools: return completed results, poll in-flight, hard-block ambiguous states (LedgerHardBlockError)
• Persist failed attempts with terminal outcomes (FAILED_BEFORE_EFFECT, FAILED_AFTER_EFFECT, etc.) for audit and reconciliation

Side-effect classes

Class	Typical use	Duplicate handling
read	search, fetch	poll / reclaim / retry
idempotent_mutate	upsert / set status	retry if boundary not crossed
keyed_mutate	Stripe-style create/charge	retry only with same provider key
non_idempotent_mutate	send email, spawn subagent	hard-block on ambiguity
irreversible	wire / on-chain burn	hard-block → human

Legacy aliases (read_only, payment, subagent, …) still parse. Set per tool in YAML with side_effect_class. Required when transition: is configured and the tool is ledgered.

Storage backends:

Backend	Use case	YAML storage
memory	Single process, tests	memory (default)
file	Local dev, single host (fcntl lock)	file + path
redis	Multi-worker, in-flight TTL	redis + url or url_env
postgres	Audit/compliance, durable SQL	postgres + dsn or dsn_env

from mycelium import ActionLedger, FileLedgerStorage, InMemoryLedgerStorage
from mycelium import RedisLedgerStorage, PostgresLedgerStorage

ledger = ActionLedger(storage=InMemoryLedgerStorage())
ledger = ActionLedger(storage=FileLedgerStorage("./mycelium-ledger.json"))
ledger = ActionLedger(storage=RedisLedgerStorage("redis://localhost:6379/0"))
ledger = ActionLedger(storage=PostgresLedgerStorage("postgresql://localhost/mycelium"))

Optional extras: pip install 'mycelium-runtime[redis]' or pip install 'mycelium-runtime[postgres]'.

Quickstart: task-level idempotency

Stop entire tasks from re-running on framework-level retries:

from mycelium import task_ledger_sync

@task_ledger_sync()
def process_invoice(invoice_id: str) -> dict:
    customer = fetch_customer(customer_id=...)
    payment = send_payment(...)
    return {"invoice_id": invoice_id, "status": "paid"}

# Framework retries the task with the same task_id
process_invoice(invoice_id="inv-42", task_id="invoice-42")  # executes
process_invoice(invoice_id="inv-42", task_id="invoice-42")  # returns stored result

Use id_from to derive the task id from business keys automatically:

@task_ledger_sync(id_from=["invoice_id"])
def process_invoice(invoice_id: str, amount: float) -> dict:
    ...

# Both calls map to the same task id because invoice_id is the same.
process_invoice(invoice_id="inv-42", amount=100.0)
process_invoice(invoice_id="inv-42", amount=200.0)  # returns first result

Correction retries

If a completed task produced a bad result and the LLM/agent needs to re-attempt it, use a new task id. The framework will normally generate fresh tool call ids for the new attempt, so the task re-executes cleanly.

r1 = process_invoice(invoice_id="inv-42", task_id="invoice-42-attempt-1")  # bad result
r2 = process_invoice(invoice_id="inv-42", task_id="invoice-42-attempt-2")  # fresh attempt

YAML configuration

Separate YAML sections per guard type. Global ledger settings inherit into tools/tasks so you do not repeat storage paths on every function.

Minimum integration (3 steps):

# mycelium.yaml: global sections (configure once)
transition:
  agent_id: payment-agent
  policy_version: "2026.07.1"
  lease_ttl: 3600

action_ledger:
  storage: file
  path: ./mycelium-ledger.json
  tools: [send_payment, search_docs]

task_ledger:
  storage: file
  path: ./mycelium-task-ledger.json
  tasks: [process_invoice]

state_flush:
  storage: file
  path: ./mycelium-state.json

audit_receipt:
  signing_key_env: MYCELIUM_SIGNING_KEY
  storage: file
  path: ./mycelium-receipts.jsonl

# Per-tool: side_effect_class + schemas
tools:
  fetch_customer:
    side_effect_class: read
    protect: {entity_param: customer_id, ttl: 60}
    bounded:
      schema:
        customer_id: {type: string, required: true, pattern: "^c\\d+$"}

  send_payment:
    side_effect_class: keyed_mutate
    bounded:
      schema:
        amount: {type: number, required: true}
        recipient: {type: string, required: true}

  search_docs:
    side_effect_class: read

tasks:
  process_invoice:
    ledger: true
    id_from: [invoice_id]

registry:
  auto: true                     # allowlist = all configured tools

history_guard:
  max_tokens: 100000

message_validator:
  enabled: true

from mycelium import load_config
import my_tools

config = load_config("mycelium.yaml")
tools = config.instrument(my_tools)   # one call wraps tools + tasks

with config.run(thread_id):
    messages = config.prepare_messages(messages)  # message validation + state flush
    ...

ledger: true inherits from action_ledger / task_ledger. When audit_receipt is configured with auto: true (default), all ledgered tools/tasks get signed receipts automatically. Set transition.agent_id for receipt identity (replaces audit_receipt.agent_id from v1.2).

Configs without transition: keep v1.2 ledger behavior. See CHANGELOG for breaking changes.

Legacy per-tool style still works. Start with mycelium init; use mycelium init --full for the all-guards reference template.

---

For contributors (repo layout)

Clone the GitHub repo to run proofs and tests. PyPI installs only the mycelium package.

git clone https://github.com/mycelium-labs/mycelium.git
cd mycelium/sdk && pip install -e ".[dev]"
pytest tests/ -v