Musubito

Musubito records execution lineage so agentic LLM workflows can safely skip redundant expensive calls instead of recomputing the same DAG steps.

Why Musubito?

LLM-heavy pipelines often re-run the same expensive steps because the runtime has no durable memory of what was executed, which inputs were used, and which upstream results contributed to the output.

Musubito gives each execution node a deterministic identity derived from the operation name, operation implementation, canonical input hash, namespace, and sorted upstream node IDs. If the same logical node is reached again and its replay policy allows reuse, Musubito returns the stored artifact instead of executing the function again.

Lineage is stored locally in SQLite, so replay decisions are fast, deterministic, and inspectable without requiring a remote service.

Fan-in DAG patterns are first-class: musubito_merge() lets an aggregate step explicitly depend on multiple upstream MusubitoResult[T] values.

Install

pip install musubito

Real-World Examples

The examples below use real LLM SDK calls. Install the provider SDK you need (pip install openai or pip install anthropic) and set the matching API key in your environment before running them.

Single LLM call with permanent cache

import time

from openai import OpenAI

from musubito import StepConfiguration, StepType, musubito_step

client = OpenAI()


@musubito_step(
    semantics=StepConfiguration(step_type=StepType.DETERMINISTIC),
)
def explain_runtime(prompt: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        temperature=0,
        messages=[{"role": "user", "content": prompt}],
    )
    return response.choices[0].message.content or ""


prompt = "Explain deterministic replay for LLM research agents in five bullets."

start = time.perf_counter()
first = explain_runtime(prompt)
first_ms = (time.perf_counter() - start) * 1000

start = time.perf_counter()
second = explain_runtime(prompt)
second_ms = (time.perf_counter() - start) * 1000

# The second call saves one OpenAI API request for the same stable prompt.
print(first.value[:200])
print(second.value[:200])
print(f"first={first_ms:.1f} ms replay={second_ms:.1f} ms")

Expiring cache for fresh answers

from openai import OpenAI

from musubito import StepConfiguration, StepType, musubito_step

client = OpenAI()

fresh_hourly = StepConfiguration(
    step_type=StepType.STOCHASTIC,
    ttl_seconds=3600,
)


@musubito_step(semantics=fresh_hourly)
def market_brief(topic: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        temperature=0.4,
        messages=[
            {
                "role": "user",
                "content": f"Write a concise market-watch brief about {topic}.",
            }
        ],
    )
    return response.choices[0].message.content or ""


result = market_brief("AI infrastructure startups")

# STOCHASTIC + TTL saves repeat API calls for one hour, then refreshes naturally.
print(result.value)

Two-step pipeline with lineage

from anthropic import Anthropic

from musubito import MusubitoResult, StepConfiguration, StepType
from musubito import musubito_merge, musubito_step

client = Anthropic()


@musubito_step()
def extract_key_facts(text: str) -> str:
    message = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=300,
        messages=[{"role": "user", "content": f"Extract key facts:\n{text}"}],
    )
    return message.content[0].text


@musubito_step(
    semantics=StepConfiguration(step_type=StepType.STOCHASTIC, ttl_seconds=86400),
)
def write_social_summary(facts: MusubitoResult[str]) -> str:
    message = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=120,
        messages=[{"role": "user", "content": f"Write one tweet:\n{facts.value}"}],
    )
    return message.content[0].text


source_text = "Musubito records execution lineage for replayable agent steps."
facts = extract_key_facts(source_text)

with musubito_merge(facts):
    summary = write_social_summary(facts)

# Re-running saves the extraction call immediately; the summary refreshes after TTL.
print(summary.value)

Custom storage path for a project

from openai import OpenAI

from musubito import MusubitoEngine, SQLiteStorage
from musubito import musubito_step, use_musubito_engine

client = OpenAI()


@musubito_step()
def classify_note(note: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        temperature=0,
        messages=[
            {
                "role": "user",
                "content": f"Classify this research note in one label:\n{note}",
            }
        ],
    )
    return response.choices[0].message.content or ""


storage = SQLiteStorage(db_path=".musubito/project-alpha.db")
engine = MusubitoEngine(storage)

with storage, use_musubito_engine(engine):
    result = classify_note("GPU scheduling dominates the serving bottleneck.")

# A project-specific DB keeps replay separate across teams or experiments.
print(result.value)

When to Use Which StepType

Core Concepts

A node is one recorded execution of a decorated function. Its identity combines the operation name, a normalized fingerprint of the function implementation, the canonical hash of its inputs, the engine namespace, and the sorted set of upstream node IDs. The identity remains stable across sessions while those ingredients remain unchanged, and a code change cannot silently replay an artifact produced by the previous implementation.

Replay means Musubito returns a previously stored artifact instead of calling the decorated function again. Replay is allowed when the stored node is successful, not stale, not forced to re-execute, and any configured TTL has not expired.

StepType.DETERMINISTIC marks work that is safe to replay freely, such as pure transformations or deterministic parsers.

StepType.STOCHASTIC marks work that may produce different outputs, such as LLM calls. It can still be replayed intentionally, often with a TTL to bound how long the stored result remains valid.

StepType.EXTERNAL_EFFECT marks work with side effects, such as network calls or tool invocations. When replay is allowed, Musubito returns the stored artifact; it does not re-run the external effect. Use force_reexecution=True when the effect must be issued again.

An Artifact is the persisted output of a node. MusubitoResult[T] points to that artifact and carries both the current DAG node ID and the lineage producer selected for the result. On historical reuse, the producer identifies the node that originally created the value.

musubito_merge() declares an explicit multi-parent context. It is used when one step depends on multiple previous MusubitoResult[T] values, making fan-in DAG edges visible to the lineage engine.

Inputs and artifacts use a deterministic, type-preserving codec. Dictionaries require string keys; lists and tuples remain distinct. Unsupported values, cyclic containers, and non-finite floats fail explicitly instead of falling through to json.dumps().

Step Configuration

The default configuration is deterministic and requires no extra setup:

from musubito import musubito_step


@musubito_step()
def normalize(text: str) -> str:
    return text.strip().lower()

The implementation fingerprint changes when the decorated function changes. Use an explicit operation version when a dependency outside the function body also affects the result, such as a prompt template loaded from a file or a remotely configured model:

from musubito import musubito_step


@musubito_step(operation_version="prompt-v3")
def build_prompt(topic: str) -> str:
    return f"Summarize {topic}"

Source-based fingerprints are normalized across supported Python versions. Dynamically created callables whose source cannot be inspected use a conservative bytecode fallback; a Python interpreter upgrade may therefore give those callables a new identity.

For stochastic work, such as an LLM call, use a TTL when the cached answer should only be reused for a bounded time:

from musubito import StepConfiguration, StepType, musubito_step

llm_semantics = StepConfiguration(
    step_type=StepType.STOCHASTIC,
    ttl_seconds=3600,
)


@musubito_step(semantics=llm_semantics)
def draft_answer(prompt: str) -> str:
    return prompt.upper()

To force a stochastic step to run again, set force_reexecution=True:

from musubito import StepConfiguration, StepType, musubito_step

fresh_semantics = StepConfiguration(
    step_type=StepType.STOCHASTIC,
    force_reexecution=True,
)


@musubito_step(semantics=fresh_semantics)
def generate_fresh_answer(prompt: str) -> str:
    return prompt.upper()

For external effects, Musubito stores and returns the artifact when replay is allowed. The side effect itself is not repeated unless force_reexecution=True is used:

from musubito import StepConfiguration, StepType, musubito_step

external_semantics = StepConfiguration(
    step_type=StepType.EXTERNAL_EFFECT,
)


@musubito_step(semantics=external_semantics)
def call_external_tool(payload: dict[str, str]) -> dict[str, str]:
    return {"status": "recorded", "id": payload["id"]}

Reuse across different upstream parent sets is disabled by default. Advanced workflows may opt in with allow_cross_parent_reuse=True when equal explicit inputs are sufficient to establish semantic equivalence. This trades stricter lineage isolation for broader historical reuse and should be enabled per step, not globally.

Using a Custom Engine

Use use_musubito_engine() when you want explicit control over the storage path or engine instance:

from musubito import (
    MusubitoEngine,
    SQLiteStorage,
    musubito_step,
    use_musubito_engine,
)


@musubito_step()
def summarize_text(text: str) -> str:
    return text.upper()


with SQLiteStorage(db_path=".my_run/run.db") as storage:
    engine = MusubitoEngine(storage, namespace="experiment-a")

    with use_musubito_engine(engine):
        result = summarize_text("Musubito records deterministic lineage.")

print(result.value)

Namespaces isolate node identity while allowing several experiments to share one database. engine.storage and engine.namespace expose the active configuration as read-only properties.

MusubitoEngine accepts any backend that satisfies the public StorageBackend protocol. A backend must provide durable node and artifact operations together with the execution-claim methods used for single-flight coordination. The engine's claim_lease_seconds, claim_wait_timeout_seconds, and claim_poll_interval_seconds settings are intended for backend-specific tuning; their defaults suit local SQLite workloads.

For direct engine calls, explain() reports why work can or cannot be replayed without running user code. Its reasons distinguish exact hits, code or upstream changes, stale, failed, or running nodes, policy constraints, missing artifacts, and integrity failures:

from musubito import MusubitoEngine, SQLiteStorage, StepConfiguration, StepType


def normalize_input(inputs: dict[str, str]) -> str:
    return inputs["text"].strip().lower()


semantics = StepConfiguration(step_type=StepType.DETERMINISTIC)
with SQLiteStorage(db_path=".my_run/run.db") as storage:
    engine = MusubitoEngine(storage)
    decision = engine.explain(
        "normalize-input",
        {"text": " Example "},
        normalize_input,
        semantics,
    )
    print(decision.replayable, decision.reason)

Storage

Musubito uses SQLite as its local relational storage layer. By default, it stores runtime data under:

.musubito/musubito.db

The path can be customized with SQLiteStorage(db_path=...).

SQLite is opened in WAL mode and uses short BEGIN IMMEDIATE write transactions for node, artifact, edge, and invalidation updates. This keeps local concurrent writes predictable while still allowing normal reads.

Downstream invalidation is performed in place with a recursive CTE. When a node output changes, dependent downstream nodes can be marked stale so future runs recompute only the affected part of the DAG.

Concurrent requests for the same missing node use a persisted execution claim. Threads and coroutines can share one storage instance; separate processes coordinate through their own SQLiteStorage connections to the same database. They converge on one successful execution while the other callers wait for its artifact. Claims use renewable leases so an abandoned claim can be recovered, and ownership is checked atomically in the persistence transaction. External effects should still be idempotent because no local lease can prevent a remote side effect from being repeated after a process failure.

Long-running projects can configure independent lifecycle controls:

from musubito import SQLiteStorage


storage = SQLiteStorage(
    db_path=".my_run/run.db",
    retention_days=30,
    max_size_mb=512,
    auto_vacuum=True,
)
print(storage.storage_stats())
storage.close()

Retention removes expired history only when it is no longer required as an ancestor or historical producer of a retained node. Active execution claims are protected. The size policy evicts old leaf nodes in batches and measures live SQLite pages; allocated_size_mb can remain larger until incremental vacuum reclaims free pages. The cap is best-effort when the active protected lineage alone exceeds the target. Incremental auto-vacuum must be selected when a database is first created or converted from SQLite's FULL mode. A database created with auto_vacuum=NONE emits a warning rather than being subjected to a blocking full VACUUM.

Upgrading to 0.3

Existing databases are migrated in place and legacy artifacts remain readable. Node identity is now version 2, so artifacts recorded by earlier releases are not replayed under the stronger identity contract. See the changelog for the complete compatibility notes. Opening a database written by a newer, unsupported schema fails explicitly with StorageSchemaVersionError instead of attempting a downgrade.

Citation

If you use Musubito in any way, including research, experiments, prototypes, internal tools, or derivative implementations, please cite the accompanying preprint.

Altieri, Domenico, Musubito: Deterministic Execution Lineage for AI-Enabled Python Workflows. Available at SSRN: https://ssrn.com/abstract=6947764 or DOI: http://dx.doi.org/10.2139/ssrn.6947764

Citations help document the use of the project, support future maintenance, and make related work easier to trace.

License

Musubito is dual-licensed:

• Open Source: AGPL-3.0-or-later — free for open-source projects. See LICENSE.
• Commercial: closed-source or proprietary use requires a separate license. See COMMERCIAL_LICENSE.md or contact softwaretamrsv@gmail.com.

Open-source projects and personal use: AGPL-3.0. Closed-source or commercial products: commercial license required.